Merge pull request #451 from mtconnect/431_websockets_subscriptions

431 websockets subscriptions

README.md

@@ -1,5 +1,5 @@
 
-MTConnect C++ Agent Version 2.3
+MTConnect C++ Agent Version 2.5
 --------
 [![Build MTConnect C++ Agent](https://github.com/mtconnect/cppagent/actions/workflows/build.yml/badge.svg)](https://github.com/mtconnect/cppagent/actions/workflows/build.yml)
 
@@ -13,6 +13,10 @@ the devices and the location of the adapter.
 
 Pre-built binary releases for Windows are available from [Releases](https://github.com/mtconnect/cppagent/releases) for those who do not want to build the agent themselves. For *NIX users, you will need libxml2, cppunit, and cmake as well as build essentials.
 
+Version 2.5.0 Added validation of observations in the stream
+
+Version 2.4.0 Added support for version 2.4
+
 Version 2.3.0 Support for all Version 2.3 standard changes and JSON ingress to MQTT adapter.
 
 Version 2.2.0 Support for all Version 2.2 standard changes and dynamic configuration from adapters. Upgrade to conan 2.
@@ -639,11 +643,17 @@ Configuration Parameters
 * `SuppressIPAddress` - Suppress the Adapter IP Address and port when creating the Agent Device ids and names. This applies to all adapters.
 
     *Default*: false
+
+* `Validation` - Turns on validation of model components and observations
+
+    *Default*: false
 
 * `WorkerThreads` - The number of operating system threads dedicated to the Agent
 
     *Default*: 1
 
+
+
 #### Adapter General Configuration
 
 These can be overridden on a per-adapter basis

agent_lib/CMakeLists.txt

@@ -246,13 +246,11 @@ set(AGENT_SOURCES
 
 # src/sink/mqtt_sink HEADER_FILE_ONLY
 
-        "${SOURCE_DIR}/sink/mqtt_sink/mqtt_service.hpp"
-	"${SOURCE_DIR}/sink/mqtt_sink/mqtt2_service.hpp"
+	"${SOURCE_DIR}/sink/mqtt_sink/mqtt_service.hpp"
 
 #src/sink/mqtt_sink SOURCE_FILES_ONLY
 
-        "${SOURCE_DIR}/sink/mqtt_sink/mqtt_service.cpp"
-	"${SOURCE_DIR}/sink/mqtt_sink/mqtt2_service.cpp"
+	"${SOURCE_DIR}/sink/mqtt_sink/mqtt_service.cpp"
 
 # src/sink/rest_sink HEADER_FILE_ONLY
 
@@ -267,6 +265,7 @@ set(AGENT_SOURCES
         "${SOURCE_DIR}/sink/rest_sink/session.hpp"
         "${SOURCE_DIR}/sink/rest_sink/session_impl.hpp"
         "${SOURCE_DIR}/sink/rest_sink/tls_dector.hpp"
+        "${SOURCE_DIR}/sink/rest_sink/websocket_session.hpp"
 
 # src/sink/rest_sink SOURCE_FILES_ONLY
 
@@ -335,7 +334,6 @@ if(MSVC)
   # The modules including Beast required the /bigobj option in Windows
   set_property(SOURCE    
     "${SOURCE_DIR}/sink/mqtt_sink/mqtt_service.cpp"
-    "${SOURCE_DIR}/sink/mqtt_sink/mqtt2_service.cpp"
     "${SOURCE_DIR}/sink/rest_sink/session_impl.cpp"
     "${SOURCE_DIR}/source/adapter/mqtt/mqtt_adapter.cpp"
     "${SOURCE_DIR}/source/adapter/agent_adapter/agent_adapter.cpp"

conan/mqtt_cpp/conanfile.py

@@ -10,7 +10,7 @@ class MqttcppConan(ConanFile):
     url = "https://github.com/redboltz/mqtt_cpp"
     description = "MQTT client/server for C++14 based on Boost.Asio"
     topics = ("mqtt")
-    requires = ["boost/1.82.0"]
+    requires = ["boost/1.84.0"]
     no_copy_source = True
     exports_sources = "include/*"
 

conan/profiles/vs32

@@ -8,3 +8,5 @@ compiler.runtime=static
 compiler.runtime_type=Release
 build_type=Release
 
+[options]
+winver=0x0600

conan/profiles/vs32debug

@@ -7,3 +7,6 @@ arch=x86
 compiler.runtime=static
 compiler.runtime_type=Debug
 build_type=Debug
+
+[options]
+winver=0x0600

conan/profiles/vs32shared

@@ -10,3 +10,4 @@ build_type=Release
 
 [options]
 shared=True
+winver=0x0600

conanfile.py

@@ -35,7 +35,7 @@ class MTConnectAgentConan(ConanFile):
         "with_ruby": True,
         "development": False,
         "shared": False,
-        "winver": "0x600",
+        "winver": "0x0602",
         "with_docs": False,
         "cpack": False,
         "agent_prefix": None,
@@ -118,7 +118,7 @@ def build_requirements(self):
             self.tool_requires_version("doxygen", [1, 9, 4])
 
     def requirements(self):
-        self.requires("boost/1.82.0", headers=True, libs=True, transitive_headers=True, transitive_libs=True)
+        self.requires("boost/1.84.0", headers=True, libs=True, transitive_headers=True, transitive_libs=True)
         self.requires("libxml2/2.10.3", headers=True, libs=True, visible=True, transitive_headers=True, transitive_libs=True)
         self.requires("date/2.4.1", headers=True, libs=True, transitive_headers=True, transitive_libs=True)
         self.requires("nlohmann_json/3.9.1", headers=True, libs=False, transitive_headers=True, transitive_libs=False)
@@ -139,6 +139,9 @@ def configure(self):
         if self.options.shared:
             self.options["boost/*"].shared = True
             self.package_type = "shared-library"
+
+        if is_msvc(self):
+            self.options["boost/*"].extra_b2_flags = ("define=BOOST_USE_WINAPI_VERSION=" + str(self.options.winver))
 
         # Make sure shared builds use shared boost
         if is_msvc(self) and self.options.shared:
@@ -227,6 +230,7 @@ def package_info(self):
             winver=str(self.options.winver)
             self.cpp_info.defines.append("WINVER=" + winver)
             self.cpp_info.defines.append("_WIN32_WINNT=" + winver)
+            self.cpp_info.defines.append("BOOST_USE_WINAPI_VERSION=" + winver)
 
     def package(self):
         cmake = CMake(self)

docker/ubuntu/Dockerfile

@@ -63,7 +63,7 @@ RUN  apt-get update \
        rake \
        ruby \
   && rm -rf /var/lib/apt/lists/* \
-  && pip install conan -v 'conan==2.0.9'
+  && pip install conan
 
 # make an agent directory and cd into it
 WORKDIR /root/agent

src/mtconnect/agent.cpp

@@ -299,15 +299,15 @@ namespace mtconnect {
       {
         if (item.expired())
           continue;
-        
+
         auto di = item.lock();
         if (di->hasInitialValue())
         {
           m_loopback->receive(di, *di->getInitialValue());
         }
       }
     }
-    
+
     std::lock_guard<buffer::CircularBuffer> lock(m_circularBuffer);
     if (m_circularBuffer.addToBuffer(observation) != 0)
     {

src/mtconnect/configuration/agent_config.cpp

@@ -58,7 +58,6 @@
 #include "mtconnect/configuration/config_options.hpp"
 #include "mtconnect/device_model/device.hpp"
 #include "mtconnect/printer/xml_printer.hpp"
-#include "mtconnect/sink/mqtt_sink/mqtt2_service.hpp"
 #include "mtconnect/sink/mqtt_sink/mqtt_service.hpp"
 #include "mtconnect/sink/rest_sink/rest_service.hpp"
 #include "mtconnect/source/adapter/agent_adapter/agent_adapter.hpp"
@@ -113,7 +112,6 @@ namespace mtconnect::configuration {
     bool success = false;
 
     sink::mqtt_sink::MqttService::registerFactory(m_sinkFactory);
-    sink::mqtt_sink::Mqtt2Service::registerFactory(m_sinkFactory);
     sink::rest_sink::RestService::registerFactory(m_sinkFactory);
     adapter::shdr::ShdrAdapter::registerFactory(m_sourceFactory);
     adapter::mqtt_adapter::MqttAdapter::registerFactory(m_sourceFactory);

src/mtconnect/device_model/data_item/data_item.cpp

@@ -201,7 +201,7 @@ namespace mtconnect {
           }
         }
       }
-      
+
       if (const auto &init = maybeGet<string>("InitialValue"); init)
       {
         m_initialValue = *init;

src/mtconnect/device_model/data_item/data_item.hpp

@@ -139,11 +139,11 @@ namespace mtconnect {
         /// @brief get the topic name leaf node for this data item
         /// @return the topic name
         const auto &getTopicName() const { return m_topicName; }
-        
+
         /// @brief get the initial value if one is set
         /// @return optional initial value
         const auto &getInitialValue() const { return m_initialValue; }
-        
+
         Category getCategory() const { return m_category; }
         Representation getRepresentation() const { return m_representation; }
         SpecialClass getSpecialClass() const { return m_specialClass; }

src/mtconnect/mqtt/mqtt_client_impl.hpp

@@ -164,7 +164,6 @@ namespace mtconnect {
           m_connected = false;
           if (m_handler && m_handler->m_disconnected)
             m_handler->m_disconnected(shared_from_this());
-            m_handler->m_disconnected(shared_from_this());
           if (m_running)
           {
             reconnect();
@@ -419,7 +418,7 @@ namespace mtconnect {
       {
         return static_pointer_cast<MqttTcpClient>(shared_from_this());
       }
-      
+
       /// @brief Get the Mqtt TCP Client
       /// @return pointer to the Mqtt TCP Client
       auto &getClient()
@@ -501,7 +500,7 @@ namespace mtconnect {
       {
         return static_pointer_cast<MqttTlsWSClient>(shared_from_this());
       }
-      
+
       /// @brief Get the Mqtt TLS WebSocket Client
       /// @return pointer to the Mqtt TLS WebSocket Client
       auto &getClient()
@@ -540,7 +539,7 @@ namespace mtconnect {
       {
         return static_pointer_cast<MqttWSClient>(shared_from_this());
       }
-      
+
       /// @brief Get the Mqtt TLS WebSocket Client
       /// @return pointer to the Mqtt TLS WebSocket Client
       auto &getClient()

src/mtconnect/observation/change_observer.cpp

@@ -29,14 +29,25 @@ using namespace std;
 namespace mtconnect::observation {
   ChangeObserver::~ChangeObserver()
   {
+    std::lock_guard<std::recursive_mutex> scopedLock(m_mutex);
+    clear();
+  }
+
+  void ChangeObserver::clear()
+  {
+    std::unique_lock<std::recursive_mutex> lock(m_mutex);
+    m_timer.cancel();
+    m_handler.clear();
     for (const auto signaler : m_signalers)
       signaler->removeObserver(this);
+    m_signalers.clear();
   }
 
   void ChangeObserver::addSignaler(ChangeSignaler *sig) { m_signalers.emplace_back(sig); }
 
   bool ChangeObserver::removeSignaler(ChangeSignaler *sig)
   {
+    std::lock_guard<std::recursive_mutex> scopedLock(m_mutex);
     auto newEndPos = std::remove(m_signalers.begin(), m_signalers.end(), sig);
     if (newEndPos == m_signalers.end())
       return false;
@@ -47,7 +58,8 @@ namespace mtconnect::observation {
 
   void ChangeObserver::handler(boost::system::error_code ec)
   {
-    boost::asio::post(m_strand, boost::bind(m_handler, ec));
+    if (m_handler)
+      boost::asio::post(m_strand, boost::bind(m_handler, ec));
   }
 
   // Signaler Management
@@ -99,7 +111,7 @@ namespace mtconnect::observation {
                                buffer::CircularBuffer &buffer, FilterSet &&filter,
                                std::chrono::milliseconds interval,
                                std::chrono::milliseconds heartbeat)
-    : m_interval(interval),
+    : AsyncResponse(interval),
       m_heartbeat(heartbeat),
       m_last(std::chrono::system_clock::now()),
       m_filter(std::move(filter)),
@@ -145,9 +157,12 @@ namespace mtconnect::observation {
 
   void AsyncObserver::handlerCompleted()
   {
+    NAMED_SCOPE("AsyncObserver::handlerCompleted");
+
     m_last = std::chrono::system_clock::now();
     if (m_endOfBuffer)
     {
+      LOG(trace) << "End of buffer";
       using std::placeholders::_1;
       m_observer.waitForSignal(m_heartbeat);
     }
@@ -159,6 +174,7 @@ namespace mtconnect::observation {
 
   void AsyncObserver::handleSignal(boost::system::error_code ec)
   {
+    NAMED_SCOPE("AsyncObserver::handleSignal");
     using namespace buffer;
 
     using std::placeholders::_1;

src/mtconnect/observation/change_observer.hpp

@@ -143,6 +143,9 @@ namespace mtconnect::observation {
     auto try_lock() { return m_mutex.try_lock(); }
     ///@}
 
+    /// @brief clear the observer information.
+    void clear();
+
   private:
     boost::asio::io_context::strand &m_strand;
     mutable std::recursive_mutex m_mutex;
@@ -185,6 +188,32 @@ namespace mtconnect::observation {
     std::list<ChangeObserver *> m_observers;
   };
 
+  /// @brief Abstract class for things asynchronouos timers
+  class AGENT_LIB_API AsyncResponse : public std::enable_shared_from_this<AsyncResponse>
+  {
+  public:
+    AsyncResponse(std::chrono::milliseconds interval) : m_interval(interval) {}
+
+    virtual bool cancel() = 0;
+
+    /// @brief method to determine if the sink is running
+    virtual bool isRunning() = 0;
+
+    /// @brief get the request id for webservices
+    const auto &getRequestId() const { return m_requestId; }
+
+    /// @brief sets the optonal request id for webservices.
+    void setRequestId(const std::optional<std::string> &id) { m_requestId = id; }
+
+    /// @brief Get the interval
+    const auto &getInterval() const { return m_interval; }
+
+  protected:
+    std::chrono::milliseconds m_interval {
+        0};  //! the minimum amout of time to wait before calling the handler
+    std::optional<std::string> m_requestId;  //! request id
+  };
+
   /// @brief Asyncronous change context for waiting for changes
   ///
   /// This class must be subclassed and provide a fail and isRunning method.
@@ -196,7 +225,7 @@ namespace mtconnect::observation {
   ///
   /// The handler and sequence numbers are handled inside the circular buffer lock to prevent race
   /// conditions with incoming data.
-  class AGENT_LIB_API AsyncObserver : public std::enable_shared_from_this<AsyncObserver>
+  class AGENT_LIB_API AsyncObserver : public AsyncResponse
   {
   public:
     /// @Brief callback when observations are ready
@@ -217,7 +246,7 @@ namespace mtconnect::observation {
     virtual ~AsyncObserver() = default;
 
     /// @brief Get a shared pointed
-    auto getptr() const { return const_cast<AsyncObserver *>(this)->shared_from_this(); }
+    auto getptr() { return std::dynamic_pointer_cast<AsyncObserver>(shared_from_this()); }
 
     /// @brief sets up the `ChangeObserver` using the filter and initializes the references to the
     /// buffer
@@ -235,8 +264,12 @@ namespace mtconnect::observation {
     /// @brief abstract call to failure handler
     virtual void fail(boost::beast::http::status status, const std::string &message) = 0;
 
-    /// @brief method to determine if the sink is running
-    virtual bool isRunning() = 0;
+    /// @brief Stop all timers and release resources.
+    bool cancel() override
+    {
+      m_observer.clear();
+      return true;
+    }
 
     /// @brief handler callback when an action needs to be taken
     ///
@@ -250,9 +283,7 @@ namespace mtconnect::observation {
     auto getSequence() const { return m_sequence; }
     auto isEndOfBuffer() const { return m_endOfBuffer; }
     const auto &getFilter() const { return m_filter; }
-
     ///@}
-    ///
 
     mutable bool m_endOfBuffer {false};  //! Public indicator that we are at the end of the buffer
 
@@ -266,8 +297,6 @@ namespace mtconnect::observation {
 
   protected:
     SequenceNumber_t m_sequence {0};  //! the current sequence number
-    std::chrono::milliseconds m_interval {
-        0};  //! the minimum amout of time to wait before calling the handler
     std::chrono::milliseconds m_heartbeat {
         0};  //! the maximum amount of time to wait before sending a heartbeat
     std::chrono::system_clock::time_point m_last;  //! the last time the handler completed