pvillard31 commented on code in PR #11780:
URL: https://github.com/apache/nifi/pull/11780#discussion_r4216739587
##########
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:
Could we retain the bounded async subscribe wait, since blocking `send()`
has no deadline and a missing SUBACK can block both processing and shutdown?
##########
nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/MqttTestClient.java:
##########
@@ -73,9 +73,13 @@ public void publish(String topic, StandardMqttMessage
message) {
}
@Override
- public void subscribe(String topicFilter, int qos,
ReceivedMqttMessageHandler handler) {
+ public void subscribe(String topicFilter, int qos) {
subscribedTopic = topicFilter;
subscribedQos = qos;
+ }
+
+ @Override
+ public void setReceivedMessageHandler(ReceivedMqttMessageHandler handler) {
Review Comment:
Could we strengthen the existing persistent-session tests to deliver queued
messages during `connect()`, so they fail if handler registration moves after
connection?
--
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]
