bump mqtt version to 4.1.66.Final (#73)

streamnative · Jul 27, 2021 · 29a27f7 · 29a27f7
1 parent cd58719
commit 29a27f7
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 2 deletions.
diff --git a/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/ProxyHandler.java b/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/ProxyHandler.java
@@ -117,6 +117,9 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
             log.info("channel read: {}", message);
+            if (message instanceof MqttMessage && ((MqttMessage) message).decoderResult().isFailure()) {
+                log.error("Failed to decode mqttMessage.", ((MqttMessage) message).decoderResult().cause());
+            }
             switch (state) {
                 case Init:
                     MqttMessage msg = (MqttMessage) message;

diff --git a/...c/main/java/io/streamnative/pulsar/handlers/mqtt/support/ProtocolMethodProcessorImpl.java b/...c/main/java/io/streamnative/pulsar/handlers/mqtt/support/ProtocolMethodProcessorImpl.java
@@ -462,7 +462,7 @@ private MqttSubAckMessage doAckMessageFromValidateFilters(List<MqttTopicSubscrip
             grantedQoSLevels.add(req.qualityOfService().value());
         }
 
-        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_LEAST_ONCE,
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE,
                 false, 0);
         MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels);
         return new MqttSubAckMessage(fixedHeader, MqttMessageIdVariableHeader.from(messageId), payload);

diff --git a/pom.xml b/pom.xml
@@ -50,7 +50,7 @@
         <testng.version>6.14.3</testng.version>
         <awaitility.version>4.0.2</awaitility.version>
         <pulsar.version>2.8.0</pulsar.version>
-        <mqtt.codec.version>4.1.49.Final</mqtt.codec.version>
+        <mqtt.codec.version>4.1.66.Final</mqtt.codec.version>
         <log4j2.version>2.13.3</log4j2.version>
         <lombok.version>1.18.4</lombok.version>
         <mqtt.client.version>1.16</mqtt.client.version>