Merge pull request #222 from mtconnect/mqtt_topic_initialization

Mqtt topic initialization

Author: wsobel

CMakeLists.txt

@@ -3,7 +3,7 @@ set(AGENT_VERSION_MAJOR 2)
 set(AGENT_VERSION_MINOR 0)
 set(AGENT_VERSION_PATCH 0)
 set(AGENT_VERSION_BUILD 12)
-set(AGENT_VERSION_RC "_RC21")
+set(AGENT_VERSION_RC "_RC22")
 
 # This minimum version is to support Visual Studio 2017 and C++ feature checking and FetchContent
 cmake_minimum_required(VERSION 3.16 FATAL_ERROR)

conan/mqtt_cpp/conanfile.py

@@ -3,7 +3,7 @@
 
 class MqttcppConan(ConanFile):
     name = "mqtt_cpp"
-    version = "11.0.0"
+    version = "13.1.0"
     license = "Boost Software License, Version 1.0"
     author = "Takatoshi Kondo [email protected]"
     url = "https://github.com/redboltz/mqtt_cpp"

conanfile.py

@@ -17,7 +17,7 @@ class CppAgentConan(ConanFile):
                 "date/2.4.1@#178e4ada4fefd011aaa81ab2bca646db",
                 "nlohmann_json/3.9.1@#a41bc0deaf7f40e7b97e548359ccf14d", 
                 "openssl/1.1.1k@#f40064b74987c778d5c4e0416d75f1f0",
-                "mqtt_cpp/11.0.0"]
+                "mqtt_cpp/13.1.0"]
 
     build_policy = "missing"
     default_options = {

demo/agent/agent.cfg

@@ -30,6 +30,11 @@ Directories {
   }
 }
 
+Sinks {
+  MqttService {
+  }
+}
+
 DevicesStyle { Location = /styles/styles.xsl }
 StreamsStyle { Location = /styles/styles.xsl }
 

src/configuration/config_options.hpp

@@ -66,6 +66,7 @@ namespace mtconnect {
     DECLARE_CONFIGURATION(MqttTls);
     DECLARE_CONFIGURATION(MqttPort);
     DECLARE_CONFIGURATION(MqttHost);
+    DECLARE_CONFIGURATION(MqttConnectInterval);
 
     // Adapter Configuration
     DECLARE_CONFIGURATION(AdapterIdentity);

src/mqtt/mqtt_client.hpp

@@ -40,7 +40,7 @@ namespace mtconnect {
     {
     public:
       MqttClient(boost::asio::io_context &ioc, std::unique_ptr<ClientHandler> &&handler)
-        : m_ioContext(ioc), m_handler(std::move(handler))
+        : m_ioContext(ioc), m_handler(std::move(handler)), m_connectInterval(5000)
       {}
       virtual ~MqttClient() = default;
       const auto &getIdentity() const { return m_identity; }
@@ -51,12 +51,14 @@ namespace mtconnect {
       virtual bool publish(const std::string &topic, const std::string &payload) = 0;
       auto isConnected() { return m_connected; }
       auto isRunning() { return m_running; }
+      void connectComplete() { m_connected = true; }
 
     protected:
       boost::asio::io_context &m_ioContext;
       std::string m_url;
       std::string m_identity;
       std::unique_ptr<ClientHandler> m_handler;
+      std::chrono::milliseconds m_connectInterval;
 
       bool m_running {false};
       bool m_connected {false};

src/mqtt/mqtt_client_impl.hpp

@@ -78,6 +78,10 @@ namespace mtconnect {
         identity.str("");
         identity << std::hex << digest[0] << digest[1] << digest[2];
         m_identity = std::string("_") + (identity.str()).substr(0, 10);
+
+        auto ci = GetOption<Seconds>(options, configuration::MqttConnectInterval);
+        if (ci)
+          m_connectInterval = *ci;
       }
 
       ~MqttClientImpl() { stop(); }
@@ -101,14 +105,21 @@ namespace mtconnect {
           }
           else if (ec == mqtt::connect_return_code::accepted)
           {
-            LOG(info) << "MQTT Connected";
-            m_connected = true;
+            LOG(info) << "MQTT ConnAck: MQTT Connected";
+
             if (m_handler && m_handler->m_connected)
+            {
               m_handler->m_connected(shared_from_this());
+            }
+            else
+            {
+              LOG(debug) << "No connect handler, setting connected";
+              m_connected = true;
+            }
           }
           else
           {
-            LOG(info) << "MQTT connection failed: " << ec;
+            LOG(info) << "MQTT ConnAck: MQTT connection failed: " << ec;
             reconnect();
           }
           return true;
@@ -236,7 +247,18 @@ namespace mtconnect {
         if (m_handler && m_handler->m_connecting)
           m_handler->m_connecting(shared_from_this());
 
-        derived().getClient()->async_connect();
+        derived().getClient()->async_connect([this](mqtt::error_code ec) {
+          if (ec)
+          {
+            LOG(warning) << "MqttClientImpl::connect: cannot connect: " << ec.message()
+                         << ", will retry";
+            reconnect();
+          }
+          else
+          {
+            LOG(info) << "MqttClientImpl::connect: connected";
+          }
+        });
       }
 
       void receive(mqtt::buffer &topic, mqtt::buffer &contents)
@@ -258,23 +280,24 @@ namespace mtconnect {
         LOG(info) << "Start reconnect timer";
 
         // Set an expiry time relative to now.
-        m_reconnectTimer.expires_after(std::chrono::seconds(5));
-
-        m_reconnectTimer.async_wait([this](const boost::system::error_code &error) {
-          if (error != boost::asio::error::operation_aborted)
-          {
-            LOG(info) << "MqttClientImpl::reconnect: reconnect now";
+        m_reconnectTimer.expires_after(m_connectInterval);
 
-            // Connect
-            derived().getClient()->async_connect([this](mqtt::error_code ec) {
-              LOG(info) << "MqttClientImpl::reconnect async_connect callback: " << ec.message();
-              if (ec && ec != boost::asio::error::operation_aborted)
+        m_reconnectTimer.async_wait(boost::asio::bind_executor(
+            derived().getClient()->get_executor(), [this](const boost::system::error_code &error) {
+              if (error != boost::asio::error::operation_aborted)
               {
-                reconnect();
+                LOG(info) << "MqttClientImpl::reconnect: reconnect now";
+
+                // Connect
+                derived().getClient()->async_connect([this](mqtt::error_code ec) {
+                  LOG(info) << "MqttClientImpl::reconnect async_connect callback: " << ec.message();
+                  if (ec && ec != boost::asio::error::operation_aborted)
+                  {
+                    reconnect();
+                  }
+                });
               }
-            });
-          }
-        });
+            }));
       }
 
     protected:

src/sink/mqtt_sink/mqtt_service.cpp

@@ -63,12 +63,16 @@ namespace mtconnect {
         auto clientHandler = make_unique<ClientHandler>();
         clientHandler->m_connected = [this](shared_ptr<MqttClient> client) {
           // Publish latest devices, assets, and observations
+          auto &circ = m_sinkContract->getCircularBuffer();
+          std::lock_guard<buffer::CircularBuffer> lock(circ);
+          client->connectComplete();
+
           for (auto &dev : m_sinkContract->getDevices())
           {
             publish(dev);
           }
 
-          for (auto &obs : m_sinkContract->getCircularBuffer().getLatest().getObservations())
+          for (auto &obs : circ.getLatest().getObservations())
           {
             observation::ObservationPtr p {obs.second};
             publish(p);

src/source/adapter/shdr/shdr_adapter.hpp

@@ -108,7 +108,7 @@ namespace mtconnect {
         for (auto &o : options)
           m_options.insert_or_assign(o.first, o.second);
         m_pipeline.build(m_options);
-        if (m_pipeline.started())
+        if (!m_pipeline.started())
           m_pipeline.start();
       }