Fixed draining of queue on ACK messages, related to issue #441

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);

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<SessionRegistry.EnqueuedMessage> sessionQueue) {
-        this(clean, clientId, sessionQueue);
+        this(clientId, clean, sessionQueue);
         this.will = will;
     }
 
-    Session(boolean clean, String clientId, Queue<SessionRegistry.EnqueuedMessage> sessionQueue) {
+    Session(String clientId, boolean clean, Queue<SessionRegistry.EnqueuedMessage> 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{" +

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();

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<SessionRegistry.EnqueuedMessage> 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);
+    }
+
+}