[mqtt] Correctly unsubscribe after message is received #7723

bodiroga · 2020-05-21T19:54:59Z

The future attribute was overwritten in the constructor and, as a consequence, the whenComplete lambda method was not called. This made it impossible to re-subscribe to the same topic from the same connection. This fix is critical for Homie users, as the extensions uses this function in the discovery process.

A couple of typos are also fixed.

Closes #6975
Closes #7252

Reviewers: @jochen314, @J-N-K, @cpmeister

Signed-off-by: Aitor Iturrioz riturrioz@gmail.com

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

cpmeister

That bug is amazing.

cpmeister · 2020-05-21T20:47:35Z

...ing.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/tools/WaitForTopicValue.java

@@ -46,11 +46,11 @@ public WaitForTopicValue(MqttBrokerConnection connection, String topic)
 final MqttMessageSubscriber mqttMessageSubscriber = (t, payload) -> {
 future.complete(new String(payload));


The default charset varies from system to system so it is better to specify it yourself instead of using the default.

cpmeister · 2020-05-21T20:54:01Z

...ing.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/tools/WaitForTopicValue.java

@@ -24,14 +24,14 @@
 import org.eclipse.smarthome.io.transport.mqtt.MqttMessageSubscriber;

 /**
- * Waits for a topic value to appear on a MQTT topic. One-time useable only per instance.
+ * Waits for a topic value to appear on a MQTT topic. One-time usable only per instance.
 *
 * @author David Graeff - Initial contribution
 */
 @NonNullByDefault
 public class WaitForTopicValue {
 private CompletableFuture<String> future = new CompletableFuture<>();


Suggested change

private CompletableFuture<String> future = new CompletableFuture<>();

private final CompletableFuture<String> future = new CompletableFuture<>();

cpmeister · 2020-05-21T20:55:03Z

...ing.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/tools/WaitForTopicValue.java

@@ -68,7 +68,7 @@ public void stop() {
 */
 public @Nullable String waitForTopicValue(int timeoutInMS) {
 try {
- return subscripeFuture.thenCompose(b -> future).get(timeoutInMS, TimeUnit.MILLISECONDS);
+ return subscribeFuture.thenCompose(b -> future).get(timeoutInMS, TimeUnit.MILLISECONDS);


seems a bit wasteful to create a new compose future every time. Could you cache subscribeFuture.thenCompose(b -> future) in a field?

Ok, I will modify it, but keep in mind that the WaitForTopicValue is a one-time use class and a new object should be created in each use. What would be the benefit?

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

bodiroga · 2020-05-21T22:02:34Z

Done!

cpmeister · 2020-05-22T03:18:07Z

...ing.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/tools/WaitForTopicValue.java

+ subscribeFuture = connection.subscribe(topic, mqttMessageSubscriber);
+
+ composeFuture = subscribeFuture.thenCompose(b -> future);


You probably don't need the subscribeFuture field anymore.

Suggested change

subscribeFuture = connection.subscribe(topic, mqttMessageSubscriber);

composeFuture = subscribeFuture.thenCompose(b -> future);

composeFuture = connection.subscribe(topic, mqttMessageSubscriber).thenCompose(b -> future);

cpmeister · 2020-05-22T03:30:49Z

...ing.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/tools/WaitForTopicValue.java

 *
 * @author David Graeff - Initial contribution
 */
 @NonNullByDefault
 public class WaitForTopicValue {
- private CompletableFuture<String> future = new CompletableFuture<>();
- private final CompletableFuture<Boolean> subscripeFuture;
+ private final CompletableFuture<String> future = new CompletableFuture<>();


I suspect that you should replace usages of future in this class with composeFuture.
Your timeout method completes the future but the result of that future isn't available until the subscribeFuture completes first so your timeout wouldn't actually cause the timeout as you would expect.

However if you caused the timeout on the composeFuture everything would operate as you would expect.

Thanks @cpmeister, although we're going to change more lines than the class itself has 😄 Anyway, it seems that the Homie discovery process has another unnoticed error. The Homie300Discovery class does the following:

try { WaitForTopicValue w = new WaitForTopicValue(connection, topic.replace("$homie", "$name")); w.waitForTopicValueAsync(scheduler, 700).thenAccept(name -> { publishDevice(connectionBridge, connection, deviceID, topic, name); }); } catch (InterruptedException | ExecutionException ignored) { // The name is nice to have, but not required, use deviceId as fallback publishDevice(connectionBridge, connection, deviceID, topic, deviceID); }

But, if I intentionally remove the $name message and thus, trigger the timeout, the composeFuture.completeExceptionally function doesn't throw any error, so no discoveryResult is published. I'm sorry, but do you know how can I receive the completeExceptionally message in the w.waitForTopicValueAsync() sequence?

This CompleteFuture topic is driving me crazy, I'm more used to synchronous programming 😞

Would this work?

WaitForTopicValue w = new WaitForTopicValue(connection, topic.replace("$homie", "$name")); w.waitForTopicValueAsync(scheduler, 700).whenComplete((name, ex) -> { String deviceName = ex == null ? deviceID : name; publishDevice(connectionBridge, connection, deviceID, topic, deviceName); });

Awesome, it works! It seems that I need to read more about whenComplete, exceptionally, thenApply... 😉

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

bodiroga · 2020-05-23T00:21:09Z

Let's see if this is the final review! 👍

TravisBuddy · 2020-05-23T00:36:31Z

Travis tests were successful

Hey @bodiroga,
we found no major flaws with your code. Still you might want to look at this logfile, as we usually suggest some optional improvements.

cpmeister · 2020-05-23T01:42:48Z

...ing.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/tools/WaitForTopicValue.java

@@ -44,20 +45,24 @@
 public WaitForTopicValue(MqttBrokerConnection connection, String topic)
 throws InterruptedException, ExecutionException {


I don't see anywhere this would be thrown. It should be safe to remove these exceptions.

cpmeister · 2020-05-23T01:44:35Z

...ing.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/tools/WaitForTopicValue.java

 *
 * @author David Graeff - Initial contribution
 */
 @NonNullByDefault
 public class WaitForTopicValue {
- private CompletableFuture<String> future = new CompletableFuture<>();
- private final CompletableFuture<Boolean> subscripeFuture;
+ private final CompletableFuture<String> future = new CompletableFuture<>();


I don't think this needs to be a field anymore since it isn't used outside of the constructor.

cweitkamp · 2020-05-23T12:10:24Z

...ing.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/tools/WaitForTopicValue.java

@@ -44,20 +45,24 @@
 public WaitForTopicValue(MqttBrokerConnection connection, String topic)
 throws InterruptedException, ExecutionException {
 final MqttMessageSubscriber mqttMessageSubscriber = (t, payload) -> {
- future.complete(new String(payload));
+ try {
+ future.complete(new String(payload, "UTF-8"));


Is it possible to use the String(byte[] bytes, Charset charset) method with const StandardCharsets.UTF_8 to omit the UnsupportedEncodingException?

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

bodiroga · 2020-05-23T12:42:47Z

Done! Thank you so much for you comments guys!

TravisBuddy · 2020-05-23T12:54:46Z

Travis tests were successful

Hey @bodiroga,
we found no major flaws with your code. Still you might want to look at this logfile, as we usually suggest some optional improvements.

TravisBuddy · 2020-05-23T13:03:28Z

Travis tests were successful

Hey @bodiroga,
we found no major flaws with your code. Still you might want to look at this logfile, as we usually suggest some optional improvements.

jochen314

Not to myself: futures are not builders :-)

LGTM

cpmeister

LGTM

* Correctly unsubscribe after message is received * Address review comments * Use StandardCharset for UTF-8 Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

* Correctly unsubscribe after message is received * Address review comments * Use StandardCharset for UTF-8 Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com> Signed-off-by: CSchlipp <christian@schlipp.de>

* Correctly unsubscribe after message is received * Address review comments * Use StandardCharset for UTF-8 Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

* Correctly unsubscribe after message is received * Address review comments * Use StandardCharset for UTF-8 Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com> Signed-off-by: Daan Meijer <daan@studioseptember.nl>

* Correctly unsubscribe after message is received * Address review comments * Use StandardCharset for UTF-8 Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

Correctly unsubscribe after message is received

4e49754

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

bodiroga requested a review from davidgraeff as a code owner May 21, 2020 19:55

cpmeister requested changes May 21, 2020

View reviewed changes

Address review comments

bd89d05

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

cpmeister requested changes May 22, 2020

View reviewed changes

Address more review comments

4f6810a

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

cpmeister requested changes May 23, 2020

View reviewed changes

cweitkamp requested changes May 23, 2020

View reviewed changes

openhab deleted a comment from TravisBuddy May 23, 2020

bodiroga added 2 commits May 23, 2020 14:25

Address even more review comments

ebcd85b

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

Use StandardCharset for UTF-8

d284edd

Signed-off-by: Aitor Iturrioz <riturrioz@gmail.com>

jochen314 approved these changes May 23, 2020

View reviewed changes

cpmeister approved these changes May 23, 2020

View reviewed changes

cpmeister added the bug An unexpected problem or unintended behavior of an add-on label May 23, 2020

cpmeister merged commit b883512 into openhab:2.5.x May 23, 2020

cpmeister added this to the 2.5.6 milestone May 23, 2020

bodiroga deleted the fix-7252-name branch June 25, 2020 18:16

bodiroga mentioned this pull request Jun 27, 2020

[mqtt][homie] Delete '$name' subscription in discovery service #8017

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[mqtt] Correctly unsubscribe after message is received #7723

[mqtt] Correctly unsubscribe after message is received #7723

bodiroga commented May 21, 2020 •

edited

Loading

cpmeister left a comment

cpmeister May 21, 2020

cpmeister May 21, 2020

cpmeister May 21, 2020

bodiroga May 21, 2020

bodiroga commented May 21, 2020

cpmeister May 22, 2020

bodiroga May 22, 2020

cpmeister May 22, 2020

bodiroga May 22, 2020

cpmeister May 22, 2020

bodiroga May 23, 2020

bodiroga commented May 23, 2020

TravisBuddy commented May 23, 2020

cpmeister May 23, 2020

cpmeister May 23, 2020

cweitkamp May 23, 2020

bodiroga commented May 23, 2020

TravisBuddy commented May 23, 2020

TravisBuddy commented May 23, 2020

jochen314 left a comment

cpmeister left a comment

		@@ -46,11 +46,11 @@ public WaitForTopicValue(MqttBrokerConnection connection, String topic)
		final MqttMessageSubscriber mqttMessageSubscriber = (t, payload) -> {
		future.complete(new String(payload));

	private CompletableFuture<String> future = new CompletableFuture<>();
	private final CompletableFuture<String> future = new CompletableFuture<>();

		subscribeFuture = connection.subscribe(topic, mqttMessageSubscriber);

		composeFuture = subscribeFuture.thenCompose(b -> future);

		@@ -44,20 +45,24 @@
		public WaitForTopicValue(MqttBrokerConnection connection, String topic)
		throws InterruptedException, ExecutionException {

[mqtt] Correctly unsubscribe after message is received #7723

[mqtt] Correctly unsubscribe after message is received #7723

Conversation

bodiroga commented May 21, 2020 • edited Loading

cpmeister left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bodiroga commented May 21, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bodiroga commented May 23, 2020

TravisBuddy commented May 23, 2020

Travis tests were successful

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bodiroga commented May 23, 2020

TravisBuddy commented May 23, 2020

Travis tests were successful

TravisBuddy commented May 23, 2020

Travis tests were successful

jochen314 left a comment

Choose a reason for hiding this comment

cpmeister left a comment

Choose a reason for hiding this comment

bodiroga commented May 21, 2020 •

edited

Loading