From cd908f7e44ebeef67b701111b5032f4c81b01d92 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 6 Mar 2016 19:50:26 +0100 Subject: [PATCH] Added memory queue to before pushing to Netty, first step to have a persistent queue per channel --- .../java/io/moquette/server/Constants.java | 1 + .../server/netty/AutoFlushHandler.java | 184 ++++++++++++++++++ .../server/netty/NettyMQTTHandler.java | 8 + .../java/io/moquette/spi/ClientSession.java | 18 ++ .../moquette/spi/impl/ProtocolProcessor.java | 72 +++++-- 5 files changed, 268 insertions(+), 15 deletions(-) create mode 100644 broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java diff --git a/broker/src/main/java/io/moquette/server/Constants.java b/broker/src/main/java/io/moquette/server/Constants.java index b22a7aac..bf39b9f8 100644 --- a/broker/src/main/java/io/moquette/server/Constants.java +++ b/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 } diff --git a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java new file mode 100644 index 00000000..d87f766b --- /dev/null +++ b/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); + } + } + } +} diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index f2edcda7..d13976a3 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/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(); + } } diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 8fb53d63..520d1b33 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/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 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(); + } } \ No newline at end of file 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 c8c730de..d0413deb 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; @@ -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) { - String clientID = NettyUtils.clientID(session); + public void processPubAck(Channel channel, PubAckMessage msg) { + 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); - channel.writeAndFlush(pubMessage); + LOG.trace("Session for clientId {}", clientId); +// 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(); + } }