From 3f7c8fe3ea6761d2617318a7d910d99a60362e23 Mon Sep 17 00:00:00 2001 From: andsel Date: Sat, 23 Feb 2019 17:09:02 +0100 Subject: [PATCH] Fixed draining of queue on ACK messages, related to issue #441 --- .../java/io/moquette/broker/PostOffice.java | 6 --- .../main/java/io/moquette/broker/Session.java | 20 ++++---- .../io/moquette/broker/SessionRegistry.java | 2 +- .../java/io/moquette/broker/SessionTest.java | 48 +++++++++++++++++++ 4 files changed, 58 insertions(+), 18 deletions(-) create mode 100644 broker/src/test/java/io/moquette/broker/SessionTest.java diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java index 2dcf9d0a..3adec2e5 100644 --- a/broker/src/main/java/io/moquette/broker/PostOffice.java +++ b/broker/src/main/java/io/moquette/broker/PostOffice.java @@ -216,12 +216,6 @@ class PostOffice { if (isSessionPresent) { LOG.debug("Sending PUBLISH message to active subscriber CId: {}, topicFilter: {}, qos: {}", sub.getClientId(), sub.getTopicFilter(), qos); -// final String login = targetSession.login(); -// if (!authorizator.canRead(topic, login, sub.getClientId())) { -// LOG.debug("Authorizator prohibit Client {} to be notified on {}", sub.getClientId(), topic); -// return; -// } - // we need to retain because duplicate only copy r/w indexes and don't retain() causing refCnt = 0 ByteBuf payload = origPayload.retainedDuplicate(); targetSession.sendPublishOnSessionAtQos(topic, qos, payload); diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java index 0662ef89..35622184 100644 --- a/broker/src/main/java/io/moquette/broker/Session.java +++ b/broker/src/main/java/io/moquette/broker/Session.java @@ -97,11 +97,11 @@ class Session { private final AtomicInteger inflightSlots = new AtomicInteger(INFLIGHT_WINDOW_SIZE); // this should be configurable Session(String clientId, boolean clean, Will will, Queue sessionQueue) { - this(clean, clientId, sessionQueue); + this(clientId, clean, sessionQueue); this.will = will; } - Session(boolean clean, String clientId, Queue sessionQueue) { + Session(String clientId, boolean clean, Queue sessionQueue) { this.clientId = clientId; this.clean = clean; this.sessionQueue = sessionQueue; @@ -267,6 +267,12 @@ class Session { mqttConnection.channel.isWritable(); } + private boolean inflighHasSlotsAndConnectionIsUp() { + return inflightSlots.get() > 0 && + connected() && + mqttConnection.channel.isWritable(); + } + void pubAckReceived(int ackPacketId) { // TODO remain to invoke in somehow m_interceptor.notifyMessageAcknowledged inflightWindow.remove(ackPacketId); @@ -315,7 +321,7 @@ class Session { private void drainQueueToConnection() { // consume the queue - while (!canSkipQueue()) { + while (!sessionQueue.isEmpty() && inflighHasSlotsAndConnectionIsUp()) { final SessionRegistry.EnqueuedMessage msg = sessionQueue.remove(); inflightSlots.decrementAndGet(); int sendPacketId = mqttConnection.nextPacketId(); @@ -362,14 +368,6 @@ class Session { ReferenceCountUtil.release(removedMsg); } -// public String login() { -// if (!connected()) { -// throw new IllegalStateException("Can't ask login on disconnected session"); -// } -// -// return NettyUtils.userName(mqttConnection.channel); -// } - @Override public String toString() { return "Session{" + diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java index dba89e47..fbd16229 100644 --- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java +++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java @@ -197,7 +197,7 @@ public class SessionRegistry { final Session.Will will = createWill(msg); newSession = new Session(clientId, clean, will, sessionQueue); } else { - newSession = new Session(clean, clientId, sessionQueue); + newSession = new Session(clientId, clean, sessionQueue); } newSession.markConnected(); diff --git a/broker/src/test/java/io/moquette/broker/SessionTest.java b/broker/src/test/java/io/moquette/broker/SessionTest.java new file mode 100644 index 00000000..adb203fc --- /dev/null +++ b/broker/src/test/java/io/moquette/broker/SessionTest.java @@ -0,0 +1,48 @@ +package io.moquette.broker; + +import io.moquette.broker.subscriptions.Topic; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufUtil; +import io.netty.buffer.UnpooledByteBufAllocator; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.mqtt.MqttQoS; +import org.junit.Test; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +import static org.junit.Assert.*; + +public class SessionTest { + + @Test + public void testPubAckDrainMessagesRemainingInQueue() { + final Queue queuedMessages = new ConcurrentLinkedQueue<>(); + final Session client = new Session("Subscriber", true, null, queuedMessages); + final EmbeddedChannel testChannel = new EmbeddedChannel(); + MQTTConnection mqttConnection = new MQTTConnection(testChannel, null, null, null, null); + client.markConnected(); + client.bind(mqttConnection); + + final Topic destinationTopic = new Topic("/a/b"); + sendQoS1To(client, destinationTopic, "Hello World!"); + // simulate a filling of inflight space and start pushing on queue + for (int i = 0; i < 10; i++) { + sendQoS1To(client, destinationTopic, "Hello World " + i + "!"); + } + + assertEquals("Inflight zone must be full, and the 11th message must be queued", + 1, queuedMessages.size()); + // Exercise + client.pubAckReceived(1); + + // Verify + assertTrue("Messages should be drained", queuedMessages.isEmpty()); + } + + private void sendQoS1To(Session client, Topic destinationTopic, String message) { + final ByteBuf payload = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, message); + client.sendPublishOnSessionAtQos(destinationTopic, MqttQoS.AT_LEAST_ONCE, payload); + } + +}