Merge pull request mtconnect#374 from mtconnect/mqtt_last_will_and_client_id_fix

Mqtt last will and client id fix

Author: wsobel

README.md

@@ -833,6 +833,13 @@ Sinks {
 
     *Default*: `MTConnect/Current/[device]`
 
+* `MqttLastWillTopic` - The topic used for the last will and testement for an agent
+
+    > Note: The value will be `AVAILABLE` when the Agent is publishing and connected and will
+    > publish `UNAVAILABLE` when the agent disconnects from the broker.
+
+    *Default*: `MTConnect/Probe/[device]/Availability"`
+
 * `MqttCurrentInterval` - The frequency to publish currents. Acts like a keyframe in a video stream.
 
     *Default*: 10000ms
@@ -844,7 +851,7 @@ Sinks {
 * `MqttSampleCount` - The maxmimum number of observations to publish at one time.
 
     *Default*: 1000
-
+    
 ### Adapter Configuration Items ###
 
 * `Adapters` - Adapters begins a list of device blocks. If the Adapters

src/mtconnect/configuration/config_options.hpp

@@ -99,6 +99,7 @@ namespace mtconnect {
     DECLARE_CONFIGURATION(MqttUserName);
     DECLARE_CONFIGURATION(MqttPassword);
     DECLARE_CONFIGURATION(MqttMaxTopicDepth);
+    DECLARE_CONFIGURATION(MqttLastWillTopic);
     ///@}
 
     /// @name Adapter Configuration

src/mtconnect/mqtt/mqtt_client.hpp

@@ -47,8 +47,14 @@ namespace mtconnect {
       /// @param ClientHandler configuration options
       /// - ConnectInterval, defaults to 5000
 
-      MqttClient(boost::asio::io_context &ioc, std::unique_ptr<ClientHandler> &&handler)
-        : m_ioContext(ioc), m_handler(std::move(handler)), m_connectInterval(5000)
+      MqttClient(boost::asio::io_context &ioc, std::unique_ptr<ClientHandler> &&handler,
+                 const std::optional<std::string> willTopic = std::nullopt,
+                 const std::optional<std::string> willPayload = std::nullopt)
+        : m_ioContext(ioc),
+          m_handler(std::move(handler)),
+          m_connectInterval(5000),
+          m_willTopic(willTopic),
+          m_willPayload(willPayload)
       {}
       virtual ~MqttClient() = default;
 
@@ -101,6 +107,8 @@ namespace mtconnect {
       std::string m_identity;
       std::unique_ptr<ClientHandler> m_handler;
       std::chrono::milliseconds m_connectInterval;
+      std::optional<std::string> m_willTopic;
+      std::optional<std::string> m_willPayload;
 
       bool m_running {false};
       bool m_connected {false};

src/mtconnect/mqtt/mqtt_client_impl.hpp

@@ -20,12 +20,12 @@
 #include <boost/log/trivial.hpp>
 #include <boost/uuid/name_generator_sha1.hpp>
 
-#include <inttypes.h>
-#include <random>
 #include <chrono>
-
+#include <inttypes.h>
 #include <mqtt/async_client.hpp>
 #include <mqtt/setup_log.hpp>
+#include <mqtt/will.hpp>
+#include <random>
 
 #include "mqtt_client.hpp"
 #include "mtconnect/configuration/config_options.hpp"
@@ -72,8 +72,10 @@ namespace mtconnect {
       /// - MqttTls, defaults to false
       /// - MqttHost, defaults to LocalHost
       MqttClientImpl(boost::asio::io_context &ioContext, const ConfigOptions &options,
-                     std::unique_ptr<ClientHandler> &&handler)
-        : MqttClient(ioContext, std::move(handler)),
+                     std::unique_ptr<ClientHandler> &&handler,
+                     const std::optional<std::string> willTopic = std::nullopt,
+                     const std::optional<std::string> willPayload = std::nullopt)
+        : MqttClient(ioContext, std::move(handler), willTopic, willPayload),
           m_options(options),
           m_host(GetOption<std::string>(options, configuration::MqttHost).value_or("localhost")),
           m_port(GetOption<int>(options, configuration::MqttPort).value_or(1883)),
@@ -197,6 +199,19 @@ namespace mtconnect {
           }
         });
 
+        if (m_willTopic && m_willPayload)
+        {
+          uint32_t will_expiry_interval = 1;
+          MQTT_NS::v5::properties ps {
+              MQTT_NS::v5::property::message_expiry_interval(will_expiry_interval),
+          };
+
+          mqtt::buffer topic(std::string_view(m_willTopic->c_str()));
+          mqtt::buffer payload(std::string_view(m_willPayload->c_str()));
+
+          client->set_will(mqtt::will(topic, payload, mqtt::retain::yes, mqtt::force_move(ps)));
+        }
+
         m_running = true;
         connect();
 
@@ -383,7 +398,6 @@ namespace mtconnect {
       std::uint16_t m_packetId {0};
 
       std::optional<std::string> m_username;
-
       std::optional<std::string> m_password;
 
       boost::asio::steady_timer m_reconnectTimer;

src/mtconnect/mqtt/mqtt_server.hpp

@@ -44,10 +44,13 @@ namespace mtconnect {
       /// @brief Shutdown the Mqtt server
       virtual void stop() = 0;
 
+      auto &getWill() { return m_will; }
+
     protected:
       boost::asio::io_context &m_ioContext;
       std::string m_url;
       uint16_t m_port;
+      std::optional<mqtt::will> m_will;
     };
   }  // namespace mqtt_server
 }  // namespace mtconnect

src/mtconnect/mqtt/mqtt_server_impl.hpp

@@ -125,8 +125,8 @@ namespace mtconnect {
           ep.set_connect_handler([this, wp](MQTT_NS::buffer client_id,
                                             MQTT_NS::optional<MQTT_NS::buffer> username,
                                             MQTT_NS::optional<MQTT_NS::buffer> password,
-                                            MQTT_NS::optional<MQTT_NS::will>, bool clean_session,
-                                            std::uint16_t keep_alive) {
+                                            MQTT_NS::optional<MQTT_NS::will> will,
+                                            bool clean_session, std::uint16_t keep_alive) {
             using namespace MQTT_NS::literals;
             LOG(info) << "Server: Client_id    : " << client_id << std::endl;
             LOG(info) << "Server: User Name     : " << (username ? username.value() : "none"_mb)
@@ -141,6 +141,8 @@ namespace mtconnect {
               LOG(error) << "Server: Endpoint has been deleted";
               return false;
             }
+            if (will)
+              m_will = will;
             m_connections.insert(sp);
             sp->connack(false, MQTT_NS::connect_return_code::accepted);
             return true;

src/mtconnect/pipeline/json_mapper.cpp

@@ -175,20 +175,14 @@ namespace mtconnect::pipeline {
     Forward m_forward;
     std::list<pair<DataItemPtr, entity::Properties>> m_queue;
   };
-  
+
   /// @brief consume value in case of error
   struct ErrorHandler : rj::BaseReaderHandler<rj::UTF8<>, ErrorHandler>
   {
     ErrorHandler(int depth = 0) : m_depth(depth) {}
-    
-    bool Default()
-    {
-      return true;
-    }
-    bool Key(const Ch *str, rj::SizeType length, bool copy)
-    {
-      return true;
-    }
+
+    bool Default() { return true; }
+    bool Key(const Ch *str, rj::SizeType length, bool copy) { return true; }
     bool StartObject()
     {
       m_depth++;
@@ -199,21 +193,21 @@ namespace mtconnect::pipeline {
       m_depth--;
       return true;
     }
-    bool StartArray() 
+    bool StartArray()
     {
       m_depth++;
       return true;
     }
-    bool EndArray(rj::SizeType elementCount) 
+    bool EndArray(rj::SizeType elementCount)
     {
       m_depth--;
       return true;
     }
-    
+
     bool operator()(rj::Reader &reader, rj::StringStream &buff)
     {
       LOG(warning) << "Consuming value due to error";
-      
+
       if (!reader.IterativeParseNext<rj::kParseNanAndInfFlag>(buff, *this))
         return false;
 
@@ -223,7 +217,7 @@ namespace mtconnect::pipeline {
         if (!reader.IterativeParseNext<rj::kParseNanAndInfFlag>(buff, *this))
           return false;
       }
-      
+
       return true;
     }
 
@@ -618,7 +612,7 @@ namespace mtconnect::pipeline {
     bool m_object {false};
     std::string m_key;
     Expectation m_expectation {Expectation::NONE};
-    int m_depth{0};
+    int m_depth {0};
   };
 
   struct TimestampHandler : rj::BaseReaderHandler<rj::UTF8<>, TimestampHandler>
@@ -862,7 +856,7 @@ namespace mtconnect::pipeline {
               m_expectation = Expectation::KEY;
               break;
             }
-              
+
             case Expectation::VALUE_ERROR:
             {
               ErrorHandler handler;

src/mtconnect/sink/mqtt_sink/mqtt2_service.cpp

@@ -69,27 +69,19 @@ namespace mtconnect {
                     {configuration::MqttClientId, string()},
                     {configuration::MqttUserName, string()},
                     {configuration::MqttPassword, string()}});
-        AddDefaultedOptions(config, m_options,
-                            {{configuration::MqttHost, "127.0.0.1"s},
-                             {configuration::DeviceTopic, "MTConnect/Probe/[device]"s},
-                             {configuration::AssetTopic, "MTConnect/Asset/[device]"s},
-                             {configuration::CurrentTopic, "MTConnect/Current/[device]"s},
-                             {configuration::SampleTopic, "MTConnect/Sample/[device]"s},
-                             {configuration::MqttCurrentInterval, 10000ms},
-                             {configuration::MqttSampleInterval, 500ms},
-                             {configuration::MqttSampleCount, 1000},
-                             {configuration::MqttPort, 1883},
-                             {configuration::MqttTls, false}});
-
-        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();
-
-          pubishInitialContent();
-        };
+        AddDefaultedOptions(
+            config, m_options,
+            {{configuration::MqttHost, "127.0.0.1"s},
+             {configuration::DeviceTopic, "MTConnect/Probe/[device]"s},
+             {configuration::AssetTopic, "MTConnect/Asset/[device]"s},
+             {configuration::MqttLastWillTopic, "MTConnect/Probe/[device]/Availability"s},
+             {configuration::CurrentTopic, "MTConnect/Current/[device]"s},
+             {configuration::SampleTopic, "MTConnect/Sample/[device]"s},
+             {configuration::MqttCurrentInterval, 10000ms},
+             {configuration::MqttSampleInterval, 500ms},
+             {configuration::MqttSampleCount, 1000},
+             {configuration::MqttPort, 1883},
+             {configuration::MqttTls, false}});
 
         int maxTopicDepth {GetOption<int>(options, configuration::MqttMaxTopicDepth).value_or(7)};
 
@@ -103,23 +95,38 @@ namespace mtconnect {
         m_sampleInterval = *GetOption<Milliseconds>(m_options, configuration::MqttSampleInterval);
 
         m_sampleCount = *GetOption<int>(m_options, configuration::MqttSampleCount);
-
-        if (IsOptionSet(m_options, configuration::MqttTls))
-        {
-          m_client = make_shared<MqttTlsClient>(m_context, m_options, std::move(clientHandler));
-        }
-        else
-        {
-          m_client = make_shared<MqttTcpClient>(m_context, m_options, std::move(clientHandler));
-        }
       }
 
       void Mqtt2Service::start()
       {
-        // mqtt client side not a server side...
         if (!m_client)
-          return;
-
+        {
+          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();
+
+            client->publish(m_lastWillTopic, "AVAILABLE");
+            pubishInitialContent();
+          };
+
+          auto agentDevice = m_sinkContract->getDeviceByName("Agent");
+          auto lwtTopic = get<string>(m_options[configuration::MqttLastWillTopic]);
+          m_lastWillTopic = formatTopic(lwtTopic, agentDevice, "Agent");
+
+          if (IsOptionSet(m_options, configuration::MqttTls))
+          {
+            m_client = make_shared<MqttTlsClient>(m_context, m_options, std::move(clientHandler),
+                                                  m_lastWillTopic, "UNAVAILABLE"s);
+          }
+          else
+          {
+            m_client = make_shared<MqttTcpClient>(m_context, m_options, std::move(clientHandler),
+                                                  m_lastWillTopic, "UNAVAILABLE"s);
+          }
+        }
         m_client->start();
       }
 

src/mtconnect/sink/mqtt_sink/mqtt2_service.hpp

@@ -132,12 +132,13 @@ namespace mtconnect {
           return filter->second;
         }
 
-        std::string formatTopic(const std::string &topic, const DevicePtr device)
+        std::string formatTopic(const std::string &topic, const DevicePtr device,
+                                const std::string defaultUuid = "Unknown")
         {
           string uuid;
           string formatted {topic};
           if (!device)
-            uuid = "Unknown";
+            uuid = defaultUuid;
           else
           {
             uuid = *(device->getUuid());
@@ -173,10 +174,11 @@ namespace mtconnect {
         }
 
       protected:
-        std::string m_deviceTopic;   //! Device topic prefix
-        std::string m_assetTopic;    //! Asset topic prefix
-        std::string m_currentTopic;  //! Current topic prefix
-        std::string m_sampleTopic;   //! Sample topic prefix
+        std::string m_deviceTopic;    //! Device topic prefix
+        std::string m_assetTopic;     //! Asset topic prefix
+        std::string m_currentTopic;   //! Current topic prefix
+        std::string m_sampleTopic;    //! Sample topic prefix
+        std::string m_lastWillTopic;  //! Topic to publish the last will when disconnected
 
         std::chrono::milliseconds m_currentInterval;  //! Interval in ms to update current
         std::chrono::milliseconds m_sampleInterval;   //! min interval in ms to update sample

test_package/json_mapping_test.cpp

@@ -1009,13 +1009,8 @@ TEST_F(JsonMappingTest, should_parse_xml_asset)
 TEST_F(JsonMappingTest, should_skip_erroneous_values)
 {
   auto dev = makeDevice("Device", {{"id", "device"s}, {"name", "device"s}, {"uuid", "device"s}});
-  makeDataItem("device", {{"id", "a"s},
-    {"type", "EXECUTION"s},
-    {"category", "EVENT"s}});
-  makeDataItem("device", {{"id", "b"s},
-    {"type", "CONTROLLER_MODE"s},
-    {"category", "EVENT"s}});
-  
+  makeDataItem("device", {{"id", "a"s}, {"type", "EXECUTION"s}, {"category", "EVENT"s}});
+  makeDataItem("device", {{"id", "b"s}, {"type", "CONTROLLER_MODE"s}, {"category", "EVENT"s}});
 
   Properties props {{"VALUE", R"(
 {
@@ -1031,7 +1026,7 @@ TEST_F(JsonMappingTest, should_skip_erroneous_values)
     },
    "b": "MANUAL"
 })"s}};
-  
+
   auto jmsg = std::make_shared<JsonMessage>("JsonMessage", props);
   jmsg->m_device = dev;
 
@@ -1049,6 +1044,6 @@ TEST_F(JsonMappingTest, should_skip_erroneous_values)
   ASSERT_EQ("b", obs->getDataItem()->getId());
   ASSERT_EQ("MANUAL", obs->getValue<string>());
 }
-    
+
 /// @test verify the json mapper can an asset in json
 TEST_F(JsonMappingTest, should_parse_json_asset) { GTEST_SKIP(); }