Merge pull request #372 from mtconnect/mqtt_last_will_and_client_id_fix

Author: wsobel

CMakeLists.txt

@@ -2,7 +2,7 @@
 set(AGENT_VERSION_MAJOR 2)
 set(AGENT_VERSION_MINOR 2)
 set(AGENT_VERSION_PATCH 0)
-set(AGENT_VERSION_BUILD 15)
+set(AGENT_VERSION_BUILD 16)
 set(AGENT_VERSION_RC "")
 
 # This minimum version is to support Visual Studio 2019 and C++ feature checking and FetchContent

src/mtconnect/mqtt/mqtt_client_impl.hpp

@@ -21,6 +21,9 @@
 #include <boost/uuid/name_generator_sha1.hpp>
 
 #include <inttypes.h>
+#include <random>
+#include <chrono>
+
 #include <mqtt/async_client.hpp>
 #include <mqtt/setup_log.hpp>
 
@@ -92,17 +95,14 @@ namespace mtconnect {
         }
         else
         {
+          using namespace boost::uuids;
           std::stringstream identity;
-          identity << '_' << m_host << '_' << m_port;
-
-          boost::uuids::detail::sha1 sha1;
-          sha1.process_bytes(identity.str().c_str(), identity.str().length());
-          boost::uuids::detail::sha1::digest_type digest;
-          sha1.get_digest(digest);
+          const auto now = std::chrono::high_resolution_clock::now();
 
-          identity.str("");
-          identity << std::hex << digest[0] << digest[1] << digest[2];
-          m_identity = std::string("_") + (identity.str()).substr(0, 10);
+          auto seed = now.time_since_epoch().count();
+          std::mt19937_64 gen(seed);
+          identity << "mtc_" << std::hex << gen();
+          m_identity = identity.str();
         }
 
         LOG(debug) << "Using ClientID " << m_identity;

src/mtconnect/pipeline/json_mapper.cpp

@@ -47,6 +47,7 @@ namespace mtconnect::pipeline {
     DATA_SET,
     TIMESTAMP,
     ASSET,
+    VALUE_ERROR,
 
     VALUE,
     KEY,
@@ -174,6 +175,60 @@ 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 StartObject()
+    {
+      m_depth++;
+      return true;
+    }
+    bool EndObject(rj::SizeType memberCount)
+    {
+      m_depth--;
+      return true;
+    }
+    bool StartArray() 
+    {
+      m_depth++;
+      return true;
+    }
+    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;
+
+      while (m_depth > 0 && !reader.IterativeParseComplete())
+      {
+        // Read the key
+        if (!reader.IterativeParseNext<rj::kParseNanAndInfFlag>(buff, *this))
+          return false;
+      }
+      
+      return true;
+    }
+
+    int m_depth {0};
+  };
 
   /// @brief SAX Parser handler for JSON Parsing
   struct DataSetHandler : rj::BaseReaderHandler<rj::UTF8<>, DataSetHandler>
@@ -410,6 +465,7 @@ namespace mtconnect::pipeline {
 
     bool StartArray()
     {
+      m_depth++;
       if (m_dataItem->isTimeSeries() || m_dataItem->isThreeSpace())
       {
         auto &value = m_props["VALUE"];
@@ -420,11 +476,13 @@ namespace mtconnect::pipeline {
       else
       {
         LOG(warning) << "Unexpected vector type for data item " << m_dataItem->getId();
-        return false;
+        m_expectation = Expectation::VALUE_ERROR;
+        return true;
       }
     }
     bool EndArray(rj::SizeType elementCount)
     {
+      m_depth--;
       if (m_expectation == Expectation::VECTOR)
       {
         m_vector = nullptr;
@@ -434,7 +492,8 @@ namespace mtconnect::pipeline {
       else
       {
         LOG(warning) << "Unexpected vector type for data item " << m_dataItem->getId();
-        return false;
+        m_expectation = Expectation::VALUE_ERROR;
+        return true;
       }
     }
 
@@ -449,7 +508,8 @@ namespace mtconnect::pipeline {
         if (f == e)
         {
           LOG(warning) << "Unexpected key: " << sv << " for condition " << m_dataItem->getId();
-          return false;
+          m_expectation = Expectation::VALUE_ERROR;
+          return true;
         }
       }
       else
@@ -470,7 +530,8 @@ namespace mtconnect::pipeline {
         else
         {
           LOG(warning) << "Unexpected key " << sv << " for data item " << m_dataItem->getId();
-          return false;
+          m_expectation = Expectation::VALUE_ERROR;
+          return true;
         }
         m_key = sv;
       }
@@ -480,13 +541,15 @@ namespace mtconnect::pipeline {
 
     bool StartObject()
     {
+      m_depth++;
       m_object = true;
       m_expectation = Expectation::KEY;
       return true;
     }
 
     bool EndObject(rj::SizeType memberCount)
     {
+      m_depth--;
       m_done = true;
       return true;
     }
@@ -509,7 +572,15 @@ namespace mtconnect::pipeline {
 
         if (!m_done)
         {
-          if (m_expectation == Expectation::DATA_SET)
+          if (m_expectation == Expectation::VALUE_ERROR)
+          {
+            ErrorHandler handler(m_depth);
+            if (!handler(reader, buff))
+              return false;
+            m_props.clear();
+            return true;
+          }
+          else if (m_expectation == Expectation::DATA_SET)
           {
             auto &value = m_props["VALUE"];
             DataSet &set = value.emplace<DataSet>();
@@ -547,6 +618,7 @@ namespace mtconnect::pipeline {
     bool m_object {false};
     std::string m_key;
     Expectation m_expectation {Expectation::NONE};
+    int m_depth{0};
   };
 
   struct TimestampHandler : rj::BaseReaderHandler<rj::UTF8<>, TimestampHandler>
@@ -690,7 +762,8 @@ namespace mtconnect::pipeline {
     bool Default()
     {
       LOG(warning) << "Expecting a key";
-      return false;
+      m_expectation = Expectation::VALUE_ERROR;
+      return true;
     }
 
     bool Key(const Ch *str, rj::SizeType length, bool copy)
@@ -789,6 +862,16 @@ namespace mtconnect::pipeline {
               m_expectation = Expectation::KEY;
               break;
             }
+              
+            case Expectation::VALUE_ERROR:
+            {
+              ErrorHandler handler;
+              if (!handler(reader, buff))
+                return false;
+
+              m_expectation = Expectation::KEY;
+              break;
+            }
 
             case Expectation::KEY:
             case Expectation::NONE:
@@ -802,7 +885,8 @@ namespace mtconnect::pipeline {
                 PropertiesHandler props(m_dataItem, m_props);
                 if (!props(reader, buff))
                   return false;
-                m_context.send(m_dataItem, m_props);
+                if (m_props.size() > 0)
+                  m_context.send(m_dataItem, m_props);
               }
               else
               {

test_package/json_mapping_test.cpp

@@ -1005,5 +1005,50 @@ TEST_F(JsonMappingTest, should_parse_xml_asset)
   ASSERT_EQ("CuttingToolArchetype", asset->getName());
 }
 
+/// @test if  observation is incorrect, skip levels
+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}});
+  
+
+  Properties props {{"VALUE", R"(
+{
+  "timestamp": "2023-11-09T11:20:00Z",
+  "a": {
+      "r1": {
+        "k1": 123.45
+      },
+      "r2": {
+        "k2": "ABCDEF",
+        "k3": 6789
+      }
+    },
+   "b": "MANUAL"
+})"s}};
+  
+  auto jmsg = std::make_shared<JsonMessage>("JsonMessage", props);
+  jmsg->m_device = dev;
+
+  auto res = (*m_mapper)(std::move(jmsg));
+  ASSERT_TRUE(res);
+
+  auto value = res->getValue();
+  ASSERT_TRUE(std::holds_alternative<EntityList>(value));
+  auto list = get<EntityList>(value);
+  ASSERT_EQ(1, list.size());
+
+  auto obs = dynamic_pointer_cast<Observation>(list.front());
+  ASSERT_TRUE(obs);
+  ASSERT_EQ("ControllerMode", obs->getName());
+  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(); }