From ae734ded8ec9a1da640cb5da3d122a862a777c1f Mon Sep 17 00:00:00 2001
From: andsel <selva.andre@gmail.com>
Date: Wed, 23 Mar 2016 23:24:54 +0100
Subject: [PATCH] Fixed ProtocolProcessor with bad merged code

---
 .../moquette/spi/impl/ProtocolProcessor.java  | 30 ++++++++++++++++++-
 1 file changed, 29 insertions(+), 1 deletion(-)

diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
index 99297786..7c484f58 100644
--- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
@@ -19,8 +19,10 @@ import java.nio.ByteBuffer;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
 
 import io.moquette.server.ConnectionDescriptor;
+import io.moquette.server.netty.AutoFlushHandler;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.ClientSession;
 import io.moquette.spi.IMatchingCondition;
@@ -241,8 +243,19 @@ public class ProtocolProcessor {
             //force the republish of stored QoS1 and QoS2
             republishStoredInSession(clientSession);
         }
+        int flushIntervalMs = 500/*(keepAlive * 1000) / 2*/;
+        setupAutoFlusher(channel.pipeline(), flushIntervalMs);
         LOG.info("CONNECT processed");
-//        LOG.info("CONNECT clients descriptors {}", m_clientIDs);
+    }
+
+    private void setupAutoFlusher(ChannelPipeline pipeline, int flushIntervalMs) {
+        AutoFlushHandler autoFlushHandler = new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS);
+        try {
+            pipeline.addAfter("idleEventHandler", "autoFlusher", autoFlushHandler);
+        } catch (NoSuchElementException nseex) {
+            //the idleEventHandler is not present on the pipeline
+            pipeline.addFirst("autoFlusher", autoFlushHandler);
+        }
     }
 
     private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
@@ -747,4 +760,19 @@ public class ProtocolProcessor {
         m_interceptor.notifyTopicSubscribed(newSubscription, username);
         return true;
     }
+
+    public void notifyChannelWritable(Channel channel) {
+        String clientID = NettyUtils.clientID(channel);
+        ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
+        boolean emptyQueue = false;
+        while (channel.isWritable()  && !emptyQueue) {
+            AbstractMessage msg = clientSession.dequeue();
+            if (msg == null) {
+                emptyQueue = true;
+            } else {
+                channel.write(msg);
+            }
+        }
+        channel.flush();
+    }
 }