Lehel44 commented on code in PR #7032:
URL: https://github.com/apache/nifi/pull/7032#discussion_r1139467597
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java:
##########
@@ -189,4 +172,53 @@ private static org.eclipse.paho.client.mqttv3.MqttClient
createClient(URI broker
}
}
+ /**
+ * Paho API uses the same callback for the publisher and consumer as well.
+ * Because of that, DefaultMqttCallback sets some reasonable default logs
+ * to make it easier to track misconfiguration errors.
+ * <p>
+ * In case of subscribing clients messageArrived needs to be overridden.
+ */
+ private class DefaultMqttCallback implements MqttCallback {
+
+ @Override
+ public void connectionLost(Throwable cause) {
+ logger.error("Connection to {} lost",
clientProperties.getRawBrokerUris(), cause);
+ }
+
+ @Override
+ public void messageArrived(String topic, MqttMessage message) {
+ logger.error(String.format("MQTT message arrived { topic:%s;
payload:%s}", topic, Arrays.toString(message.getPayload())));
Review Comment:
I think this is supposed to be a debug log.
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClient.java:
##########
@@ -57,13 +57,7 @@ public interface MqttClient {
* published at a lower quality of service will be received at
the published
* QoS. Messages published at a higher quality of service will
be received using
* the QoS specified on the subscribe.
+ * @param handler that further processes message received by the client
Review Comment:
typo: messages or the message
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java:
##########
@@ -383,9 +383,8 @@ private void initializeClient(ProcessContext context) {
// non-null but not connected, so we need to handle each case and only
create a new client when it is null
try {
mqttClient = createMqttClient();
- mqttClient.setCallback(this);
mqttClient.connect();
- mqttClient.subscribe(topicPrefix + topicFilter, qos);
+ mqttClient.subscribe(topicPrefix + topicFilter, qos, this);
Review Comment:
It's uneasy to follow the code here. A processor class being a callback
itself is strange. What do you think of implementing the interface as an
anonymous class instead of passing 'this'?
--
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]
