Skip to content

Commit

Permalink
[MQTT 5] Avoid to pub retries on timeout (moquette-io#697)
Browse files Browse the repository at this point in the history 
This PR stores the MQTT version into the Session instance and use that to keep the existing behavior for inflight resends (happening on a timeout basis on ACK received) in case the version is MQTT 3.1 or MQTT 3.1.1.
When the version of the Session is MQTT 5 it removes the resend on PUB ACK timeouts and switch to send only in case the same client reconnects with cleanStart = 0 and there is any peding publishes in the flight zone to get acknowledged.

To test this use the raw Client has been extended, so now can also subscribe and collect publish messages.
  • Loading branch information
@andsel
andsel committed Jul 31, 2023
1 parent 4f06ae4 commit dc99e84
Show file tree
Hide file tree
Showing 8 changed files with 326 additions and 46 deletions.
4 changes: 2 additions & 2 deletions .github/workflows/maven_build.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,9 @@ name: Java CI with Maven

on:
push:
branches: [ main ]
branches: [ main, mqtt5_development ]
pull_request:
branches: [ main ]
branches: [ main, mqtt5_development ]

jobs:
test:
Expand Down
8 changes: 8 additions & 0 deletions broker/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
https://github.com/netty/netty/blob/netty-4.1.93.Final/pom.xml#L625 -->
<netty.tcnative.version>2.0.61.Final</netty.tcnative.version>
<paho.version>1.2.5</paho.version>
<hivemqclient.version>1.3.0</hivemqclient.version>
<h2.version>2.1.212</h2.version>
</properties>

Expand Down Expand Up @@ -146,6 +147,13 @@
<scope>test</scope>
</dependency>

<dependency>
<groupId>com.hivemq</groupId>
<artifactId>hivemq-mqtt-client</artifactId>
<version>${hivemqclient.version}</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>org.eclipse.jetty.websocket</groupId>
<artifactId>websocket-client</artifactId>
Expand Down
11 changes: 9 additions & 2 deletions broker/src/main/java/io/moquette/broker/MQTTConnection.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,10 @@ PostOffice.RouteResult processConnect(MqttConnectMessage msg) {
final String username = payload.userName();
LOG.trace("Processing CONNECT message. CId: {} username: {}", clientId, username);

if (isNotProtocolVersion(msg, MqttVersion.MQTT_3_1) && isNotProtocolVersion(msg, MqttVersion.MQTT_3_1_1)) {
if (isNotProtocolVersion(msg, MqttVersion.MQTT_3_1) &&
isNotProtocolVersion(msg, MqttVersion.MQTT_3_1_1) &&
isNotProtocolVersion(msg, MqttVersion.MQTT_5)
) {
LOG.warn("MQTT protocol version is not valid. CId: {}", clientId);
abortConnection(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
return PostOffice.RouteResult.failed(clientId);
Expand Down Expand Up @@ -240,12 +243,16 @@ public void operationComplete(ChannelFuture future) throws Exception {
final Session session = result.session;
postOffice.routeCommand(session.getClientID(), "sendOfflineMessages", () -> {
session.sendQueuedMessagesWhileOffline();
session.reconnectSession();
return null;
});
}

initializeKeepAliveTimeout(channel, msg, clientIdUsed);
setupInflightResender(channel);
if (isNotProtocolVersion(msg, MqttVersion.MQTT_5)) {
// In MQTT5 MQTT-4.4.0-1 avoid retries messages on timer base.
setupInflightResender(channel);
}

postOffice.dispatchConnection(msg);
LOG.trace("dispatch connection: {}", msg);
Expand Down
75 changes: 52 additions & 23 deletions broker/src/main/java/io/moquette/broker/Session.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -46,13 +47,15 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

class Session {

private static final Logger LOG = LoggerFactory.getLogger(Session.class);
// By specification session expiry value of 0xEFFFFFFF (UINT_MAX) (seconds) means
// session that doesn't expire, it's ~68 years.
static final int INFINITE_EXPIRY = Integer.MAX_VALUE;
private final boolean resendInflightOnTimeout;

static class InFlightPacket implements Delayed {

Expand Down Expand Up @@ -109,6 +112,7 @@ static final class Will {
private MQTTConnection mqttConnection;
private final Set<Subscription> subscriptions = new HashSet<>();
private final Map<Integer, SessionRegistry.EnqueuedMessage> inflightWindow = new HashMap<>();
// used only in MQTT3 where resends are done on timeout of ACKs.
private final DelayQueue<InFlightPacket> inflightTimeouts = new DelayQueue<>();
private final Map<Integer, MqttPublishMessage> qos2Receiving = new HashMap<>();
private final AtomicInteger inflightSlots = new AtomicInteger(INFLIGHT_WINDOW_SIZE); // this should be configurable
Expand All @@ -126,6 +130,9 @@ static final class Will {
this.data = data;
this.clean = clean;
this.sessionQueue = sessionQueue;
// in MQTT3 cleanSession = true means expiryInterval=0 else infinite
// expiryInterval = clean ? 0 : 0xFFFFFFFF;
this.resendInflightOnTimeout = data.protocolVersion() != MqttVersion.MQTT_5;
}

public boolean expireImmediately() {
Expand Down Expand Up @@ -230,7 +237,9 @@ public void processPubRec(int pubRecPacketId) {
return;
}
inflightWindow.put(pubRecPacketId, new SessionRegistry.PubRelMarker());
inflightTimeouts.add(new InFlightPacket(pubRecPacketId, FLIGHT_BEFORE_RESEND_MS));
if (resendInflightOnTimeout) {
inflightTimeouts.add(new InFlightPacket(pubRecPacketId, FLIGHT_BEFORE_RESEND_MS));
}
MqttMessage pubRel = MQTTConnection.pubrel(pubRecPacketId);
mqttConnection.sendIfWritableElseDrop(pubRel);

Expand Down Expand Up @@ -299,7 +308,9 @@ private void sendPublishQos1(Topic topic, MqttQoS qos, ByteBuf payload, boolean
old.release();
inflightSlots.incrementAndGet();
}
inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));
if (resendInflightOnTimeout) {
inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));
}

MqttPublishMessage publishMsg = MQTTConnection.createNotRetainedPublishMessage(topic.toString(), qos,
payload, packetId);
Expand Down Expand Up @@ -333,8 +344,9 @@ private void sendPublishQos2(Topic topic, MqttQoS qos, ByteBuf payload, boolean
old.release();
inflightSlots.incrementAndGet();
}
inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));

if (resendInflightOnTimeout) {
inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));
}
MqttPublishMessage publishMsg = MQTTConnection.createNotRetainedPublishMessage(topic.toString(), qos,
payload, packetId);
localMqttConnectionRef.sendPublish(publishMsg);
Expand All @@ -356,7 +368,7 @@ private boolean canSkipQueue(MQTTConnection localMqttConnectionRef) {
localMqttConnectionRef.channel.isWritable();
}

private boolean inflighHasSlotsAndConnectionIsUp() {
private boolean inflightHasSlotsAndConnectionIsUp() {
return inflightSlots.get() > 0 &&
connected() &&
mqttConnection.channel.isWritable();
Expand All @@ -380,20 +392,30 @@ public void flushAllQueuedMessages() {
}

public void resendInflightNotAcked() {
Collection<InFlightPacket> expired = new ArrayList<>(INFLIGHT_WINDOW_SIZE);
inflightTimeouts.drainTo(expired);
Collection<Integer> nonAckPacketIds;
if (resendInflightOnTimeout) {
// MQTT3 behavior, resend on timeout
Collection<InFlightPacket> expired = new ArrayList<>(INFLIGHT_WINDOW_SIZE);
inflightTimeouts.drainTo(expired);
nonAckPacketIds = expired.stream().map(p -> p.packetId).collect(Collectors.toList());
} else {
// MQTT5 behavior resend only not acked present in reopened session.
nonAckPacketIds = inflightWindow.keySet();
}

debugLogPacketIds(expired);
debugLogPacketIds(nonAckPacketIds);

for (InFlightPacket notAckPacketId : expired) {
final SessionRegistry.EnqueuedMessage msg = inflightWindow.get(notAckPacketId.packetId);
for (Integer notAckPacketId : nonAckPacketIds) {
final SessionRegistry.EnqueuedMessage msg = inflightWindow.get(notAckPacketId);
if (msg == null) {
// Already acked...
continue;
}
if (msg instanceof SessionRegistry.PubRelMarker) {
MqttMessage pubRel = MQTTConnection.pubrel(notAckPacketId.packetId);
inflightTimeouts.add(new InFlightPacket(notAckPacketId.packetId, FLIGHT_BEFORE_RESEND_MS));
MqttMessage pubRel = MQTTConnection.pubrel(notAckPacketId);
if (resendInflightOnTimeout) {
inflightTimeouts.add(new InFlightPacket(notAckPacketId, FLIGHT_BEFORE_RESEND_MS));
}
mqttConnection.sendIfWritableElseDrop(pubRel);
} else {
final SessionRegistry.PublishedMessage pubMsg = (SessionRegistry.PublishedMessage) msg;
Expand All @@ -402,34 +424,36 @@ public void resendInflightNotAcked() {
final ByteBuf payload = pubMsg.payload;
// message fetched from map, but not removed from map. No need to duplicate or release.
MqttPublishMessage publishMsg = publishNotRetainedDuplicated(notAckPacketId, topic, qos, payload);
inflightTimeouts.add(new InFlightPacket(notAckPacketId.packetId, FLIGHT_BEFORE_RESEND_MS));
if (resendInflightOnTimeout) {
inflightTimeouts.add(new InFlightPacket(notAckPacketId, FLIGHT_BEFORE_RESEND_MS));
}
mqttConnection.sendPublish(publishMsg);
}
}
}

private void debugLogPacketIds(Collection<InFlightPacket> expired) {
if (!LOG.isDebugEnabled() || expired.isEmpty()) {
private void debugLogPacketIds(Collection<Integer> packetIds) {
if (!LOG.isDebugEnabled() || packetIds.isEmpty()) {
return;
}

StringBuilder sb = new StringBuilder();
for (InFlightPacket packet : expired) {
sb.append(packet.packetId).append(", ");
for (Integer packetId : packetIds) {
sb.append(packetId).append(", ");
}
LOG.debug("Resending {} in flight packets [{}]", expired.size(), sb);
LOG.debug("Resending {} in flight packets [{}]", packetIds.size(), sb);
}

private MqttPublishMessage publishNotRetainedDuplicated(InFlightPacket notAckPacketId, Topic topic, MqttQoS qos,
private MqttPublishMessage publishNotRetainedDuplicated(int packetId, Topic topic, MqttQoS qos,
ByteBuf payload) {
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, true, qos, false, 0);
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic.toString(), notAckPacketId.packetId);
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic.toString(), packetId);
return new MqttPublishMessage(fixedHeader, varHeader, payload);
}

private void drainQueueToConnection() {
// consume the queue
while (connected() && !sessionQueue.isEmpty() && inflighHasSlotsAndConnectionIsUp()) {
while (connected() && !sessionQueue.isEmpty() && inflightHasSlotsAndConnectionIsUp()) {
final SessionRegistry.EnqueuedMessage msg = sessionQueue.dequeue();
if (msg == null) {
// Our message was already fetched by another Thread.
Expand All @@ -444,7 +468,9 @@ private void drainQueueToConnection() {
old.release();
inflightSlots.incrementAndGet();
}
inflightTimeouts.add(new InFlightPacket(sendPacketId, FLIGHT_BEFORE_RESEND_MS));
if (resendInflightOnTimeout) {
inflightTimeouts.add(new InFlightPacket(sendPacketId, FLIGHT_BEFORE_RESEND_MS));
}
final SessionRegistry.PublishedMessage msgPub = (SessionRegistry.PublishedMessage) msg;
MqttPublishMessage publishMsg = MQTTConnection.createNotRetainedPublishMessage(
msgPub.topic.toString(),
Expand All @@ -460,8 +486,11 @@ public void writabilityChanged() {
drainQueueToConnection();
}

public void sendQueuedMessagesWhileOffline() {
public void reconnectSession() {
LOG.trace("Republishing all saved messages for session {}", this);
resendInflightNotAcked();

// send queued messages while offline
drainQueueToConnection();
}

Expand Down
4 changes: 2 additions & 2 deletions broker/src/main/java/io/moquette/broker/SessionRegistry.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -300,9 +300,9 @@ private Session createNewSession(MqttConnectMessage msg, String clientId) {
}
// in MQTT3 cleanSession = true means expiryInterval=0 else infinite
final int expiryInterval = clean ? 0 : globalExpirySeconds;

final MqttVersion mqttVersion = Utils.versionFromConnect(msg);
final ISessionsRepository.SessionData sessionData = new ISessionsRepository.SessionData(clientId,
MqttVersion.MQTT_3_1_1, expiryInterval, clock);
mqttVersion, expiryInterval, clock);
if (msg.variableHeader().isWillFlag()) {
final Session.Will will = createWill(msg);
newSession = new Session(sessionData, clean, will, queue);
Expand Down
7 changes: 7 additions & 0 deletions broker/src/main/java/io/moquette/broker/Utils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,11 @@
package io.moquette.broker;

import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttVersion;

import java.util.Map;

/**
Expand Down Expand Up @@ -46,6 +49,10 @@ public static byte[] readBytesAndRewind(ByteBuf payload) {
return payloadContent;
}

public static MqttVersion versionFromConnect(MqttConnectMessage msg) {
return MqttVersion.fromProtocolNameAndLevel(msg.variableHeader().name(), (byte) msg.variableHeader().version());
}

private Utils() {
}
}
Loading

0 comments on commit dc99e84

Please sign in to comment.