Add RercordConverter struct

lordgamez · lordgamez · commit bf24a9e30d9b · 2025-10-20T12:10:06.000+02:00
diff --git a/extensions/mqtt/processors/AbstractMQTTProcessor.cpp b/extensions/mqtt/processors/AbstractMQTTProcessor.cpp
@@ -109,6 +109,20 @@ void AbstractMQTTProcessor::onSchedule(core::ProcessContext& context, core::Proc
   readProperties(context);
   checkProperties(context);
   initializeClient();
+
+  auto record_set_reader = utils::parseOptionalControllerService<core::RecordSetReader>(context, RecordReader, getUUID());
+  auto record_set_writer = utils::parseOptionalControllerService<core::RecordSetWriter>(context, RecordWriter, getUUID());
+
+  if ((record_set_reader == nullptr) != (record_set_writer == nullptr)) {
+    throw Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "ConsumeMQTT requires both or neither Record Reader and Record Writer to be set");
+  }
+
+  if (record_set_reader) {
+    record_converter_ = core::RecordConverter{
+      .record_set_reader = gsl::make_not_null(std::move(record_set_reader)),
+      .record_set_writer = gsl::make_not_null(std::move(record_set_writer)),
+    };
+  }
 }
 
 void AbstractMQTTProcessor::initializeClient() {
diff --git a/extensions/mqtt/processors/AbstractMQTTProcessor.h b/extensions/mqtt/processors/AbstractMQTTProcessor.h
@@ -23,6 +23,7 @@
 #include <vector>
 #include <shared_mutex>
 #include <future>
+#include <optional>
 
 #include "minifi-cpp/core/PropertyDefinition.h"
 #include "core/ProcessorImpl.h"
@@ -32,8 +33,7 @@
 #include "core/logging/LoggerFactory.h"
 #include "utils/Enum.h"
 #include "MQTTAsync.h"
-#include "controllers/RecordSetReader.h"
-#include "controllers/RecordSetWriter.h"
+#include "minifi-cpp/controllers/RecordConverter.h"
 
 namespace org::apache::nifi::minifi::processors::mqtt {
 enum class MqttVersions {
@@ -168,6 +168,14 @@ class AbstractMQTTProcessor : public core::ProcessorImpl {
       .withDescription("Private key passphrase")
       .isSensitive(true)
       .build();
+  EXTENSIONAPI static constexpr auto RecordReader = core::PropertyDefinitionBuilder<>::createProperty("Record Reader")
+      .withDescription("The Record Reader to use for parsing received MQTT Messages into Records.")
+      .withAllowedTypes<minifi::core::RecordSetReader>()
+      .build();
+  EXTENSIONAPI static constexpr auto RecordWriter = core::PropertyDefinitionBuilder<>::createProperty("Record Writer")
+      .withDescription("The Record Writer to use for serializing Records before writing them to a FlowFile.")
+      .withAllowedTypes<minifi::core::RecordSetWriter>()
+      .build();
   EXTENSIONAPI static constexpr auto BasicProperties = std::to_array<core::PropertyReference>({
       BrokerURI,
       ClientID,
@@ -188,7 +196,9 @@ class AbstractMQTTProcessor : public core::ProcessorImpl {
       SecurityCA,
       SecurityCert,
       SecurityPrivateKey,
-      SecurityPrivateKeyPassword
+      SecurityPrivateKeyPassword,
+      RecordReader,
+      RecordWriter
   });
 
   void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& factory) override;
@@ -258,8 +268,7 @@ class AbstractMQTTProcessor : public core::ProcessorImpl {
   std::optional<std::chrono::seconds> maximum_session_expiry_interval_;
   std::optional<std::chrono::seconds> server_keep_alive_;
 
-  std::shared_ptr<core::RecordSetReader> record_set_reader_;
-  std::shared_ptr<core::RecordSetWriter> record_set_writer_;
+  std::optional<core::RecordConverter> record_converter_;
 
  private:
   using ConnectFinishedTask = std::packaged_task<void(MQTTAsync_successData*, MQTTAsync_successData5*, MQTTAsync_failureData*, MQTTAsync_failureData5*)>;
diff --git a/extensions/mqtt/processors/ConsumeMQTT.cpp b/extensions/mqtt/processors/ConsumeMQTT.cpp
@@ -39,12 +39,7 @@ void ConsumeMQTT::initialize() {
 
 void ConsumeMQTT::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& factory) {
   AbstractMQTTProcessor::onSchedule(context, factory);
-  record_set_reader_ = utils::parseOptionalControllerService<core::RecordSetReader>(context, RecordReader, getUUID());
-  record_set_writer_ = utils::parseOptionalControllerService<core::RecordSetWriter>(context, RecordWriter, getUUID());
 
-  if ((record_set_reader_ == nullptr) != (record_set_writer_ == nullptr)) {
-    throw Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "ConsumeMQTT requires both or neither Record Reader and Record Writer to be set");
-  }
   add_attributes_as_fields_ = utils::parseBoolProperty(context, AddAttributesAsFields);
 }
 
@@ -89,13 +84,13 @@ void ConsumeMQTT::addAttributesAsRecordFields(core::RecordSet& new_records, cons
 }
 
 void ConsumeMQTT::transferMessagesAsRecords(core::ProcessSession& session) {
-  gsl_Expects(record_set_reader_ && record_set_writer_);
+  gsl_Expects(record_converter_);
   auto msg_queue = getReceivedMqttMessages();
   core::RecordSet record_set;
   while (!msg_queue.empty()) {
     io::BufferStream buffer_stream;
     buffer_stream.write(reinterpret_cast<const uint8_t*>(msg_queue.front().contents->payload), gsl::narrow<size_t>(msg_queue.front().contents->payloadlen));
-    auto new_records_result = record_set_reader_->read(buffer_stream);
+    auto new_records_result = record_converter_->record_set_reader->read(buffer_stream);
     if (!new_records_result) {
       logger_->log_error("Failed to read records from MQTT message: {}", new_records_result.error());
       msg_queue.pop();
@@ -112,7 +107,7 @@ void ConsumeMQTT::transferMessagesAsRecords(core::ProcessSession& session) {
     return;
   }
   std::shared_ptr<core::FlowFile> flow_file = session.create();
-  record_set_writer_->write(record_set, flow_file, session);
+  record_converter_->record_set_writer->write(record_set, flow_file, session);
   session.putAttribute(*flow_file, RecordCountOutputAttribute.name, std::to_string(record_set.size()));
   session.putAttribute(*flow_file, BrokerOutputAttribute.name, uri_);
   session.transfer(flow_file, Success);
@@ -152,7 +147,7 @@ void ConsumeMQTT::transferMessagesAsFlowFiles(core::ProcessSession& session) {
 }
 
 void ConsumeMQTT::onTriggerImpl(core::ProcessContext&, core::ProcessSession& session) {
-  if (record_set_reader_) {
+  if (record_converter_) {
     transferMessagesAsRecords(session);
   } else {
     transferMessagesAsFlowFiles(session);
diff --git a/extensions/mqtt/processors/ConsumeMQTT.h b/extensions/mqtt/processors/ConsumeMQTT.h
@@ -82,14 +82,6 @@ class ConsumeMQTT : public processors::AbstractMQTTProcessor {
       .withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR)
       .withDefaultValue(MQTT_MAX_RECEIVE_MAXIMUM_STR)
       .build();
-  EXTENSIONAPI static constexpr auto RecordReader = core::PropertyDefinitionBuilder<>::createProperty("Record Reader")
-      .withDescription("The Record Reader to use for parsing received MQTT Messages into Records.")
-      .withAllowedTypes<minifi::core::RecordSetReader>()
-      .build();
-  EXTENSIONAPI static constexpr auto RecordWriter = core::PropertyDefinitionBuilder<>::createProperty("Record Writer")
-      .withDescription("The Record Writer to use for serializing Records before writing them to a FlowFile.")
-      .withAllowedTypes<minifi::core::RecordSetWriter>()
-      .build();
   EXTENSIONAPI static constexpr auto AddAttributesAsFields = core::PropertyDefinitionBuilder<>::createProperty("Add Attributes As Fields")
       .withDescription("If setting this property to true, default fields are going to be added in each record: _topic, _qos, _isDuplicate, _isRetained.")
       .withValidator(core::StandardPropertyValidators::BOOLEAN_VALIDATOR)
@@ -104,8 +96,6 @@ class ConsumeMQTT : public processors::AbstractMQTTProcessor {
       AttributeFromContentType,
       TopicAliasMaximum,
       ReceiveMaximum,
-      RecordReader,
-      RecordWriter,
       AddAttributesAsFields
   }), AbstractMQTTProcessor::AdvancedProperties);
 
diff --git a/extensions/mqtt/processors/PublishMQTT.cpp b/extensions/mqtt/processors/PublishMQTT.cpp
@@ -44,16 +44,6 @@ void PublishMQTT::initialize() {
   setSupportedRelationships(Relationships);
 }
 
-void PublishMQTT::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& factory) {
-  AbstractMQTTProcessor::onSchedule(context, factory);
-  record_set_reader_ = utils::parseOptionalControllerService<core::RecordSetReader>(context, RecordReader, getUUID());
-  record_set_writer_ = utils::parseOptionalControllerService<core::RecordSetWriter>(context, RecordWriter, getUUID());
-
-  if ((record_set_reader_ == nullptr) != (record_set_writer_ == nullptr)) {
-    throw Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "PublishMQTT requires both or neither Record Reader and Record Writer to be set");
-  }
-}
-
 void PublishMQTT::readProperties(core::ProcessContext& context) {
   if (!context.getProperty(Topic).has_value()) {
     throw Exception(PROCESS_SCHEDULE_EXCEPTION, "PublishMQTT: Topic is required");
@@ -74,10 +64,10 @@ void PublishMQTT::onTriggerImpl(core::ProcessContext& context, core::ProcessSess
   }
 
   std::vector<std::shared_ptr<core::FlowFile>> flow_files;
-  if (record_set_reader_) {
+  if (record_converter_) {
     nonstd::expected<core::RecordSet, std::error_code> record_set;
     session.read(original_flow_file, [this, &record_set](const std::shared_ptr<io::InputStream>& input_stream) {
-      record_set = record_set_reader_->read(*input_stream);
+      record_set = record_converter_->record_set_reader->read(*input_stream);
       return gsl::narrow<int64_t>(input_stream->size());
     });
 
@@ -95,7 +85,7 @@ void PublishMQTT::onTriggerImpl(core::ProcessContext& context, core::ProcessSess
       }
       std::vector<core::Record> records;
       records.emplace_back(std::move(record));
-      record_set_writer_->write(records, new_flow_file, session);
+      record_converter_->record_set_writer->write(records, new_flow_file, session);
       flow_files.push_back(std::move(new_flow_file));
     }
 
diff --git a/extensions/mqtt/processors/PublishMQTT.h b/extensions/mqtt/processors/PublishMQTT.h
@@ -64,21 +64,11 @@ class PublishMQTT : public processors::AbstractMQTTProcessor {
       .withDescription("Content type of the message. MQTT 5.x only.")
       .supportsExpressionLanguage(true)
       .build();
-  EXTENSIONAPI static constexpr auto RecordReader = core::PropertyDefinitionBuilder<>::createProperty("Record Reader")
-      .withDescription("The Record Reader to use for parsing the incoming FlowFile into Records.")
-      .withAllowedTypes<minifi::core::RecordSetReader>()
-      .build();
-  EXTENSIONAPI static constexpr auto RecordWriter = core::PropertyDefinitionBuilder<>::createProperty("Record Writer")
-      .withDescription("The Record Writer to use for serializing Records before publishing them as an MQTT Message.")
-      .withAllowedTypes<minifi::core::RecordSetWriter>()
-      .build();
   EXTENSIONAPI static constexpr auto Properties = utils::array_cat(AbstractMQTTProcessor::BasicProperties, std::to_array<core::PropertyReference>({
       Topic,
       Retain,
       MessageExpiryInterval,
-      ContentType,
-      RecordReader,
-      RecordWriter
+      ContentType
   }), AbstractMQTTProcessor::AdvancedProperties);
 
   EXTENSIONAPI static constexpr auto Success = core::RelationshipDefinition{"success", "FlowFiles that are sent successfully to the destination are transferred to this relationship"};
@@ -92,7 +82,6 @@ class PublishMQTT : public processors::AbstractMQTTProcessor {
 
   ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
 
-  void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& factory) override;
   void readProperties(core::ProcessContext& context) override;
   void onTriggerImpl(core::ProcessContext& context, core::ProcessSession& session) override;
   void initialize() override;
diff --git a/extensions/standard-processors/processors/ConvertRecord.cpp b/extensions/standard-processors/processors/ConvertRecord.cpp
@@ -20,17 +20,20 @@
 #include "nonstd/expected.hpp"
 #include "utils/GeneralUtils.h"
 #include "utils/ProcessorConfigUtils.h"
+#include "minifi-cpp/utils/gsl.h"
 
 namespace org::apache::nifi::minifi::processors {
 
 void ConvertRecord::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
-  record_set_reader_ = utils::parseControllerService<core::RecordSetReader>(context, RecordReader, getUUID());
-  record_set_writer_ = utils::parseControllerService<core::RecordSetWriter>(context, RecordWriter, getUUID());
+  record_converter_ = core::RecordConverter{
+    .record_set_reader = utils::parseControllerService<core::RecordSetReader>(context, RecordReader, getUUID()),
+    .record_set_writer = utils::parseControllerService<core::RecordSetWriter>(context, RecordWriter, getUUID())
+  };
   include_zero_record_flow_files_ = utils::parseBoolProperty(context, IncludeZeroRecordFlowFiles);
 }
 
 void ConvertRecord::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
-  gsl_Expects(record_set_reader_ && record_set_writer_);
+  gsl_Expects(record_converter_);
   const auto flow_file = session.get();
   if (!flow_file) {
     context.yield();
@@ -39,7 +42,7 @@ void ConvertRecord::onTrigger(core::ProcessContext& context, core::ProcessSessio
 
   nonstd::expected<core::RecordSet, std::error_code> record_set;
   session.read(flow_file, [this, &record_set](const std::shared_ptr<io::InputStream>& input_stream) {
-    record_set = record_set_reader_->read(*input_stream);
+    record_set = record_converter_->record_set_reader->read(*input_stream);
     return gsl::narrow<int64_t>(input_stream->size());
   });
   if (!record_set) {
@@ -55,7 +58,7 @@ void ConvertRecord::onTrigger(core::ProcessContext& context, core::ProcessSessio
     return;
   }
 
-  record_set_writer_->write(*record_set, flow_file, session);
+  record_converter_->record_set_writer->write(*record_set, flow_file, session);
   flow_file->setAttribute(processors::ConvertRecord::RecordCountOutputAttribute.name, std::to_string(record_set->size()));
   session.transfer(flow_file, Success);
 }
diff --git a/extensions/standard-processors/processors/ConvertRecord.h b/extensions/standard-processors/processors/ConvertRecord.h
@@ -19,13 +19,13 @@
 #include <memory>
 #include <string_view>
 #include <utility>
+#include <optional>
 
 #include "core/AbstractProcessor.h"
 #include "core/ProcessSession.h"
 #include "core/PropertyDefinitionBuilder.h"
 #include "minifi-cpp/core/RelationshipDefinition.h"
-#include "controllers/RecordSetReader.h"
-#include "controllers/RecordSetWriter.h"
+#include "minifi-cpp/controllers/RecordConverter.h"
 
 namespace org::apache::nifi::minifi::processors {
 
@@ -77,8 +77,7 @@ class ConvertRecord : public core::AbstractProcessor<ConvertRecord> {
   void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override;
 
  private:
-  std::shared_ptr<core::RecordSetReader> record_set_reader_;
-  std::shared_ptr<core::RecordSetWriter> record_set_writer_;
+  std::optional<core::RecordConverter> record_converter_;
   bool include_zero_record_flow_files_ = true;
 };
 
diff --git a/extensions/standard-processors/processors/SplitRecord.cpp b/extensions/standard-processors/processors/SplitRecord.cpp
@@ -20,12 +20,15 @@
 #include "nonstd/expected.hpp"
 #include "utils/GeneralUtils.h"
 #include "utils/ProcessorConfigUtils.h"
+#include "minifi-cpp/utils/gsl.h"
 
 namespace org::apache::nifi::minifi::processors {
 
 void SplitRecord::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
-  record_set_reader_ = utils::parseControllerService<core::RecordSetReader>(context, RecordReader, getUUID());
-  record_set_writer_ = utils::parseControllerService<core::RecordSetWriter>(context, RecordWriter, getUUID());
+  record_converter_ = core::RecordConverter{
+    .record_set_reader = utils::parseControllerService<core::RecordSetReader>(context, RecordReader, getUUID()),
+    .record_set_writer = utils::parseControllerService<core::RecordSetWriter>(context, RecordWriter, getUUID())
+  };
 }
 
 nonstd::expected<std::size_t, std::string> SplitRecord::readRecordsPerSplit(core::ProcessContext& context, const core::FlowFile& original_flow_file) {
@@ -37,6 +40,7 @@ nonstd::expected<std::size_t, std::string> SplitRecord::readRecordsPerSplit(core
 }
 
 void SplitRecord::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
+  gsl_Expects(record_converter_);
   const auto original_flow_file = session.get();
   if (!original_flow_file) {
     context.yield();
@@ -52,7 +56,7 @@ void SplitRecord::onTrigger(core::ProcessContext& context, core::ProcessSession&
 
   nonstd::expected<core::RecordSet, std::error_code> record_set;
   session.read(original_flow_file, [this, &record_set](const std::shared_ptr<io::InputStream>& input_stream) {
-    record_set = record_set_reader_->read(*input_stream);
+    record_set = record_converter_->record_set_reader->read(*input_stream);
     return gsl::narrow<int64_t>(input_stream->size());
   });
   if (!record_set) {
@@ -85,7 +89,7 @@ void SplitRecord::onTrigger(core::ProcessContext& context, core::ProcessSession&
     split_flow_file->setAttribute("fragment.count", std::to_string(fragment_count));
     split_flow_file->setAttribute("segment.original.filename", original_flow_file->getAttribute("filename").value_or(""));
 
-    record_set_writer_->write(slice_record_set, split_flow_file, session);
+    record_converter_->record_set_writer->write(slice_record_set, split_flow_file, session);
     session.transfer(split_flow_file, Splits);
     ++fragment_index;
   }
diff --git a/extensions/standard-processors/processors/SplitRecord.h b/extensions/standard-processors/processors/SplitRecord.h
@@ -16,6 +16,8 @@
  */
 #pragma once
 
+#include <optional>
+
 #include "minifi-cpp/core/Annotation.h"
 #include "minifi-cpp/core/ProcessContext.h"
 #include "core/ProcessSession.h"
@@ -24,8 +26,7 @@
 #include "core/PropertyDefinitionBuilder.h"
 #include "minifi-cpp/core/RelationshipDefinition.h"
 #include "minifi-cpp/core/logging/Logger.h"
-#include "minifi-cpp/controllers/RecordSetReader.h"
-#include "minifi-cpp/controllers/RecordSetWriter.h"
+#include "minifi-cpp/controllers/RecordConverter.h"
 #include "core/AbstractProcessor.h"
 
 namespace org::apache::nifi::minifi::processors {
@@ -87,8 +88,7 @@ class SplitRecord final : public core::AbstractProcessor<SplitRecord> {
  private:
   static nonstd::expected<std::size_t, std::string> readRecordsPerSplit(core::ProcessContext& context, const core::FlowFile& original_flow_file);
 
-  std::shared_ptr<core::RecordSetReader> record_set_reader_;
-  std::shared_ptr<core::RecordSetWriter> record_set_writer_;
+  std::optional<core::RecordConverter> record_converter_;
 };
 
 }  // namespace org::apache::nifi::minifi::processors
diff --git a/minifi-api/include/minifi-cpp/controllers/RecordConverter.h b/minifi-api/include/minifi-cpp/controllers/RecordConverter.h
@@ -0,0 +1,32 @@
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <memory>
+
+#include "RecordSetReader.h"
+#include "RecordSetWriter.h"
+#include "minifi-cpp/utils/gsl.h"
+
+namespace org::apache::nifi::minifi::core {
+
+struct RecordConverter {
+  gsl::not_null<std::shared_ptr<core::RecordSetReader>> record_set_reader;
+  gsl::not_null<std::shared_ptr<core::RecordSetWriter>> record_set_writer;
+};
+
+}  // namespace org::apache::nifi::minifi::core
