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-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org