adam-markovics commented on code in PR #1363:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1363#discussion_r934650630
##########
extensions/mqtt/processors/ConsumeMQTT.cpp:
##########
@@ -37,48 +34,64 @@ void ConsumeMQTT::initialize() {
setSupportedRelationships(relationships());
}
-bool ConsumeMQTT::enqueueReceiveMQTTMsg(MQTTClient_message *message) {
+void ConsumeMQTT::enqueueReceivedMQTTMsg(std::unique_ptr<MQTTAsync_message,
MQTTMessageDeleter> message) {
if (queue_.size_approx() >= maxQueueSize_) {
logger_->log_warn("MQTT queue full");
- return false;
- } else {
- if (gsl::narrow<uint64_t>(message->payloadlen) > maxSegSize_)
- message->payloadlen = maxSegSize_;
- queue_.enqueue(message);
- logger_->log_debug("enqueue MQTT message length %d", message->payloadlen);
- return true;
+ return;
+ }
+
+ if (gsl::narrow<uint64_t>(message->payloadlen) > max_seg_size_) {
+ logger_->log_debug("MQTT message was truncated while enqueuing, original
length: %d", message->payloadlen);
+ message->payloadlen = gsl::narrow<int>(max_seg_size_);
}
+
+ logger_->log_debug("enqueuing MQTT message with length %d",
message->payloadlen);
+ queue_.enqueue(std::move(message));
}
void ConsumeMQTT::onSchedule(const std::shared_ptr<core::ProcessContext>
&context, const std::shared_ptr<core::ProcessSessionFactory> &factory) {
- AbstractMQTTProcessor::onSchedule(context, factory);
std::string value;
int64_t valInt;
value = "";
+ if (context->getProperty(CleanSession.getName(), value)) {
+ cleanSession_ = utils::StringUtils::toBool(value).value_or(cleanSession_);
+ logger_->log_debug("ConsumeMQTT: CleanSession [%d]", cleanSession_);
+ }
+ value = "";
if (context->getProperty(QueueBufferMaxMessage.getName(), value) &&
!value.empty() && core::Property::StringToInt(value, valInt)) {
maxQueueSize_ = valInt;
logger_->log_debug("ConsumeMQTT: Queue Max Message [%" PRIu64 "]",
maxQueueSize_);
}
- value = "";
- if (context->getProperty(MaxFlowSegSize.getName(), value) && !value.empty()
&& core::Property::StringToInt(value, valInt)) {
- maxSegSize_ = valInt;
- logger_->log_debug("ConsumeMQTT: Max Flow Segment Size [%" PRIu64 "]",
maxSegSize_);
+
+ AbstractMQTTProcessor::onSchedule(context, factory);
Review Comment:
Done. It could only be made with a virtual function due to
`ConsumeMQTT::cleanSession_`.
--
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]
