turcsanyip commented on code in PR #11780:
URL: https://github.com/apache/nifi/pull/11780#discussion_r4228220763
##########
nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java:
##########
@@ -128,32 +127,33 @@ public void publish(String topic, StandardMqttMessage
message) {
}
@Override
- public void subscribe(String topicFilter, int qos,
ReceivedMqttMessageHandler handler) {
+ public void subscribe(String topicFilter, int qos) {
logger.debug("Subscribing to {} with QoS: {}", topicFilter, qos);
- CompletableFuture<Mqtt5SubAck> futureAck =
mqtt5BlockingClient.toAsync().subscribeWith()
- .topicFilter(topicFilter)
- .qos(Objects.requireNonNull(MqttQos.fromCode(qos)))
- .callback(mqtt5Publish -> {
- final ReceivedMqttMessage receivedMessage = new
ReceivedMqttMessage(
- mqtt5Publish.getPayloadAsBytes(),
- mqtt5Publish.getQos().getCode(),
- mqtt5Publish.isRetain(),
- mqtt5Publish.getTopic().toString());
- handler.handleReceivedMessage(receivedMessage);
- })
- .send();
-
- // Setting "listener" callback is only possible with async client,
though sending subscribe message
- // should happen in a blocking way to make sure the processor is
blocked until ack is not arrived.
try {
- final Mqtt5SubAck ack =
futureAck.get(clientProperties.getConnectionTimeout(), TimeUnit.SECONDS);
+ Mqtt5SubAck ack = mqtt5BlockingClient.subscribeWith()
Review Comment:
Makes sense. Will revert the change.
[connect()](https://github.com/apache/nifi/blob/702e6d29f94e57c4d0da2c614265856573e6820c/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java#L103)
and
[publish()](https://github.com/apache/nifi/blob/702e6d29f94e57c4d0da2c614265856573e6820c/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java#L122)
calls also use the blocking client and those should be moved to async with
timeout as well. Will create a follow-up ticket for it.
--
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]
