Merge pull request #455 from mtconnect/merge_main_to_main_dev

Merge main to main dev

CMakeLists.txt

@@ -3,7 +3,7 @@ set(AGENT_VERSION_MAJOR 2)
 set(AGENT_VERSION_MINOR 5)
 set(AGENT_VERSION_PATCH 0)
 set(AGENT_VERSION_BUILD 0)
-set(AGENT_VERSION_RC "RC2")
+set(AGENT_VERSION_RC "RC4")
 
 # This minimum version is to support Visual Studio 2019 and C++ feature checking and FetchContent
 cmake_minimum_required(VERSION 3.23 FATAL_ERROR)

conanfile.py

@@ -9,7 +9,7 @@
 
 class MTConnectAgentConan(ConanFile):
     name = "mtconnect_agent"
-    version = "2.2"
+    version = "2.3"
     url = "https://github.com/mtconnect/cppagent.git"
     license = "Apache License 2.0"
     settings = "os", "compiler", "arch", "build_type"

src/mtconnect/mqtt/mqtt_client_impl.hpp

@@ -164,6 +164,7 @@ namespace mtconnect {
           m_connected = false;
           if (m_handler && m_handler->m_disconnected)
             m_handler->m_disconnected(shared_from_this());
+
           if (m_running)
           {
             reconnect();

src/mtconnect/sink/mqtt_sink/mqtt_service.cpp

@@ -187,7 +187,7 @@ namespace mtconnect {
           }
         }
 
-        auto seq = m_sinkContract->getCircularBuffer().getSequence();
+        auto seq = publishCurrent(boost::system::error_code {});
         for (auto &dev : m_sinkContract->getDevices())
         {
           FilterSet filterSet = filterForDevice(dev);
@@ -199,10 +199,8 @@ namespace mtconnect {
           sampler->observe(seq, [this](const std::string &id) {
             return m_sinkContract->getDataItemById(id).get();
           });
-          sampler->handlerCompleted();
+          publishSample(sampler);
         }
-
-        publishCurrent(boost::system::error_code {});
       }
 
       /// @brief publish sample when observations arrive.
@@ -246,18 +244,20 @@ namespace mtconnect {
         return end;
       }
 
-      void MqttService::publishCurrent(boost::system::error_code ec)
+      SequenceNumber_t MqttService::publishCurrent(boost::system::error_code ec)
       {
+        SequenceNumber_t firstSeq, seq = 0;
+
         if (ec)
         {
           LOG(warning) << "Mqtt2Service::publishCurrent: " << ec.message();
-          return;
+          return 0;
         }
 
         if (!m_client->isRunning() || !m_client->isConnected())
         {
           LOG(warning) << "Mqtt2Service::publishCurrent: client stopped";
-          return;
+          return 0;
         }
 
         for (auto &device : m_sinkContract->getDevices())
@@ -266,7 +266,6 @@ namespace mtconnect {
           LOG(debug) << "Publishing current for: " << topic;
 
           ObservationList observations;
-          SequenceNumber_t firstSeq, seq;
           auto filterSet = filterForDevice(device);
 
           {
@@ -290,6 +289,8 @@ namespace mtconnect {
         m_currentTimer.expires_after(m_currentInterval);
         m_currentTimer.async_wait(boost::asio::bind_executor(
             m_strand, boost::bind(&MqttService::publishCurrent, this, _1)));
+
+          return seq;
       }
 
       bool MqttService::publish(observation::ObservationPtr &observation)
@@ -327,8 +328,9 @@ namespace mtconnect {
 
         LOG(debug) << "Publishing Asset to topic: " << topic;
 
-        auto doc = m_jsonPrinter->print(asset);
-
+        asset::AssetList list {asset};
+        auto doc = m_printer->printAssets(
+            m_instanceId, uint32_t(m_sinkContract->getAssetStorage()->getMaxAssets()), 1, list);
         stringstream buffer;
         buffer << doc;
 

src/mtconnect/sink/mqtt_sink/mqtt_service.hpp

@@ -96,7 +96,7 @@ namespace mtconnect {
         void pubishInitialContent();
 
         /// @brief Publish a current using `CurrentInterval` option.
-        void publishCurrent(boost::system::error_code ec);
+        SequenceNumber_t publishCurrent(boost::system::error_code ec);
 
         /// @brief publish sample when observations arrive.
         SequenceNumber_t publishSample(std::shared_ptr<observation::AsyncObserver> sampler);

src/mtconnect/sink/rest_sink/rest_service.cpp

@@ -544,16 +544,18 @@ namespace mtconnect {
 
       auto idHandler = [&](SessionPtr session, RequestPtr request) -> bool {
         auto asset = request->parameter<string>("assetIds");
+        auto pretty = *request->parameter<bool>("pretty");
         if (asset)
         {
           auto pretty = request->parameter<bool>("pretty").value_or(false);
           auto printer = m_sinkContract->getPrinter(acceptFormat(request->m_accepts));
-
+          
           list<string> ids;
           stringstream str(*asset);
           string id;
           while (getline(str, id, ';'))
             ids.emplace_back(id);
+
           respond(session, assetIdsRequest(printer, ids, pretty, request->m_requestId),
                   request->m_requestId);
         }

test_package/mqtt_sink_test.cpp

@@ -62,7 +62,6 @@ class MqttSinkTest : public testing::Test
   void TearDown() override
   {
     const auto agent = m_agentTestHelper->getAgent();
-    m_agentTestHelper->m_ioContext.run_for(500ms);
     if (agent)
     {
       m_agentTestHelper->getAgent()->stop();
@@ -249,15 +248,23 @@ TEST_F(MqttSinkTest, mqtt_sink_should_publish_Sample)
 
   auto handler = make_unique<ClientHandler>();
   bool gotSample = false;
-  handler->m_receive = [&gotSample](std::shared_ptr<MqttClient> client, const std::string &topic,
+  bool first = true;
+  handler->m_receive = [&gotSample, &first](std::shared_ptr<MqttClient> client, const std::string &topic,
                                     const std::string &payload) {
+    if (first)
+    {
+      first = false;
+    }
+    else
+    {
     EXPECT_EQ("MTConnect/Sample/000", topic);
 
     auto jdoc = json::parse(payload);
     auto streams = jdoc.at("/MTConnectStreams/Streams/0/DeviceStream"_json_pointer);
     EXPECT_EQ(string("LinuxCNC"), streams.at("/name"_json_pointer).get<string>());
 
     gotSample = true;
+    }
   };
 
   createClient(options, std::move(handler));
@@ -268,8 +275,8 @@ TEST_F(MqttSinkTest, mqtt_sink_should_publish_Sample)
 
   auto service = m_agentTestHelper->getMqttService();
 
-  ASSERT_TRUE(waitFor(60s, [&service]() { return service->isConnected(); }));
-  ASSERT_FALSE(gotSample);
+  ASSERT_TRUE(waitFor(60s, [&first]() { return !first; }));
+  ASSERT_FALSE(first);
 
   m_agentTestHelper->m_adapter->processData("2021-02-01T12:00:00Z|line|204");
   ASSERT_TRUE(waitFor(10s, [&gotSample]() { return gotSample; }));