Examples/request response

examples/src/main/java/com/hivemq/client/mqtt/examples/AsyncDemo.java

@@ -24,6 +24,8 @@
 import java.util.concurrent.TimeUnit;
 
 /**
+ * Small completely asynchronous example.
+ *
  * @author Silvio Giebl
  */
 public class AsyncDemo {

examples/src/main/java/com/hivemq/client/mqtt/examples/Mqtt5Features.java

@@ -30,7 +30,7 @@
 import java.util.concurrent.CountDownLatch;
 
 /**
- * Shows MQTT 5 features like session expiry, message expiry, user properties, topic aliases, flow control
+ * Shows MQTT 5 features like session expiry, message expiry, user properties, topic aliases, flow control.
  *
  * @author Silvio Giebl
  */
@@ -90,7 +90,7 @@ public static void main(final String[] args)throws InterruptedException {
                 .retain(true)
                 .contentType("text/plain")  // our payload is text
                 .messageExpiryInterval(120) // not so important, expire message after 2min if can not be delivered
-                .userProperties()           // add some user properties to the will message
+                .userProperties()           // add some user properties to the message
                     .add("sender", "demo-sender-1")
                     .add("receiver", "you")
                     .applyUserProperties()

examples/src/main/java/com/hivemq/client/mqtt/examples/ReconnectStrategy.java

@@ -34,6 +34,7 @@
 public class ReconnectStrategy {
 
     public static void main(final String[] args) throws InterruptedException {
+//        defaultReconnect();
 //        customizedReconnect();
         completelyCustom();
     }
@@ -103,7 +104,7 @@ private static CompletableFuture<byte[]> getOAuthToken() {
                     TimeUnit.SECONDS.sleep(1);
                     System.out.println("OAuth server is slow to respond ...");
                 }
-            } catch (InterruptedException e) {
+            } catch (final InterruptedException e) {
                 e.printStackTrace();
             }
             return new byte[] {1, 2, 3};

examples/src/main/java/com/hivemq/client/mqtt/examples/RequestResponse.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2018 dc-square and the HiveMQ MQTT Client Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.hivemq.client.mqtt.examples;
+
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
+
+/**
+ * Shows how to implement a request/response pattern using response topic and correlation data.
+ *
+ * @author Silvio Giebl
+ */
+public class RequestResponse {
+
+    public static void main(final String[] args) {
+        final Mqtt5Client requester = Mqtt5Client.builder().serverHost("broker.hivemq.com").build();
+        final Mqtt5Client responder = Mqtt5Client.builder().serverHost("broker.hivemq.com").build();
+
+        requester.toBlocking().connect();
+        responder.toBlocking().connect();
+
+        responder.toRx()
+                .publish(responder.toRx()
+                        .subscribeStreamWith()
+                        .topicFilter("request/topic")
+                        .applySubscribe()
+                        .map(requestPublish -> Mqtt5Publish.builder()
+                                .topic(requestPublish.getResponseTopic().get())
+                                .qos(requestPublish.getQos())
+                                .payload("response".getBytes())
+                                .correlationData(requestPublish.getCorrelationData().orElse(null))
+                                .build()))
+                .subscribe(); // this call is a reactive streams subscribe call, not an MQTT subscribe
+
+        requester.toAsync()
+                .subscribeWith()
+                .topicFilter("response/topic")
+                .callback(responsePublish -> System.out.println("received response"))
+                .send()
+                .thenCompose(subAck -> requester.toAsync()
+                        .publishWith()
+                        .topic("request/topic")
+                        .responseTopic("response/topic")
+                        .correlationData("1234".getBytes())
+                        .qos(MqttQos.EXACTLY_ONCE)
+                        .payload("request".getBytes())
+                        .send());
+    }
+}

src/main/java/com/hivemq/client/internal/mqtt/message/publish/MqttPublish.java

@@ -187,7 +187,7 @@ public long getRawMessageExpiryInterval() {
                 ((payloadFormatIndicator == null) ? "" : ", payloadFormatIndicator=" + payloadFormatIndicator) +
                 ((contentType == null) ? "" : ", contentType=" + contentType) +
                 ((responseTopic == null) ? "" : ", responseTopic=" + responseTopic) +
-                ((correlationData == null) ? "" : ", correlationData=" + correlationData) +
+                ((correlationData == null) ? "" : ", correlationData=" + correlationData.remaining() + "byte") +
                 StringUtil.prepend(", ", super.toAttributeString());
     }
 

src/main/java/com/hivemq/client/internal/mqtt/message/publish/MqttPublishResult.java

@@ -168,7 +168,7 @@ public MqttQos2Result(
 
         @Override
         public @NotNull String toString() {
-            return "MqttQos2PublishResult{" + toAttributeString() + '}';
+            return "MqttQos2Result{" + toAttributeString() + '}';
         }
 
         @Override