Merge pull request #58 from Wolkabout/feature/details-gateway-synchro…

…nization

Added more 22.GA functionality, such as `details_synchronization` and `children_synchronization`, and various fixes, now verified in the subdevices flow.

.github/workflows/cmake-build-and-test.yml

@@ -15,7 +15,7 @@ jobs:
     # cross-platform coverage.
     # See: https://docs.github.com/en/free-pro-team@latest/actions/learn-github-actions/managing-complex-workflows#using-a-build-matrix
     runs-on: ubuntu-20.04
-    timeout-minutes: 15
+    timeout-minutes: 20
 
     steps:
       - uses: actions/checkout@v1
@@ -49,7 +49,7 @@ jobs:
       - name: Run Tests
         working-directory: ${{runner.workspace}}/out
         shell: bash
-        run: ctest .
+        run: ctest --rerun-failed --output-on-failure .
 
       - name: Check Format
         uses: jidicula/clang-format-action@v3.5.2

configure.sh

@@ -23,5 +23,5 @@ chmod +x .git/hooks/pre-commit
 
 # Enter the directory and start the CMake build
 pushd out || exit
-cmake -G "Unix Makefiles" -DCMAKE_BUILD_TYPE=Release ..
+cmake -G "Unix Makefiles" -DCMAKE_BUILD_TYPE=Release -DBUILD_GTEST=OFF -DBUILD_TESTS=OFF -DBUILD_EXAMPLES=OFF ..
 popd || exit

debian/rules

@@ -2,4 +2,8 @@
 DH_VERBOSE = 1
 
 %:
-        dh $@
+	dh $@ --builddirectory=out
+override_dh_auto_clean:
+override_dh_auto_test:
+override_dh_shlibdeps:
+	dh_shlibdeps -l$(shell pwd)/out/lib

examples/full_feature/Application.cpp

@@ -34,9 +34,9 @@
  * In here, you can enter the device credentials to successfully identify the device on the platform.
  * And also, the target platform path, and the SSL certificate that is used to establish a secure connection.
  */
-const std::string DEVICE_KEY = "AWC";
-const std::string DEVICE_PASSWORD = "VZ8R3MI87R";
-const std::string PLATFORM_HOST = "ssl://integration5.wolkabout.com:8883";
+const std::string DEVICE_KEY = "<DEVICE_KEY>";
+const std::string DEVICE_PASSWORD = "<DEVICE_PASSWORD>";
+const std::string PLATFORM_HOST = "ssl://demo.wolkabout.com:8883";
 const std::string CA_CERT_PATH = "./ca.crt";
 const std::string FILE_MANAGEMENT_LOCATION = "./files";
 const std::string FIRMWARE_VERSION = "4.0.0";
@@ -109,6 +109,21 @@ class DeviceDataChangeHandler : public wolkabout::connect::FeedUpdateHandler
     DeviceData& m_deviceData;
 };
 
+/**
+ * This is an example implementation of the `ParameterHandler` interface. This class will receive information about
+ * parameter value updates.
+ */
+class ExampleParameterHandler : public wolkabout::connect::ParameterHandler
+{
+public:
+    void handleUpdate(const std::string& deviceKey, const std::vector<wolkabout::Parameter>& parameters) override
+    {
+        LOG(INFO) << "ExampleParameterHandler received parameter values for device '" << deviceKey << "':";
+        for (const auto& parameter : parameters)
+            LOG(INFO) << "\t" << toString(parameter.first) << " -> '" << parameter.second << "'";
+    }
+};
+
 /**
  * This is an example implementation of the `FirmwareInstaller` interface. This class will ask the user about
  * preferences for return values of the methods.
@@ -233,6 +248,7 @@ int main(int /* argc */, char** /* argv */)
     auto device = wolkabout::Device{DEVICE_KEY, DEVICE_PASSWORD, wolkabout::OutboundDataMode::PUSH};
     auto deviceInfo = DeviceData{0, false, std::chrono::seconds(5)};
     auto deviceInfoHandler = std::make_shared<DeviceDataChangeHandler>(deviceInfo);
+    auto parameterHandler = std::make_shared<ExampleParameterHandler>();
 
     /**
      * Now we can start creating the Wolk instance that is right for us.
@@ -244,6 +260,7 @@ int main(int /* argc */, char** /* argv */)
                   .host(PLATFORM_HOST)
                   .caCertPath(CA_CERT_PATH)
                   .feedUpdateHandler(deviceInfoHandler)
+                  .parameterHandler(parameterHandler)
                   .withPersistence(std::move(inMemoryPersistence))
                   .withFileTransfer(FILE_MANAGEMENT_LOCATION)
                   // Uncomment for FileURLDownload
@@ -269,10 +286,21 @@ int main(int /* argc */, char** /* argv */)
     };
     signal(SIGINT, sigintResponse);
 
+    std::this_thread::sleep_for(std::chrono::seconds(5));
+    wolk->obtainDetails([&](const std::vector<std::string>& feeds, const std::vector<std::string>& attributes) {
+        LOG(INFO) << "Received device details: ";
+        LOG(INFO) << "\tFeeds: ";
+        for (const auto& feed : feeds)
+            LOG(INFO) << "\t\t" << feed;
+        LOG(INFO) << "\tAttributes: ";
+        for (const auto& attribute : attributes)
+            LOG(INFO) << "\t\t" << attribute;
+    });
+
     /**
      * We want to randomize the temperature data too, so we need the generator for random information.
      */
-    auto engine = std::mt19937{static_cast<std::uint64_t>(std::chrono::system_clock::now().time_since_epoch().count())};
+    auto engine = std::mt19937{static_cast<std::uint32_t>(std::chrono::system_clock::now().time_since_epoch().count())};
     auto distribution = std::uniform_real_distribution<std::double_t>{-20.0, 80.0};
 
     while (running)

examples/register_feed_and_attribute/Application.cpp

@@ -39,7 +39,7 @@ std::uint64_t generateRandomValue()
 {
     // Here we will create the random engine and distribution
     static auto engine =
-      std::mt19937(static_cast<std::uint64_t>(std::chrono::system_clock::now().time_since_epoch().count()));
+      std::mt19937(static_cast<std::uint32_t>(std::chrono::system_clock::now().time_since_epoch().count()));
     static auto distribution = std::uniform_real_distribution<>(0, 100);
 
     // And generate a random value

examples/simple/Application.cpp

@@ -38,7 +38,7 @@ std::uint64_t generateRandomValue()
 {
     // Here we will create the random engine and distribution
     static auto engine =
-      std::mt19937(static_cast<std::uint64_t>(std::chrono::system_clock::now().time_since_epoch().count()));
+      std::mt19937(static_cast<std::uint32_t>(std::chrono::system_clock::now().time_since_epoch().count()));
     static auto distribution = std::uniform_real_distribution<>(-20, 80);
 
     // And generate a random value

examples/subdevices/Application.cpp

@@ -18,6 +18,8 @@
 #include "wolk/WolkBuilder.h"
 #include "wolk/WolkMulti.h"
 
+using namespace wolkabout;
+
 class ExamplePlatformStatusListener : public wolkabout::connect::PlatformStatusListener
 {
 public:
@@ -39,14 +41,14 @@ int main(int /* argc */, char** /* argv */)
     auto deviceThree = wolkabout::Device{"ThirdDevice", "", wolkabout::OutboundDataMode::PUSH};
 
     // And now we can create the wolk session
-    auto wolk = wolkabout::connect::WolkMulti::newBuilder({deviceOne, deviceTwo})
+    auto wolk = wolkabout::connect::WolkMulti::newBuilder({})
                   .host("tcp://localhost:1883")
                   .withFileTransfer("./files")
                   .withPlatformStatus(std::unique_ptr<ExamplePlatformStatusListener>(new ExamplePlatformStatusListener))
                   .withErrorProtocol(std::chrono::minutes{10})
+                  .withRegistration()
                   .buildWolkMulti();
     wolk->connect();
-
     wolk->addReading(deviceOne.getKey(), "π", 3.14);
     wolk->publish();
     wolk->pullFeedValues(deviceTwo.getKey());
@@ -67,13 +69,6 @@ int main(int /* argc */, char** /* argv */)
 
         // We can sleep again
         std::this_thread::sleep_for(std::chrono::seconds(5));
-
-        // Do some things for the devices
-        for (const auto& device : devices)
-        {
-            LOG(INFO) << "Count of errors for device: '" << device.getKey() << "' -> "
-                      << wolk->peekErrorCount(device.getKey()) << ".";
-        }
     }
 
     // And that's it

out/coverage.sh

@@ -5,5 +5,5 @@ echo "${WORK_DIR}"
 rm ./coverage.info
 rm -rf ./coverage
 lcov -b . -c -d . -o ./coverage.info
-lcov -r ./coverage.info '/usr/*' "${WORK_DIR}/WolkSDK-Cpp/*" "${WORK_DIR}/out/*" "${WORK_DIR}/tests/*" -o ./coverage.info
+lcov -r ./coverage.info '/usr/*' "${WORK_DIR}/WolkSDK-Cpp/*" "${WORK_DIR}/out/*" "${WORK_DIR}/tests/*" "${WORK_DIR}/**.h" -o ./coverage.info
 genhtml -o ./coverage ./coverage.info

tests/DataServiceTests.cpp

@@ -29,6 +29,8 @@
 #include "core/utilities/Logger.h"
 #include "tests/mocks/ConnectivityServiceMock.h"
 #include "tests/mocks/DataProtocolMock.h"
+#include "tests/mocks/OutboundMessageHandlerMock.h"
+#include "tests/mocks/OutboundRetryMessageHandlerMock.h"
 #include "tests/mocks/PersistenceMock.h"
 
 #include <gtest/gtest.h>
@@ -43,6 +45,9 @@ class DataServiceTests : public ::testing::Test
     {
         // Set up the mocks
         connectivityServiceMock = std::make_shared<ConnectivityServiceMock>();
+        outboundMessageHandlerMock = std::make_shared<OutboundMessageHandlerMock>();
+        outboundRetryMessageHandlerMock =
+          std::make_shared<OutboundRetryMessageHandlerMock>(*outboundMessageHandlerMock);
         dataProtocolMock = std::make_shared<DataProtocolMock>();
         persistenceMock = std::make_shared<PersistenceMock>();
 
@@ -56,10 +61,16 @@ class DataServiceTests : public ::testing::Test
             if (parameterSyncHandler)
                 parameterSyncHandler(deviceKey, parameters);
         };
+        _internalDetailsSyncHandler = [&](const std::string& deviceKey, const std::vector<std::string>& feeds,
+                                          const std::vector<std::string>& parameters) {
+            if (detailsSyncHandler)
+                detailsSyncHandler(deviceKey, feeds, parameters);
+        };
 
         // Create the service
         service = std::make_shared<DataService>(*dataProtocolMock, *persistenceMock, *connectivityServiceMock,
-                                                _internalFeedUpdateSetHandler, _internalParameterSyncHandler);
+                                                *outboundRetryMessageHandlerMock, _internalFeedUpdateSetHandler,
+                                                _internalParameterSyncHandler, _internalDetailsSyncHandler);
     }
 
     static void SetUpTestCase() { Logger::init(LogLevel::TRACE, Logger::Type::CONSOLE); }
@@ -72,8 +83,14 @@ class DataServiceTests : public ::testing::Test
 
     ParameterSyncHandler parameterSyncHandler;
 
+    DetailsSyncHandler detailsSyncHandler;
+
     std::shared_ptr<ConnectivityServiceMock> connectivityServiceMock;
 
+    std::shared_ptr<OutboundMessageHandlerMock> outboundMessageHandlerMock;
+
+    std::shared_ptr<OutboundRetryMessageHandlerMock> outboundRetryMessageHandlerMock;
+
     std::shared_ptr<DataProtocolMock> dataProtocolMock;
 
     std::shared_ptr<PersistenceMock> persistenceMock;
@@ -82,6 +99,7 @@ class DataServiceTests : public ::testing::Test
 
     FeedUpdateSetHandler _internalFeedUpdateSetHandler;
     ParameterSyncHandler _internalParameterSyncHandler;
+    DetailsSyncHandler _internalDetailsSyncHandler;
 };
 
 TEST_F(DataServiceTests, MakePersistenceKey)
@@ -143,8 +161,7 @@ TEST_F(DataServiceTests, PublishReadingsHappyFlow)
 
 TEST_F(DataServiceTests, CheckIfSubscriptionExistButItsEmpty)
 {
-    ASSERT_FALSE(
-      service->checkIfSubscriptionIsWaiting(std::make_shared<ParametersUpdateMessage>(std::vector<Parameter>{})));
+    ASSERT_FALSE(service->checkIfSubscriptionIsWaiting(ParametersUpdateMessage{{}}));
 }
 
 TEST_F(DataServiceTests, CheckIfSubscriptionExistTwoSubscription)
@@ -167,8 +184,8 @@ TEST_F(DataServiceTests, CheckIfSubscriptionExistTwoSubscription)
                                             }});
 
     // Now parse the subscription
-    ASSERT_TRUE(service->checkIfSubscriptionIsWaiting(
-      std::make_shared<ParametersUpdateMessage>(std::vector<Parameter>{{ParameterName::EXTERNAL_ID, "TestValue"}})));
+    ASSERT_TRUE(
+      service->checkIfSubscriptionIsWaiting(ParametersUpdateMessage{{{ParameterName::EXTERNAL_ID, "TestValue"}}}));
     if (!callbackCalled)
     {
         std::unique_lock<std::mutex> lock{mutex};
@@ -177,6 +194,31 @@ TEST_F(DataServiceTests, CheckIfSubscriptionExistTwoSubscription)
     EXPECT_TRUE(callbackCalled);
 }
 
+TEST_F(DataServiceTests, CheckIfCallbackNoCallbacks)
+{
+    ASSERT_TRUE(service->m_detailsCallbacks.empty());
+    ASSERT_FALSE(service->checkIfCallbackIsWaiting({{}, {}}));
+}
+
+TEST_F(DataServiceTests, CheckIfCallbackFinallyACallback)
+{
+    std::atomic_bool called;
+    std::mutex mutex;
+    std::condition_variable conditionVariable;
+    ASSERT_NO_FATAL_FAILURE(
+      service->m_detailsCallbacks.push([&](const std::vector<std::string>&, const std::vector<std::string>&) {
+          called = true;
+          conditionVariable.notify_one();
+      }));
+    ASSERT_TRUE(service->checkIfCallbackIsWaiting({{}, {}}));
+    if (!called)
+    {
+        std::unique_lock<std::mutex> lock{mutex};
+        conditionVariable.wait_for(lock, std::chrono::milliseconds{100});
+    }
+    EXPECT_TRUE(called);
+}
+
 TEST_F(DataServiceTests, AddReadingSingleStringReading)
 {
     EXPECT_CALL(*persistenceMock, putReading).Times(1);
@@ -355,11 +397,32 @@ TEST_F(DataServiceTests, SynchronizeParametersTestFailsToPublish)
 TEST_F(DataServiceTests, SynchronizeParametersTestFailsToParse)
 {
     EXPECT_CALL(*dataProtocolMock, makeOutboundMessage(_, A<SynchronizeParametersMessage>()))
-      .WillOnce([&](const std::string&, const SynchronizeParametersMessage&) { return nullptr; });
+      .WillOnce(Return(ByMove(nullptr)));
     EXPECT_CALL(*connectivityServiceMock, publish).Times(0);
     ASSERT_NO_FATAL_FAILURE(service->synchronizeParameters(DEVICE_KEY, {}, [](const std::vector<Parameter>&) {}));
 }
 
+TEST_F(DataServiceTests, DetailsSynchronzationFailsToParse)
+{
+    EXPECT_CALL(*dataProtocolMock, makeOutboundMessage(_, A<DetailsSynchronizationRequestMessage>()))
+      .WillOnce(Return(ByMove(nullptr)));
+    EXPECT_CALL(*outboundRetryMessageHandlerMock, addMessage).Times(0);
+    ASSERT_NO_FATAL_FAILURE(service->detailsSynchronizationAsync(
+      DEVICE_KEY, [](const std::vector<std::string>&, const std::vector<std::string>&) {}));
+}
+
+TEST_F(DataServiceTests, DetailsSynchronizationCall)
+{
+    EXPECT_CALL(*dataProtocolMock, makeOutboundMessage(_, A<DetailsSynchronizationRequestMessage>()))
+      .WillOnce(Return(ByMove(std::unique_ptr<wolkabout::Message>{new wolkabout::Message{"", ""}})));
+    EXPECT_CALL(*dataProtocolMock, getResponseChannelForMessage(MessageType::DETAILS_SYNCHRONIZATION_REQUEST, _))
+      .Times(1);
+    EXPECT_CALL(*outboundRetryMessageHandlerMock, addMessage)
+      .WillOnce([&](const RetryMessageStruct& retryMessageStruct) { retryMessageStruct.onFail({}); });
+    ASSERT_NO_FATAL_FAILURE(service->detailsSynchronizationAsync(
+      DEVICE_KEY, [](const std::vector<std::string>&, const std::vector<std::string>&) {}));
+}
+
 TEST_F(DataServiceTests, PublishReadings)
 {
     EXPECT_CALL(*persistenceMock, getReadingsKeys).WillOnce(Return(std::vector<std::string>{DEVICE_KEY}));
@@ -643,3 +706,61 @@ TEST_F(DataServiceTests, MessageReceivedMessageParameterHappyFlowAnswersSubscrip
     }
     EXPECT_TRUE(callbackCalled);
 }
+
+TEST_F(DataServiceTests, MessageReceivedMessageDetailSynchronizationResponseFailsToParse)
+{
+    EXPECT_CALL(*dataProtocolMock, getDeviceKey).WillOnce(Return(DEVICE_KEY));
+    EXPECT_CALL(*dataProtocolMock, getMessageType).WillOnce(Return(MessageType::DETAILS_SYNCHRONIZATION_RESPONSE));
+    EXPECT_CALL(*dataProtocolMock, parseDetails).WillOnce(Return(ByMove(nullptr)));
+    ASSERT_NO_FATAL_FAILURE(service->messageReceived(std::make_shared<wolkabout::Message>("", "")));
+}
+
+TEST_F(DataServiceTests, MessageReceivedMessageCallbackIsAwaiting)
+{
+    std::atomic_bool callbackCalled{false};
+    std::mutex mutex;
+    std::condition_variable conditionVariable;
+    EXPECT_CALL(*dataProtocolMock, getDeviceKey).WillOnce(Return(DEVICE_KEY));
+    EXPECT_CALL(*dataProtocolMock, getMessageType).WillOnce(Return(MessageType::DETAILS_SYNCHRONIZATION_RESPONSE));
+    EXPECT_CALL(*dataProtocolMock, parseDetails)
+      .WillOnce(Return(ByMove(
+        std::unique_ptr<DetailsSynchronizationResponseMessage>{new DetailsSynchronizationResponseMessage{{}, {}}})));
+
+    service->m_detailsCallbacks.push([&](const std::vector<std::string>&, const std::vector<std::string>&) {
+        callbackCalled = true;
+        conditionVariable.notify_one();
+    });
+
+    ASSERT_NO_FATAL_FAILURE(service->messageReceived(std::make_shared<wolkabout::Message>("", "")));
+    if (!callbackCalled)
+    {
+        std::unique_lock<std::mutex> lock{mutex};
+        conditionVariable.wait_for(lock, std::chrono::milliseconds{100});
+    }
+    EXPECT_TRUE(callbackCalled);
+}
+
+TEST_F(DataServiceTests, MessageReceivedMessageFallbackToHandler)
+{
+    std::atomic_bool callbackCalled{false};
+    std::mutex mutex;
+    std::condition_variable conditionVariable;
+    EXPECT_CALL(*dataProtocolMock, getDeviceKey).WillOnce(Return(DEVICE_KEY));
+    EXPECT_CALL(*dataProtocolMock, getMessageType).WillOnce(Return(MessageType::DETAILS_SYNCHRONIZATION_RESPONSE));
+    EXPECT_CALL(*dataProtocolMock, parseDetails)
+      .WillOnce(Return(ByMove(
+        std::unique_ptr<DetailsSynchronizationResponseMessage>{new DetailsSynchronizationResponseMessage{{}, {}}})));
+
+    detailsSyncHandler = [&](const std::string&, const std::vector<std::string>&, const std::vector<std::string>&) {
+        callbackCalled = true;
+        conditionVariable.notify_one();
+    };
+
+    ASSERT_NO_FATAL_FAILURE(service->messageReceived(std::make_shared<wolkabout::Message>("", "")));
+    if (!callbackCalled)
+    {
+        std::unique_lock<std::mutex> lock{mutex};
+        conditionVariable.wait_for(lock, std::chrono::milliseconds{100});
+    }
+    EXPECT_TRUE(callbackCalled);
+}

tests/ErrorServiceTests.cpp

@@ -111,6 +111,11 @@ TEST_F(ErrorServiceTests, OneMessageObtainFromCache)
     EXPECT_NE(message->getArrivalTime().time_since_epoch().count(), 0);
 }
 
+TEST_F(ErrorServiceTests, ObtainAndAwaitButNothingComes)
+{
+    ASSERT_EQ(service->obtainOrAwaitMessageForDevice(DEVICE_KEY, std::chrono::milliseconds{100}), nullptr);
+}
+
 TEST_F(ErrorServiceTests, OneMessageObtainFromCacheMapExistsButNoMessage)
 {
     // Add the message and start the timer