Avoid slow responses due to missed queue drain after a client sent out all its messages

Author: andsel

broker/src/main/java/io/moquette/broker/MQTTConnection.java

@@ -406,8 +406,7 @@ void sendIfWritableElseDrop(MqttMessage msg) {
             ChannelFuture channelFuture;
             if (brokerConfig.isImmediateBufferFlush()) {
                 channelFuture = channel.writeAndFlush(msg);
-            }
-            else {
+            } else {
                 channelFuture = channel.write(msg);
             }
             channelFuture.addListener(FIRE_EXCEPTION_ON_FAILURE);
@@ -501,4 +500,9 @@ public String toString() {
     InetSocketAddress remoteAddress() {
         return (InetSocketAddress) channel.remoteAddress();
     }
+
+    public void readCompleted() {
+        // TODO drain all messages in target's session in-flight message queue
+        postOffice.flushInFlight(this);
+    }
 }

broker/src/main/java/io/moquette/broker/NewNettyMQTTHandler.java

@@ -70,6 +70,12 @@ public void operationComplete(ChannelFuture future) {
         }
     }
 
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        final MQTTConnection mqttConnection = mqttConnection(ctx.channel());
+        mqttConnection.readCompleted();
+    }
+
     @Override
     public void channelActive(ChannelHandlerContext ctx) {
         MQTTConnection connection = connectionFactory.create(ctx.channel());

broker/src/main/java/io/moquette/broker/PostOffice.java

@@ -308,4 +308,9 @@ void dispatchDisconnection(String clientId,String userName){
     void dispatchConnectionLost(String clientId,String userName){
         interceptor.notifyClientConnectionLost(clientId,userName);
     }
+
+    void flushInFlight(MQTTConnection mqttConnection) {
+        Session targetSession = sessionRegistry.retrieve(mqttConnection.getClientId());
+        targetSession.flushAllQueuedMessages();
+    }
 }

broker/src/main/java/io/moquette/broker/Session.java

@@ -281,6 +281,10 @@ void pubAckReceived(int ackPacketId) {
         drainQueueToConnection();
     }
 
+    public void flushAllQueuedMessages() {
+        drainQueueToConnection();
+    }
+
     public void resendInflightNotAcked() {
         Collection<InFlightPacket> expired = new ArrayList<>(INFLIGHT_WINDOW_SIZE);
         inflightTimeouts.drainTo(expired);