Added memory queue to before pushing to Netty, first step to have a persistent queue per channel

broker/src/main/java/io/moquette/server/Constants.java

@@ -22,4 +22,5 @@ public class Constants {
     public static final String ATTR_CLIENTID = "ClientID";
     public static final String CLEAN_SESSION = "cleanSession";
     public static final String KEEP_ALIVE = "keepAlive";
+    public static final int MAX_MESSAGE_QUEUE = 1024; //number of messages
 }

broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java

@@ -0,0 +1,184 @@
+/*
+ * Copyright (c) 2012-2016 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * The Apache License v2.0 is available at
+ * http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.moquette.server.netty;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.concurrent.EventExecutor;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Auto-flush data on channel after a read timeout.
+ * It's inspired by IdleStateHandler but it's specialized version, just flushing data after
+ * no read is done on the channel after a period. It's used to avoid aggressively flushing from
+ * the ProtocolProcessor.
+ */
+public class AutoFlushHandler extends ChannelDuplexHandler {
+
+    private static final long MIN_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(1);
+
+    private final long writerIdleTimeNanos;
+    volatile ScheduledFuture<?> writerIdleTimeout;
+    volatile long lastWriteTime;
+//    private boolean firstWriterIdleEvent = true;
+
+    private volatile int state; // 0 - none, 1 - initialized, 2 - destroyed
+
+    public AutoFlushHandler(long writerIdleTime, TimeUnit unit) {
+        if (unit == null) {
+            throw new NullPointerException("unit");
+        }
+        writerIdleTimeNanos = Math.max(unit.toNanos(writerIdleTime), MIN_TIMEOUT_NANOS);
+    }
+
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+        if (ctx.channel().isActive() && ctx.channel().isRegistered()) {
+            // channelActive() event has been fired already, which means this.channelActive() will
+            // not be invoked. We have to initialize here instead.
+            initialize(ctx);
+        } else {
+            // channelActive() event has not been fired yet.  this.channelActive() will be invoked
+            // and initialization will occur there.
+        }
+    }
+
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+        destroy();
+    }
+
+    @Override
+    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
+        // Initialize early if channel is active already.
+        if (ctx.channel().isActive()) {
+            initialize(ctx);
+        }
+        super.channelRegistered(ctx);
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        // This method will be invoked only if this handler was added
+        // before channelActive() event is fired.  If a user adds this handler
+        // after the channelActive() event, initialize() will be called by beforeAdd().
+        initialize(ctx);
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        destroy();
+        super.channelInactive(ctx);
+    }
+
+//    @Override
+//    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+//        if (writerrIdleTimeNanos > 0) {
+//            reading = true;
+//            firstReaderIdleEvent = true;
+//        }
+//        ctx.fireChannelRead(msg);
+//    }
+//
+//    @Override
+//    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+//        if (writerrIdleTimeNanos > 0) {
+//            lastReadTime = System.nanoTime();
+//            reading = false;
+//        }
+//        ctx.fireChannelReadComplete();
+//    }
+
+    private void initialize(ChannelHandlerContext ctx) {
+        // Avoid the case where destroy() is called before scheduling timeouts.
+        // See: https://github.com/netty/netty/issues/143
+        switch (state) {
+            case 1:
+            case 2:
+                return;
+        }
+
+        state = 1;
+
+        EventExecutor loop = ctx.executor();
+
+        lastWriteTime = System.nanoTime();
+        writerIdleTimeout = loop.schedule(
+                new WriterIdleTimeoutTask(ctx),
+                writerIdleTimeNanos, TimeUnit.NANOSECONDS);
+    }
+
+    private void destroy() {
+        state = 2;
+
+        if (writerIdleTimeout != null) {
+            writerIdleTimeout.cancel(false);
+            writerIdleTimeout = null;
+        }
+    }
+
+    /**
+     * Is called when the write timeout expire.
+     */
+    protected void channelIdle(ChannelHandlerContext ctx/*, IdleStateEvent evt*/) throws Exception {
+//        ctx.fireUserEventTriggered(evt);
+        ctx.channel().flush();
+    }
+
+    private final class WriterIdleTimeoutTask implements Runnable {
+
+        private final ChannelHandlerContext ctx;
+
+        WriterIdleTimeoutTask(ChannelHandlerContext ctx) {
+            this.ctx = ctx;
+        }
+
+        @Override
+        public void run() {
+            if (!ctx.channel().isOpen()) {
+                return;
+            }
+
+//            long lastWriteTime = IdleStateHandler.this.lastWriteTime;
+//            long lastWriteTime = lastWriteTime;
+            long nextDelay = writerIdleTimeNanos - (System.nanoTime() - lastWriteTime);
+            if (nextDelay <= 0) {
+                // Writer is idle - set a new timeout and notify the callback.
+                writerIdleTimeout = ctx.executor().schedule(
+                        this, writerIdleTimeNanos, TimeUnit.NANOSECONDS);
+                try {
+                    /*IdleStateEvent event;
+                    if (firstWriterIdleEvent) {
+                        firstWriterIdleEvent = false;
+                        event = IdleStateEvent.FIRST_WRITER_IDLE_STATE_EVENT;
+                    } else {
+                        event = IdleStateEvent.WRITER_IDLE_STATE_EVENT;
+                    }*/
+                    channelIdle(ctx/*, event*/);
+                } catch (Throwable t) {
+                    ctx.fireExceptionCaught(t);
+                }
+            } else {
+                // Write occurred before the timeout - set a new timeout with shorter delay.
+                writerIdleTimeout = ctx.executor().schedule(this, nextDelay, TimeUnit.NANOSECONDS);
+            }
+        }
+    }
+}

broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java

@@ -111,4 +111,12 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
         }
         ctx.close();
     }
+
+    @Override
+    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+        if (ctx.channel().isWritable()) {
+            m_processor.notifyChannelWritable(ctx.channel());
+        }
+        ctx.fireChannelWritabilityChanged();
+    }
 }

broker/src/main/java/io/moquette/spi/ClientSession.java

@@ -16,6 +16,8 @@
 package io.moquette.spi;
 
 import io.moquette.parser.proto.messages.AbstractMessage;
+import io.moquette.parser.proto.messages.PublishMessage;
+import io.moquette.server.Constants;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
 import io.moquette.spi.ISessionsStore.ClientTopicCouple;
@@ -26,6 +28,8 @@ import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 
 /**
  * Model a Session like describe on page 25 of MQTT 3.1.1 specification:
@@ -59,6 +63,8 @@ public class ClientSession {
 
     private boolean active = false;
 
+    private BlockingQueue<AbstractMessage> m_queueToPublish = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE);
+
     public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore,
                          boolean cleanSession) {
         this.clientID = clientID;
@@ -191,4 +197,16 @@ public class ClientSession {
         final String guid = m_sessionsStore.mapToGuid(clientID, messageID);
         return messagesStore.getMessageByGuid(guid);
     }
+
+    /**
+     * Enqueue a message to be sent to the client.
+     * @return false if the queue is full.
+     * */
+    public boolean enqueue(PublishMessage pubMessage) {
+        return m_queueToPublish.offer(pubMessage);
+    }
+
+    public AbstractMessage dequeue() {
+        return m_queueToPublish.poll();
+    }
 }

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;
@@ -207,8 +209,9 @@ public class ProtocolProcessor {
             byte[] willPayload = msg.getWillMessage();
             ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
             //save the will testament in the clientID store
-            WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos );
+            WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos);
             m_willStore.put(msg.getClientID(), will);
+            LOG.info("Session for clientID <{}> with will to topic {}", msg.getClientID(), msg.getWillTopic());
         }
 
         ConnAckMessage okResp = new ConnAckMessage();
@@ -238,8 +241,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) {
@@ -275,8 +289,8 @@ public class ProtocolProcessor {
         }
     }
     
-    public void processPubAck(Channel session, PubAckMessage msg) {
+    public void processPubAck(Channel channel, PubAckMessage msg) {
-        String clientID = NettyUtils.clientID(session);
+        String clientID = NettyUtils.clientID(channel);
         int messageID = msg.getMessageID();
         //Remove the message from message store
         ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
@@ -303,12 +317,12 @@ public class ProtocolProcessor {
         return pub;
     }
     
-    public void processPublish(Channel session, PublishMessage msg) {
+    public void processPublish(Channel channel, PublishMessage msg) {
         LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
-        String clientID = NettyUtils.clientID(session);
+        String clientID = NettyUtils.clientID(channel);
         final String topic = msg.getTopicName();
         //check if the topic can be wrote
-        String user = NettyUtils.userName(session);
+        String user = NettyUtils.userName(channel);
         if (!m_authorizator.canWrite(topic, user, clientID)) {
             LOG.debug("topic {} doesn't have write credentials", topic);
             return;
@@ -457,9 +471,11 @@ public class ProtocolProcessor {
         }
     }
 
-    protected void directSend(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) {
+    protected void directSend(ClientSession clientsession, String topic, AbstractMessage.QOSType qos,
+                              ByteBuffer message, boolean retained, Integer messageID) {
         String clientId = clientsession.clientID;
-        LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID);
+        LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}",
+                clientId, topic, qos, retained, messageID);
         PublishMessage pubMessage = new PublishMessage();
         pubMessage.setRetainFlag(retained);
         pubMessage.setTopicName(topic);
@@ -475,22 +491,32 @@ public class ProtocolProcessor {
             pubMessage.setMessageID(messageID);
         } else {
             if (messageID != null) {
-                throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message with PacketIdentifier: " + messageID);
+                throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message " +
+                        "with PacketIdentifier: " + messageID);
             }
         }
 
         if (m_clientIDs == null) {
-            throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
+            throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be " +
+                    "initialized, somewhere it's overwritten!!");
         }
-        LOG.debug("clientIDs are {}", m_clientIDs);
+        //LOG.trace("clientIDs are {}", m_clientIDs);
         if (m_clientIDs.get(clientId) == null) {
             //TODO while we were publishing to the target client, that client disconnected,
             // could happen is not an error HANDLE IT
-            throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs));
+            throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>",
+                    clientId, m_clientIDs));
         }
         Channel channel = m_clientIDs.get(clientId).channel;
-        LOG.debug("Session for clientId {} is {}", clientId, channel);
+        LOG.trace("Session for clientId {}", clientId);
-        channel.writeAndFlush(pubMessage);
+//        channel.writeAndFlush(pubMessage);
+        if (channel.isWritable()) {
+            //if channel is writable don't enqueue
+            channel.write(pubMessage);
+        } else {
+            //enqueue to the client session
+            clientsession.enqueue(pubMessage);
+        }
     }
     
     private void sendPubRec(String clientID, int messageID) {
@@ -580,6 +606,7 @@ public class ProtocolProcessor {
     }
     
     public void processDisconnect(Channel channel) throws InterruptedException {
+        channel.flush();
         String clientID = NettyUtils.clientID(channel);
         boolean cleanSession = NettyUtils.cleanSession(channel);
         LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
@@ -717,4 +744,19 @@ public class ProtocolProcessor {
         m_interceptor.notifyTopicSubscribed(newSubscription);
         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();
+    }
 }