repositorySubscriberIfTestSuite.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_TEST_REPOSITORYSUBSCRIBERIFTESTSUITE_HPP
#define RTCTK_COMPONENTFRAMEWORK_TEST_REPOSITORYSUBSCRIBERIFTESTSUITE_HPP
 
#include <memory>
#include <rtctk/componentFramework/repositorySubscriberIf.hpp>
 
#include <gtest/gtest.h>
 
#include <chrono>
#include <stdexcept>
#include <thread>
 
namespace rtctk::componentFramework::test {
 
static std::shared_ptr<RepositorySubscriberIf> MakeRepository();
 
template <typename T>
class ThreadSafeQ {
public:
    void PushBack(const T& path) {
        std::lock_guard lock{m_mutex};
        m_data.push_back(path);
    }
 
    T operator[](size_t idx) {
        std::lock_guard lock{m_mutex};
        return m_data.at(idx);
    }
 
    size_t Size() {
        std::lock_guard lock{m_mutex};
        return m_data.size();
    }
 
    bool Contains(const T& val) {
        std::lock_guard lock{m_mutex};
        auto it = std::find(m_data.begin(), m_data.end(), val);
        return it != m_data.end();
    }
 
    void AwaitSize(size_t target_size, std::chrono::seconds timeout = std::chrono::seconds(5)) {
        auto t_start = std::chrono::steady_clock::now();
        while (Size() < target_size) {
            std::this_thread::sleep_for(std::chrono::microseconds(100));
            if ((std::chrono::steady_clock::now() - t_start) > timeout) {
                FAIL() << "ThreadSafeQ::AwaitSize: Timed out waiting for size "
                       << std::to_string(target_size);
            }
        }
    }
 
private:
    std::mutex m_mutex;
    std::vector<T> m_data;
};
 
std::chrono::milliseconds g_sleep_duration = std::chrono::milliseconds(0);
 
void Sleep() {
    if (g_sleep_duration > std::chrono::milliseconds(0)) {
        std::this_thread::sleep_for(g_sleep_duration);
    }
}
 
class Subscription : public testing::Test {
public:
    void SetUp() override {
        repo = MakeRepository();
        path1 = "/foo"_dppath;
        path2 = "/bar"_dppath;
        link = "/link"_dppath;
    }
 
    void TearDown() override {
        if (repo == nullptr) {
            // in case we could not connect in the first place
            return;
        }
        if (repo->DataPointExists(link)) {
            repo->DeleteDataPoint(link);
        }
        if (repo->DataPointExists(path1)) {
            repo->DeleteDataPoint(path1);
        }
        if (repo->DataPointExists(path2)) {
            repo->DeleteDataPoint(path2);
        }
    }
 
protected:
    std::shared_ptr<RepositorySubscriberIf> repo;
    DataPointPath path1;
    DataPointPath path2;
    DataPointPath link;
};
 
TEST_F(Subscription, ValueSubscription) {
    {
        int int_create = 0;
        RepositoryIf::BatchRequest req;
        req.CreateDataPoint(path1, int_create);
        req.CreateDataPoint(path2, int_create);
        repo->SendRequest(req).Wait();
    }
 
    ThreadSafeQ<DataPointPath> path_q_1;
    ThreadSafeQ<int> value_q_1;
    ThreadSafeQ<std::exception_ptr> error_q_1;
    ThreadSafeQ<DataPointPath> path_q_2;
    int num_cb_q_1 = 0;
    int num_cb_q_2 = 0;
 
    {
        auto sub1 = repo->Subscribe<int>(
            path1,
            [&](const DataPointPath& path,
                const int& data,
                const RepositoryIf::MetaData& metadata) {
                path_q_1.PushBack(path);
                value_q_1.PushBack(data);
            },
            // NOLINTNEXTLINE(performance-unnecessary-value-param)
            [&](std::exception_ptr error) { error_q_1.PushBack(error); });
 
        auto sub2 = repo->Subscribe(
            path2,
            [&](const DataPointPath& path, const RepositoryIf::MetaData& metadata) {
                path_q_2.PushBack(path);
            },
            nullptr);
 
        // Wait a moment for the subscription to complete before updating the data point.
        Sleep();
 
        {
            int int_write = 3;
            RepositoryIf::BatchRequest req;
            req.WriteDataPoint(path1, int_write);
            req.WriteDataPoint(path2, int_write);
            repo->SendRequest(req).Wait();
 
            path_q_1.AwaitSize(1);
            value_q_1.AwaitSize(1);
            path_q_2.AwaitSize(1);
            EXPECT_TRUE(path_q_1.Contains(path1));
            EXPECT_TRUE(value_q_1.Contains(int_write));
            EXPECT_TRUE(path_q_2.Contains(path2));
            EXPECT_EQ(error_q_1.Size(), 0);
        }
    }  // here the subscription goes out of scope and we unsubscribe
 
    num_cb_q_1 = path_q_1.Size();
    num_cb_q_2 = path_q_2.Size();
 
    {
        int int_write = 4;
        RepositoryIf::BatchRequest req;
        req.WriteDataPoint(path1, int_write);
        req.WriteDataPoint(path2, int_write);
        repo->SendRequest(req).Wait();
 
        // best effort: sleep a short time and check that really no more callbacks were triggered
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
 
        // we ensure that no new callbacks have arrived
        EXPECT_EQ(error_q_1.Size(), 0);
        EXPECT_EQ(path_q_1.Size(), num_cb_q_1);
        EXPECT_EQ(value_q_1.Size(), num_cb_q_1);
        EXPECT_EQ(path_q_2.Size(), num_cb_q_2);
    }
}
 
TEST_F(Subscription, MultipleSubscriptionsToSameDataPoint) {
    repo->CreateDataPoint(path1, 0);
 
    ThreadSafeQ<DataPointPath> path_q_1;
    ThreadSafeQ<DataPointPath> path_q_2;
    int num_cb_q_1 = 0;
    int num_cb_q_2 = 0;
 
    auto sub1 = repo->Subscribe(
        path1,
        [&](const DataPointPath& path, const RepositoryIf::MetaData& metadata) {
            path_q_1.PushBack(path);
        },
        nullptr);
 
    {
        auto sub2 = repo->Subscribe(
            path1,
            [&](const DataPointPath& path, const RepositoryIf::MetaData& metadata) {
                path_q_2.PushBack(path);
            },
            nullptr);
 
        // Wait a moment for the subscription to complete before updating the data point.
        Sleep();
 
        {
            repo->WriteDataPoint(path1, 3);
 
            path_q_1.AwaitSize(1);
            path_q_2.AwaitSize(1);
            EXPECT_TRUE(path_q_1.Contains(path1));
            EXPECT_TRUE(path_q_2.Contains(path1));
        }
    }  // here sub 2 goes out of scope and we unsubscribe
 
    num_cb_q_1 = path_q_1.Size();
    num_cb_q_2 = path_q_2.Size();
 
    {
        repo->WriteDataPoint(path1, 4);
 
        path_q_1.AwaitSize(num_cb_q_1 + 1);
 
        // best effort: sleep a short time and check that really no more callbacks were triggered
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
 
        EXPECT_EQ(path_q_1.Size(), (num_cb_q_1 + 1));
        EXPECT_EQ(path_q_2.Size(), num_cb_q_2);
    }
}
 
TEST_F(Subscription, NotifySubscriptionCausesNoDataRace) {
    {
        int int_create = 0;
        RepositoryIf::BatchRequest req;
        req.CreateDataPoint(path1, int_create);
        repo->SendRequest(req).Wait();
    }
 
    std::atomic_bool stop = false;
 
    // this thread keeps writing the datapoint with high frequency
    auto writer_func = [&] {
        int value = 3;
        while (stop == false) {
            repo->WriteDataPoint(path1, value);
            value++;
        }
    };
 
    // this thread subscribes, waits for notification and unsubscribes with high frequency
    auto subscriber_func = [&] {
        for (unsigned i = 0; i < 1000; i++) {
            ThreadSafeQ<DataPointPath> path_q_1;
 
            auto sub = repo->Subscribe(
                path1,
                [&](const DataPointPath& path, const RepositoryIf::MetaData& metadata) {
                    path_q_1.PushBack(path);
                },
                nullptr);
 
            path_q_1.AwaitSize(1);
        }  // here the subscription goes out of scope and we unsubscribe
 
        stop = true;
    };
 
    // let the threads to their work
    std::thread t1{writer_func};
    std::thread t2{subscriber_func};
    t1.join();
    t2.join();
}
 
TEST_F(Subscription, ValueSubscriptionToSymlink) {
    {
        int int_create = 0;
        RepositoryIf::BatchRequest req;
        req.CreateDataPoint(path1, int_create);
        req.CreateSymlink(path1, link);
        repo->SendRequest(req).Wait();
    }
 
    ThreadSafeQ<DataPointPath> path_q;
    ThreadSafeQ<int> value_q;
    ThreadSafeQ<std::exception_ptr> error_q;
    int num_cb_q = 0;
 
    {
        auto sub = repo->Subscribe<int>(
            link,
            [&](const DataPointPath& path,
                const int& data,
                const RepositoryIf::MetaData& metadata) {
                path_q.PushBack(path);
                value_q.PushBack(data);
            },
            // NOLINTNEXTLINE(performance-unnecessary-value-param)
            [&](std::exception_ptr error) { error_q.PushBack(error); });
 
        // Wait a moment for the subscription to complete before updating the data point.
        Sleep();
 
        {
            int int_write = 3;
            repo->WriteDataPoint(path1, int_write);
 
            path_q.AwaitSize(1);
            value_q.AwaitSize(1);
            EXPECT_TRUE(path_q.Contains(link));
            EXPECT_TRUE(value_q.Contains(int_write));
            EXPECT_EQ(error_q.Size(), 0);
        }
    }  // here the subscription goes out of scope and we unsubscribe
 
    num_cb_q = path_q.Size();
 
    {
        int int_write = 4;
        repo->WriteDataPoint(path1, int_write);
 
        // best effort: sleep a short time and check that really no more callbacks were triggered
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
 
        EXPECT_EQ(error_q.Size(), 0);
        EXPECT_EQ(path_q.Size(), num_cb_q);
        EXPECT_EQ(value_q.Size(), num_cb_q);
    }
}
 
TEST_F(Subscription, CreateDeleteSubscription) {
    ThreadSafeQ<DataPointPath> create_q;
    ThreadSafeQ<DataPointPath> delete_q;
    ThreadSafeQ<std::exception_ptr> error_q;
 
    {
        auto subscription =
            repo->Subscribe([&](const DataPointPath& path) { create_q.PushBack(path); },
                            [&](const DataPointPath& path) { delete_q.PushBack(path); },
                            // NOLINTNEXTLINE(performance-unnecessary-value-param)
                            [&](std::exception_ptr error) { error_q.PushBack(error); });
 
        {
            int int_create = 0;
            RepositoryIf::BatchRequest req;
            req.CreateDataPoint(path1, int_create);
            req.CreateDataPoint(path2, int_create);
            repo->SendRequest(req).Wait();
 
            create_q.AwaitSize(2);
            EXPECT_TRUE(create_q.Contains(path1));
            EXPECT_TRUE(create_q.Contains(path2));
            EXPECT_EQ(delete_q.Size(), 0);
            EXPECT_EQ(error_q.Size(), 0);
        }
 
        {
            RepositoryIf::BatchRequest req;
            req.DeleteDataPoint(path2);
            req.DeleteDataPoint(path1);
            repo->SendRequest(req).Wait();
 
            delete_q.AwaitSize(2);
            EXPECT_EQ(create_q.Size(), 2);
            EXPECT_TRUE(delete_q.Contains(path1));
            EXPECT_TRUE(delete_q.Contains(path2));
            EXPECT_EQ(error_q.Size(), 0);
        }
    }  // here the subscription goes out of scope and we unsubscribe
 
    {
        int int_create = 0;
        RepositoryIf::BatchRequest req;
        req.CreateDataPoint(path1, int_create);
        req.CreateDataPoint(path2, int_create);
        repo->SendRequest(req).Wait();
 
        // best effort: sleep a short time and check that really no more callbacks were triggered
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
 
        EXPECT_EQ(create_q.Size(), 2);
        EXPECT_EQ(delete_q.Size(), 2);
        EXPECT_EQ(error_q.Size(), 0);
    }
}
 
TEST_F(Subscription, CreateDeleteSubscriptionToSymlink) {
    ThreadSafeQ<DataPointPath> create_q;
    ThreadSafeQ<DataPointPath> delete_q;
    ThreadSafeQ<std::exception_ptr> error_q;
 
    {
        auto subscription =
            repo->Subscribe([&](const DataPointPath& path) { create_q.PushBack(path); },
                            [&](const DataPointPath& path) { delete_q.PushBack(path); },
                            // NOLINTNEXTLINE(performance-unnecessary-value-param)
                            [&](std::exception_ptr error) { error_q.PushBack(error); });
 
        {
            int int_create = 0;
            RepositoryIf::BatchRequest req;
            req.CreateDataPoint(path1, int_create);
            req.CreateSymlink(path1, path2);
            repo->SendRequest(req).Wait();
 
            create_q.AwaitSize(2);
            EXPECT_TRUE(create_q.Contains(path1));
            EXPECT_TRUE(create_q.Contains(path2));
            EXPECT_EQ(delete_q.Size(), 0);
            EXPECT_EQ(error_q.Size(), 0);
        }
 
        {
            RepositoryIf::BatchRequest req;
            req.DeleteDataPoint(path2);
            req.DeleteDataPoint(path1);
            repo->SendRequest(req).Wait();
 
            delete_q.AwaitSize(2);
            EXPECT_EQ(create_q.Size(), 2);
            EXPECT_TRUE(delete_q.Contains(path1));
            EXPECT_TRUE(delete_q.Contains(path2));
            EXPECT_EQ(error_q.Size(), 0);
        }
    }  // here the subscription goes out of scope and we unsubscribe
 
    {
        int int_create = 0;
        RepositoryIf::BatchRequest req;
        req.CreateDataPoint(path1, int_create);
        req.CreateSymlink(path1, path2);
        repo->SendRequest(req).Wait();
 
        // best effort: sleep a short time and check that really no more callbacks were triggered
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
 
        EXPECT_EQ(create_q.Size(), 2);
        EXPECT_EQ(delete_q.Size(), 2);
        EXPECT_EQ(error_q.Size(), 0);
    }
}
 
TEST_F(Subscription, SubscribeShouldThrowForNonExistingDataPoint) {
    EXPECT_THROW(
        {
            auto sub = repo->Subscribe<int>(
                path1,
                [&](const DataPointPath& path,
                    const int& data,
                    const RepositoryIf::MetaData& metadata) {},
                // NOLINTNEXTLINE(performance-unnecessary-value-param)
                [&](std::exception_ptr error) {});
        },
        RepositoryIf::DataPointDoesNotExist);
}
 
}  // namespace rtctk::componentFramework::test
 
#endif  // RTCTK_COMPONENTFRAMEWORK_TEST_REPOSITORYSUBSCRIBERIFTESTSUITE_HPP