diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/MoquetteIdleTimoutHandler.java b/broker/src/main/java/org/eclipse/moquette/server/netty/MoquetteIdleTimoutHandler.java index 2a5d804e..01b75357 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/MoquetteIdleTimoutHandler.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/MoquetteIdleTimoutHandler.java @@ -16,10 +16,12 @@ package org.eclipse.moquette.server.netty; import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandler.Sharable; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.timeout.IdleState; import io.netty.handler.timeout.IdleStateEvent; +@Sharable class MoquetteIdleTimoutHandler extends ChannelDuplexHandler { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { @@ -29,9 +31,9 @@ class MoquetteIdleTimoutHandler extends ChannelDuplexHandler { //fire a channelInactive to trigger publish of Will ctx.fireChannelInactive(); ctx.close(); - } /*else if (e.getState() == IdleState.WRITER_IDLE) { - ctx.writeAndFlush(new PingMessage()); - }*/ + } + } else { + super.userEventTriggered(ctx, evt); } } } diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java index d22487db..231eb473 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java @@ -152,6 +152,7 @@ public class NettyAcceptor implements ServerAcceptor { private void initializePlainTCPTransport(IMessaging messaging, IConfig props) throws IOException { final NettyMQTTHandler handler = new NettyMQTTHandler(); + final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler(); handler.setMessaging(messaging); String host = props.getProperty(Constants.HOST_PROPERTY_NAME); int port = Integer.parseInt(props.getProperty(Constants.PORT_PROPERTY_NAME)); @@ -159,7 +160,7 @@ public class NettyAcceptor implements ServerAcceptor { @Override void init(ChannelPipeline pipeline) { pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); + pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MQTTDecoder()); @@ -180,6 +181,7 @@ public class NettyAcceptor implements ServerAcceptor { int port = Integer.parseInt(webSocketPortProp); final NettyMQTTHandler handler = new NettyMQTTHandler(); + final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler(); handler.setMessaging(messaging); String host = props.getProperty(Constants.HOST_PROPERTY_NAME); @@ -190,11 +192,10 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addLast("httpDecoder", new HttpRequestDecoder()); pipeline.addLast("aggregator", new HttpObjectAggregator(65536)); pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt, mqttv3.1, mqttv3.1.1")); - //pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler(null, "mqtt")); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); + pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MQTTDecoder()); pipeline.addLast("encoder", new MQTTEncoder()); @@ -216,6 +217,7 @@ public class NettyAcceptor implements ServerAcceptor { LOG.info("Starting SSL on port {}", sslPort); final NettyMQTTHandler handler = new NettyMQTTHandler(); + final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler(); handler.setMessaging(messaging); String host = props.getProperty(Constants.HOST_PROPERTY_NAME); initFactory(host, sslPort, new PipelineInitializer() { @@ -223,7 +225,7 @@ public class NettyAcceptor implements ServerAcceptor { void init(ChannelPipeline pipeline) throws Exception { pipeline.addLast("ssl", sslHandlerFactory.create()); pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); + pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MQTTDecoder()); @@ -243,6 +245,7 @@ public class NettyAcceptor implements ServerAcceptor { } int sslPort = Integer.parseInt(sslPortProp); final NettyMQTTHandler handler = new NettyMQTTHandler(); + final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler(); handler.setMessaging(messaging); String host = props.getProperty(Constants.HOST_PROPERTY_NAME); initFactory(host, sslPort, new PipelineInitializer() { @@ -256,7 +259,7 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); + pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MQTTDecoder()); pipeline.addLast("encoder", new MQTTEncoder()); diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java index 4f8608fe..09f50b24 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java @@ -55,11 +55,7 @@ public class NettyChannel implements ServerChannel { if (m_channel.pipeline().names().contains("idleStateHandler")) { m_channel.pipeline().remove("idleStateHandler"); } - if (m_channel.pipeline().names().contains("idleEventHandler")) { - m_channel.pipeline().remove("idleEventHandler"); - } m_channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime)); - m_channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); } public void close(boolean immediately) { diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java index 48475336..17777fb8 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java @@ -101,7 +101,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { public void init(IConfig configProps) { subscriptions = new SubscriptionsStore(); m_executor = Executors.newFixedThreadPool(1); - m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); + m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 128, m_executor); //1024 * 32 /*Disruptor m_disruptor = new Disruptor(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor, ProducerType.MULTI, new BusySpinWaitStrategy());*/ m_disruptor.handleEventsWith(this); @@ -115,13 +115,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { } private void disruptorPublish(MessagingEvent msgEvent) { - LOG.debug("disruptorPublish publishing event {}", msgEvent); + LOG.debug("disruptorPublish publishing event {}, remaining capacity {}", msgEvent, m_ringBuffer.remainingCapacity()); long sequence = m_ringBuffer.next(); ValueEvent event = m_ringBuffer.get(sequence); - event.setEvent(msgEvent); - - m_ringBuffer.publish(sequence); + m_ringBuffer.publish(sequence); } @Override