Examples/request response

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

@@ -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/RequestResponse.java

@@ -0,0 +1,63 @@
+/*
+ * 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;
+
+/**
+ * @author Silvio Giebl
+ */
+public class RequestResponse {
+
+    public static void main(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(publish -> Mqtt5Publish.builder()
+                                .topic(publish.getResponseTopic().get())
+                                .qos(publish.getQos())
+                                .payload("response".getBytes())
+                                .correlationData(publish.getCorrelationData().orElse(null))
+                                .build()))
+                .subscribe();
+
+        requester.toAsync()
+                .subscribeWith()
+                .topicFilter("response/topic")
+                .callback(publish -> 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