Merge pull request #612 from mtconnect/611_obervation_duplication_when_using_agent_adapter

611 obervation duplication when using agent adapter

Author: wsobel

CMakeLists.txt

@@ -2,7 +2,7 @@
 set(AGENT_VERSION_MAJOR 2)
 set(AGENT_VERSION_MINOR 7)
 set(AGENT_VERSION_PATCH 0)
-set(AGENT_VERSION_BUILD 11)
+set(AGENT_VERSION_BUILD 12)
 set(AGENT_VERSION_RC "")
 
 # This minimum version is to support Visual Studio 2019 and C++ feature checking and FetchContent
@@ -71,6 +71,7 @@ include(cmake/remove_link_directories.cmake)
 include(cmake/osx_no_app_or_frameworks.cmake)
 include(cmake/ClangFormat.cmake)
 include(cmake/ClangTidy.cmake)
+include(cmake/Coverage.cmake)
 
 # Add our projects
 add_subdirectory(agent_lib)

cmake/Coverage.cmake

@@ -0,0 +1,27 @@
+# Code coverage instrumentation.
+#
+# Enable with -DAGENT_ENABLE_COVERAGE=ON (or, via Conan, -o "&:coverage=True").
+# Must be included before the agent/test targets are defined so the flags
+# propagate to every subsequently-declared target.
+#
+# Clang/AppleClang -> LLVM source-based coverage (llvm-profdata + llvm-cov)
+# GCC              -> gcov instrumentation (gcov/gcovr/lcov)
+
+option(AGENT_ENABLE_COVERAGE "Instrument the agent and tests for code coverage" OFF)
+
+if(AGENT_ENABLE_COVERAGE)
+  if(CMAKE_CXX_COMPILER_ID MATCHES "Clang")
+    message(STATUS "Coverage: enabling LLVM source-based coverage instrumentation")
+    add_compile_options(
+      $<$<COMPILE_LANGUAGE:CXX>:-fprofile-instr-generate>
+      $<$<COMPILE_LANGUAGE:CXX>:-fcoverage-mapping>)
+    add_link_options(-fprofile-instr-generate)
+  elseif(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
+    message(STATUS "Coverage: enabling gcov instrumentation")
+    add_compile_options($<$<COMPILE_LANGUAGE:CXX>:--coverage>)
+    add_link_options(--coverage)
+  else()
+    message(WARNING
+      "AGENT_ENABLE_COVERAGE requested but compiler '${CMAKE_CXX_COMPILER_ID}' is not supported")
+  endif()
+endif()

conanfile.py

@@ -14,8 +14,9 @@ class MTConnectAgentConan(ConanFile):
     license = "Apache License 2.0"
     settings = "os", "compiler", "arch", "build_type"
     options = { "without_ipv6": [True, False],
-                "with_ruby": [True, False], 
+                "with_ruby": [True, False],
                  "development" : [True, False],
+                 "coverage" : [True, False],
                  "shared": [True, False],
                  "winver": [None, "ANY"],
                  "with_docs" : [True, False],
@@ -34,6 +35,7 @@ class MTConnectAgentConan(ConanFile):
         "without_ipv6": False,
         "with_ruby": True,
         "development": False,
+        "coverage": False,
         "shared": False,
         "winver": "0x0A00",
         "with_docs": False,
@@ -177,6 +179,7 @@ def generate(self):
         tc.cache_variables['AGENT_WITH_DOCS'] = self.options.with_docs.__bool__()
         tc.cache_variables['AGENT_WITHOUT_IPV6'] = self.options.without_ipv6.__bool__()
         tc.cache_variables['DEVELOPMENT'] = self.options.development.__bool__()
+        tc.cache_variables['AGENT_ENABLE_COVERAGE'] = self.options.coverage.__bool__()
         if self.options.agent_prefix:
             tc.cache_variables['AGENT_PREFIX'] = self.options.agent_prefix
         if is_msvc(self):

src/mtconnect/sink/rest_sink/rest_service.cpp

@@ -74,8 +74,19 @@ namespace mtconnect {
 
       auto base = GetOption<string>(options, config::ExternalBaseUrl);
       if (!base)
-        base = "http://localhost:" + to_string(m_server->getPort());
-      m_externalBaseAddress = *base;
+      {
+        auto ip = GetOption<string>(options, config::ServerIp);
+        if (!ip || *ip == "0.0.0.0")
+          ip = GetBestHostAddress(m_context, true);
+        string protocol;
+        if (m_server->isTlsEnabled())
+          protocol = "https://";
+        else
+          protocol = "http://";
+        base = protocol + *ip + ":" + to_string(m_server->getPort()) + '/';
+      }
+      m_baseUrl = *base;
+      m_server->setBaseUrl(m_baseUrl);
 
       auto xmlPrinter = dynamic_cast<XmlPrinter *>(m_sinkContract->getPrinter("xml"));
       auto jsonPrinter = dynamic_cast<JsonPrinter *>(m_sinkContract->getPrinter("json"));

src/mtconnect/sink/rest_sink/rest_service.hpp

@@ -380,7 +380,7 @@ namespace mtconnect {
 
       std::string externalUrl(const std::string &url)
       {
-        std::string fullUrl = m_externalBaseAddress;
+        std::string fullUrl = m_baseUrl;
         if (!fullUrl.empty() && !url.empty())
         {
           if (fullUrl.back() == '/' && url.front() == '/')
@@ -405,7 +405,7 @@ namespace mtconnect {
       std::shared_ptr<source::LoopbackSource> m_loopback;
       uint64_t m_instanceId;
       std::unique_ptr<Server> m_server;
-      std::string m_externalBaseAddress;
+      std::string m_baseUrl;
 
       // Buffers
       FileCache m_fileCache;

src/mtconnect/sink/rest_sink/server.cpp

@@ -371,15 +371,7 @@ namespace mtconnect::sink::rest_sink {
         AutoJsonArray<T> ary(writer, "servers");
         {
           AutoJsonObject<T> obj(writer);
-
-          stringstream str;
-          if (m_tlsEnabled)
-            str << "https://";
-          else
-            str << "http://";
-
-          str << GetBestHostAddress(m_context, true) << ':' << m_port << '/';
-          obj.AddPairs("url", str.str());
+          obj.AddPairs("url", m_baseUrl);
         }
       }
       {

src/mtconnect/sink/rest_sink/server.hpp

@@ -124,6 +124,14 @@ namespace mtconnect::sink::rest_sink {
     /// @return the port being bound
     auto getPort() const { return m_port; }
 
+    ///@brief Is TLS enabled?
+    ///@return `true` if TLS is enabled
+    bool isTlsEnabled() const { return m_tlsEnabled; }
+
+    /// @brief Set the external base URL
+    /// @param[in] url the base URL to set
+    void setBaseUrl(const std::string &url) { m_baseUrl = url; }
+
     /// @name PUT and POST handling
     ///@{
 
@@ -318,6 +326,7 @@ namespace mtconnect::sink::rest_sink {
 
     boost::asio::ip::address m_address;
     unsigned short m_port {5000};
+    std::string m_baseUrl;
 
     bool m_run {false};
     bool m_listening {false};

src/mtconnect/source/adapter/agent_adapter/agent_adapter.cpp

@@ -232,8 +232,12 @@ namespace mtconnect::source::adapter::agent_adapter {
         m_strand, [weak = std::weak_ptr(getptr())](boost::system::error_code ec) {
           if (auto self = weak.lock(); self && !ec)
           {
+            // Rebuild the sample request from the latest processed sequence rather
+            // than replaying the cached request. In streaming mode the cached
+            // request's `from` is baked in when the stream opens and never advances,
+            // so replaying it re-delivers every observation seen during the session.
             if (self->canRecover() && self->m_streamRequest)
-              self->m_session->makeRequest(*self->m_streamRequest);
+              self->recover();
             else
               self->run();
           }
@@ -352,7 +356,6 @@ namespace mtconnect::source::adapter::agent_adapter {
 
     clear();
 
-    m_reconnecting = false;
     if (m_probeAgent)
     {
       probe();
@@ -369,7 +372,6 @@ namespace mtconnect::source::adapter::agent_adapter {
     if (m_stopped)
       return;
 
-    m_reconnecting = false;
     sample();
   }
 

src/mtconnect/source/adapter/agent_adapter/agent_adapter.hpp

@@ -89,6 +89,10 @@ namespace mtconnect::source::adapter::agent_adapter {
     /// @return reference to the transform feedback
     auto &getFeedback() { return m_feedback; }
 
+    /// @brief get the current outstanding stream request (for testing)
+    /// @return reference to the optional stream request
+    auto &getStreamRequest() { return m_streamRequest; }
+
     ~AgentAdapter() override;
 
     /// @name Source interface
@@ -128,7 +132,6 @@ namespace mtconnect::source::adapter::agent_adapter {
     url::Url m_url;
     int m_count = 1000;
     std::chrono::milliseconds m_heartbeat;
-    bool m_reconnecting = false;
     bool m_failed = false;
     bool m_stopped = false;
     bool m_usePolling = false;