[GitHub] [nifi] Lehel44 commented on a diff in pull request #7032: NIFI-11270 Refactoring of the overly Paho-specific MQTT interface

["Lehel44 (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org