Review update

Author: lordgamez

extensions/mqtt/processors/ConsumeMQTT.cpp

@@ -69,22 +69,22 @@ void ConsumeMQTT::readProperties(core::ProcessContext& context) {
   receive_maximum_ = gsl::narrow<uint16_t>(utils::parseU64Property(context, ReceiveMaximum));
 }
 
-void ConsumeMQTT::addAttributesAsRecordFields(core::RecordSet& new_records, const std::queue<SmartMessage>& msg_queue) const {
+void ConsumeMQTT::addAttributesAsRecordFields(core::RecordSet& new_records, const SmartMessage& message) const {
   if (!add_attributes_as_fields_) {
     return;
   }
 
-  for (auto& record: new_records) {
-    record.emplace("_topic", core::RecordField(msg_queue.front().topic));
-    auto topic_segments = utils::string::split(msg_queue.front().topic, "/");
+  for (auto& record : new_records) {
+    record.emplace("_topic", core::RecordField(message.topic));
+    auto topic_segments = utils::string::split(message.topic, "/");
     core::RecordArray topic_segments_array;
     for (size_t i = 0; i < topic_segments.size(); ++i) {
       topic_segments_array.emplace_back(core::RecordField(topic_segments[i]));
     }
     record.emplace("_topicSegments", core::RecordField(std::move(topic_segments_array)));
-    record.emplace("_qos", core::RecordField(msg_queue.front().contents->qos));
-    record.emplace("_isDuplicate", core::RecordField(msg_queue.front().contents->dup > 0));
-    record.emplace("_isRetained", core::RecordField(msg_queue.front().contents->retained > 0));
+    record.emplace("_qos", core::RecordField(message.contents->qos));
+    record.emplace("_isDuplicate", core::RecordField(message.contents->dup > 0));
+    record.emplace("_isRetained", core::RecordField(message.contents->retained > 0));
   }
 }
 
@@ -102,7 +102,7 @@ void ConsumeMQTT::transferMessagesAsRecords(core::ProcessSession& session) {
       continue;
     }
     auto& new_records = new_records_result.value();
-    addAttributesAsRecordFields(new_records, msg_queue);
+    addAttributesAsRecordFields(new_records, msg_queue.front());
     record_set.reserve(record_set.size() + new_records.size());
     record_set.insert(record_set.end(), std::make_move_iterator(new_records.begin()), std::make_move_iterator(new_records.end()));
     msg_queue.pop();

extensions/mqtt/processors/ConsumeMQTT.h

@@ -221,7 +221,7 @@ class ConsumeMQTT : public processors::AbstractMQTTProcessor {
   void setProcessorSpecificMqtt5ConnectOptions(MQTTProperties& connect_props) const override;
 
   void transferMessagesAsRecords(core::ProcessSession& session);
-  void addAttributesAsRecordFields(core::RecordSet& new_records, const std::queue<SmartMessage>& msg_queue) const;
+  void addAttributesAsRecordFields(core::RecordSet& new_records, const SmartMessage& message) const;
   void transferMessagesAsFlowFiles(core::ProcessSession& session);
 
   std::string topic_;

extensions/mqtt/tests/ConsumeMQTTTests.cpp

@@ -42,19 +42,29 @@ void verifyXmlJsonResult(const std::string& json_content, size_t expected_record
 
     if (add_attributes_as_fields) {
       string_result = current_record["_topic"].GetString();
-      CHECK(string_result == "mytopic/segment");
+      CHECK(string_result == "mytopic/segment/" + std::to_string(i));
       auto array = current_record["_topicSegments"].GetArray();
-      CHECK(array.Size() == 2);
+      CHECK(array.Size() == 3);
       string_result = array[0].GetString();
       CHECK(string_result == "mytopic");
       string_result = array[1].GetString();
       CHECK(string_result == "segment");
+      string_result = array[2].GetString();
+      CHECK(string_result == std::to_string(i));
       int_result = current_record["_qos"].GetInt64();
-      CHECK(int_result == 1);
+      CHECK(int_result == i);
       bool bool_result = current_record["_isDuplicate"].GetBool();
-      CHECK_FALSE(bool_result);
+      if (i == 0) {
+        CHECK_FALSE(bool_result);
+      } else {
+        CHECK(bool_result);
+      }
       bool_result = current_record["_isRetained"].GetBool();
-      CHECK_FALSE(bool_result);
+      if (i == 0) {
+        CHECK_FALSE(bool_result);
+      } else {
+        CHECK(bool_result);
+      }
     } else {
       CHECK_FALSE(current_record.HasMember("_topic"));
       CHECK_FALSE(current_record.HasMember("_qos"));
@@ -268,9 +278,10 @@ TEST_CASE_METHOD(ConsumeMqttTestFixture, "Read XML messages and write them to js
   const std::string payload = R"(<root><int_value>42</int_value><string_value>test</string_value></root>)";
   for (size_t i = 0; i < expected_record_count; ++i) {
     TestConsumeMQTTProcessor::SmartMessage message{std::unique_ptr<MQTTAsync_message, TestConsumeMQTTProcessor::MQTTMessageDeleter>(
-        new MQTTAsync_message{.struct_id = {'M', 'Q', 'T', 'M'}, .struct_version = 1, .payloadlen = gsl::narrow<int>(payload.size()),
-                              .payload = const_cast<char*>(payload.data()), .qos = 1, .retained = 0, .dup = 0, .msgid = 42, .properties = {}}),
-      std::string{"mytopic/segment"}};
+        new MQTTAsync_message{.struct_id = {'M', 'Q', 'T', 'M'}, .struct_version = gsl::narrow<int>(i), .payloadlen = gsl::narrow<int>(payload.size()),
+                              .payload = const_cast<char*>(payload.data()), .qos = gsl::narrow<int>(i), .retained = gsl::narrow<int>(i), .dup = gsl::narrow<int>(i),
+                              .msgid = gsl::narrow<int>(i + 1), .properties = {}}),
+      std::string{"mytopic/segment/" + std::to_string(i)}};
     consume_mqtt_processor_->enqueueReceivedMQTTMsg(std::move(message));
   }
   const auto trigger_results = test_controller_.trigger();
@@ -359,7 +370,7 @@ TEST_CASE_METHOD(ConsumeMqttTestFixture, "Read MQTT message and write it to a fl
   }
 }
 
-TEST_CASE_METHOD(ConsumeMqttTestFixture, "Test scheduling failure if non-existant recordset reader or writer is set", "[consumeMQTTTest]") {
+TEST_CASE_METHOD(ConsumeMqttTestFixture, "Test scheduling failure if non-existent recordset reader or writer is set", "[consumeMQTTTest]") {
   test_controller_.plan->addController("XMLReader", "XMLReader");
   test_controller_.plan->addController("JsonRecordSetWriter", "JsonRecordSetWriter");
   REQUIRE(consume_mqtt_processor_->setProperty(minifi::processors::AbstractMQTTProcessor::BrokerURI.name, "127.0.0.1:1883"));

extensions/mqtt/tests/PublishMQTTTests.cpp

@@ -148,7 +148,7 @@ TEST_CASE_METHOD(PublishMQTTTestFixture, "Test sending XML message records", "[p
   CHECK(string_content == R"(<?xml version="1.0"?><root><record><element2>42</element2></record></root>)");
 }
 
-TEST_CASE_METHOD(PublishMQTTTestFixture, "Test scheduling failure if non-existant recordset reader or writer is set", "[publishMQTTTest]") {
+TEST_CASE_METHOD(PublishMQTTTestFixture, "Test scheduling failure if non-existent recordset reader or writer is set", "[publishMQTTTest]") {
   test_controller_.plan->addController("XMLReader", "XMLReader");
   test_controller_.plan->addController("JsonRecordSetWriter", "JsonRecordSetWriter");
   REQUIRE(publish_mqtt_processor_->setProperty(minifi::processors::AbstractMQTTProcessor::BrokerURI.name, "127.0.0.1:1883"));