dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix bug: Resending the message causes the client to interrupt

Browse Source 
In the mqtt 3.1 protocol or mqtt 3.1.1 protocol, except for messages of type pubish, the isDup of other messages must be zero，Sending a subscription message multiple times causes the client to reconnect，therefore, resending the subscription message with isDup true will cause the server to think that the message is abnormal, thus interrupting the connection with the client
This commit is contained in:
blackstar-baba 2020-05-24 23:22:51 +08:00 committed by GitHub
parent 507bd43a37
commit dda3124835
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 3 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
netty-mqtt/src/main/java/org/thingsboard/mqtt/RetransmissionHandler.java
View File

@ -19,6 +19,7 @@ import io.netty.channel.EventLoop;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.util.concurrent.ScheduledFuture;
import io.netty.handler.codec.mqtt.MqttMessageType;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
@ -44,7 +45,8 @@ final class RetransmissionHandler<T extends MqttMessage> {
private void startTimer(EventLoop eventLoop){
this.timer = eventLoop.schedule(() -> {
this.timeout += 5;
MqttFixedHeader fixedHeader = new MqttFixedHeader(this.originalMessage.fixedHeader().messageType(), true, this.originalMessage.fixedHeader().qosLevel(), this.originalMessage.fixedHeader().isRetain(), this.originalMessage.fixedHeader().remainingLength());
boolean isDup = this.originalMessage.fixedHeader().messageType() == MqttMessageType.PUBLISH ? true : this.originalMessage.fixedHeader().isDup();
MqttFixedHeader fixedHeader = new MqttFixedHeader(this.originalMessage.fixedHeader().messageType(), isDup, this.originalMessage.fixedHeader().qosLevel(), this.originalMessage.fixedHeader().isRetain(), this.originalMessage.fixedHeader().remainingLength());
handler.accept(fixedHeader, originalMessage);
startTimer(eventLoop);
}, timeout, TimeUnit.SECONDS);