fgerlits commented on code in PR #1432:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1432#discussion_r1081545367
##########
extensions/mqtt/processors/PublishMQTT.cpp:
##########
@@ -29,81 +29,240 @@
namespace org::apache::nifi::minifi::processors {
+using SendFinishedTask = std::packaged_task<bool(bool, std::optional<int>,
std::optional<MQTTReasonCodes>)>;
+
void PublishMQTT::initialize() {
setSupportedProperties(properties());
setSupportedRelationships(relationships());
}
-void PublishMQTT::onSchedule(const std::shared_ptr<core::ProcessContext>
&context, const std::shared_ptr<core::ProcessSessionFactory> &factory) {
+void PublishMQTT::readProperties(const std::shared_ptr<core::ProcessContext>&
context) {
+ if (!context->getProperty(Topic).has_value()) {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "PublishMQTT: Topic is
required");
+ }
+
if (const auto retain_opt = context->getProperty<bool>(Retain)) {
retain_ = *retain_opt;
}
logger_->log_debug("PublishMQTT: Retain [%d]", retain_);
- AbstractMQTTProcessor::onSchedule(context, factory);
+ if (const auto message_expiry_interval =
context->getProperty<core::TimePeriodValue>(MessageExpiryInterval)) {
+ message_expiry_interval_ =
std::chrono::duration_cast<std::chrono::seconds>(message_expiry_interval->getMilliseconds());
+ logger_->log_debug("PublishMQTT: MessageExpiryInterval [%" PRId64 "] s",
int64_t{message_expiry_interval_->count()});
+ }
+
+ in_flight_message_counter_.setMqttVersion(mqtt_version_);
+ in_flight_message_counter_.setQoS(qos_);
}
-void PublishMQTT::onTrigger(const std::shared_ptr<core::ProcessContext>&
/*context*/, const std::shared_ptr<core::ProcessSession> &session) {
- // reconnect if needed
- reconnect();
+void PublishMQTT::onTriggerImpl(const std::shared_ptr<core::ProcessContext>&
context, const std::shared_ptr<core::ProcessSession>& session) {
+ std::shared_ptr<core::FlowFile> flow_file = session->get();
- if (!MQTTAsync_isConnected(client_)) {
- logger_->log_error("Could not publish to MQTT broker because disconnected
to %s", uri_);
+ if (!flow_file) {
yield();
return;
}
- std::shared_ptr<core::FlowFile> flowFile = session->get();
+ // broker's Receive Maximum can change after reconnect
+
in_flight_message_counter_.setMax(broker_receive_maximum_.value_or(MQTT_MAX_RECEIVE_MAXIMUM));
- if (!flowFile) {
- return;
+ const auto topic = getTopic(context, flow_file);
+ try {
+ const auto result = session->readBuffer(flow_file);
+ if (result.status < 0 || !sendMessage(result.buffer, topic,
getContentType(context, flow_file), flow_file)) {
+ logger_->log_error("Failed to send flow file [%s] to MQTT topic '%s' on
broker %s", flow_file->getUUIDStr(), topic, uri_);
+ session->transfer(flow_file, Failure);
+ return;
+ }
+ logger_->log_debug("Sent flow file [%s] with length %" PRId64 " to MQTT
topic '%s' on broker %s", flow_file->getUUIDStr(), result.status, topic, uri_);
+ session->transfer(flow_file, Success);
+ } catch (const Exception& ex) {
+ logger_->log_error("Failed to send flow file [%s] to MQTT topic '%s' on
broker %s, exception string: '%s'", flow_file->getUUIDStr(), topic, uri_,
ex.what());
+ session->transfer(flow_file, Failure);
+ }
+}
+
+bool PublishMQTT::sendMessage(const std::vector<std::byte>& buffer, const
std::string& topic, const std::string& content_type, const
std::shared_ptr<core::FlowFile>& flow_file) {
+ static const unsigned max_packet_size = 268'435'455;
Review Comment:
fixed in 0e2d673a4f4bfeab62a5fee6659f47fcbb06aed6
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
