diff --git a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java index 97383346..48ce4e71 100644 --- a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java +++ b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java @@ -95,10 +95,10 @@ class NewNettyAcceptor { private static final Logger LOG = LoggerFactory.getLogger(NewNettyAcceptor.class); - EventLoopGroup m_bossGroup; - EventLoopGroup m_workerGroup; - BytesMetricsCollector m_bytesMetricsCollector = new BytesMetricsCollector(); - MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector(); + private EventLoopGroup bossGroup; + private EventLoopGroup workerGroup; + private BytesMetricsCollector bytesMetricsCollector = new BytesMetricsCollector(); + private MessageMetricsCollector metricsCollector = new MessageMetricsCollector(); private Optional metrics; private Optional errorsCather; @@ -125,13 +125,13 @@ class NewNettyAcceptor { boolean epoll = props.boolProp(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, false); if (epoll) { LOG.info("Netty is using Epoll"); - m_bossGroup = new EpollEventLoopGroup(); - m_workerGroup = new EpollEventLoopGroup(); + bossGroup = new EpollEventLoopGroup(); + workerGroup = new EpollEventLoopGroup(); channelClass = EpollServerSocketChannel.class; } else { LOG.info("Netty is using NIO"); - m_bossGroup = new NioEventLoopGroup(); - m_workerGroup = new NioEventLoopGroup(); + bossGroup = new NioEventLoopGroup(); + workerGroup = new NioEventLoopGroup(); channelClass = NioServerSocketChannel.class; } @@ -154,9 +154,7 @@ class NewNettyAcceptor { } initializePlainTCPTransport(mqttHandler, props); initializeWebSocketTransport(mqttHandler, props); - String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME); - String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME); - if (sslTcpPortProp != null || wssPortProp != null) { + if (securityPortsConfigured(props)) { SslContext sslContext = sslCtxCreator.initSSLContext(); if (sslContext == null) { LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks"); @@ -167,15 +165,21 @@ class NewNettyAcceptor { } } - private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) { + private boolean securityPortsConfigured(IConfig props) { + String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME); + String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME); + return sslTcpPortProp != null || wssPortProp != null; + } + + private void initFactory(String host, int port, String protocol, final PipelineInitializer pipelieInitializer) { LOG.debug("Initializing server. Protocol={}", protocol); ServerBootstrap b = new ServerBootstrap(); - b.group(m_bossGroup, m_workerGroup).channel(channelClass) + b.group(bossGroup, workerGroup).channel(channelClass) .childHandler(new ChannelInitializer() { @Override public void initChannel(SocketChannel ch) throws Exception { - pipeliner.init(ch); + pipelieInitializer.init(ch); } }) .option(ChannelOption.SO_BACKLOG, nettySoBacklog) @@ -200,7 +204,7 @@ class NewNettyAcceptor { String tcpPortProp = props.getProperty(PORT_PROPERTY_NAME, DISABLED_PORT_BIND); if (DISABLED_PORT_BIND.equals(tcpPortProp)) { LOG.info("Property {} has been set to {}. TCP MQTT will be disabled", BrokerConstants.PORT_PROPERTY_NAME, - DISABLED_PORT_BIND); + DISABLED_PORT_BIND); return; } int port = Integer.parseInt(tcpPortProp); @@ -209,33 +213,38 @@ class NewNettyAcceptor { @Override void init(SocketChannel channel) { ChannelPipeline pipeline = channel.pipeline(); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); - // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); - if (errorsCather.isPresent()) { - pipeline.addLast("bugsnagCatcher", errorsCather.get()); - } - pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); - pipeline.addLast("autoflush", new AutoFlushHandler(1, TimeUnit.SECONDS)); - pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); - pipeline.addLast("encoder", MqttEncoder.INSTANCE); - pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); - pipeline.addLast("messageLogger", new MQTTMessageLogger()); - if (metrics.isPresent()) { - pipeline.addLast("wizardMetrics", metrics.get()); - } - pipeline.addLast("handler", handler); + configureMQTTPipeline(pipeline, timeoutHandler, handler); } }); } + private void configureMQTTPipeline(ChannelPipeline pipeline, MoquetteIdleTimeoutHandler timeoutHandler, + NewNettyMQTTHandler handler) { + pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); + pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); + // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); + if (errorsCather.isPresent()) { + pipeline.addLast("bugsnagCatcher", errorsCather.get()); + } + pipeline.addFirst("bytemetrics", new BytesMetricsHandler(bytesMetricsCollector)); + pipeline.addLast("autoflush", new AutoFlushHandler(1, TimeUnit.SECONDS)); + pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); + pipeline.addLast("encoder", MqttEncoder.INSTANCE); + pipeline.addLast("metrics", new MessageMetricsHandler(metricsCollector)); + pipeline.addLast("messageLogger", new MQTTMessageLogger()); + if (metrics.isPresent()) { + pipeline.addLast("wizardMetrics", metrics.get()); + } + pipeline.addLast("handler", handler); + } + private void initializeWebSocketTransport(final NewNettyMQTTHandler handler, IConfig props) { LOG.debug("Configuring Websocket MQTT transport"); String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); if (DISABLED_PORT_BIND.equals(webSocketPortProp)) { // Do nothing no WebSocket configured LOG.info("Property {} has been setted to {}. Websocket MQTT will be disabled", - BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); + BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); return; } int port = Integer.parseInt(webSocketPortProp); @@ -254,15 +263,7 @@ class NewNettyAcceptor { new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); - pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); - pipeline.addLast("autoflush", new AutoFlushHandler(1, TimeUnit.SECONDS)); - pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); - pipeline.addLast("encoder", MqttEncoder.INSTANCE); - pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); - pipeline.addLast("messageLogger", new MQTTMessageLogger()); - pipeline.addLast("handler", handler); + configureMQTTPipeline(pipeline, timeoutHandler, handler); } }); } @@ -273,7 +274,7 @@ class NewNettyAcceptor { if (DISABLED_PORT_BIND.equals(sslPortProp)) { // Do nothing no SSL configured LOG.info("Property {} has been set to {}. SSL MQTT will be disabled", - BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); + BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); return; } @@ -290,16 +291,7 @@ class NewNettyAcceptor { void init(SocketChannel channel) throws Exception { ChannelPipeline pipeline = channel.pipeline(); pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth)); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); - // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); - pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); - pipeline.addLast("autoflush", new AutoFlushHandler(1, TimeUnit.SECONDS)); - pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); - pipeline.addLast("encoder", MqttEncoder.INSTANCE); - pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); - pipeline.addLast("messageLogger", new MQTTMessageLogger()); - pipeline.addLast("handler", handler); + configureMQTTPipeline(pipeline, timeoutHandler, handler); } }); } @@ -331,15 +323,8 @@ class NewNettyAcceptor { new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); - pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); - pipeline.addLast("autoflush", new AutoFlushHandler(1, TimeUnit.SECONDS)); - pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); - pipeline.addLast("encoder", MqttEncoder.INSTANCE); - pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); - pipeline.addLast("messageLogger", new MQTTMessageLogger()); - pipeline.addLast("handler", handler); + + configureMQTTPipeline(pipeline, timeoutHandler, handler); } }); } @@ -347,12 +332,12 @@ class NewNettyAcceptor { @SuppressWarnings("FutureReturnValueIgnored") public void close() { LOG.debug("Closing Netty acceptor..."); - if (m_workerGroup == null || m_bossGroup == null) { + if (workerGroup == null || bossGroup == null) { LOG.error("Netty acceptor is not initialized"); throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); } - Future workerWaiter = m_workerGroup.shutdownGracefully(); - Future bossWaiter = m_bossGroup.shutdownGracefully(); + Future workerWaiter = workerGroup.shutdownGracefully(); + Future bossWaiter = bossGroup.shutdownGracefully(); /* * We shouldn't raise an IllegalStateException if we are interrupted. If we did so, the @@ -366,20 +351,20 @@ class NewNettyAcceptor { LOG.warn("An InterruptedException was caught while waiting for event loops to terminate..."); } - if (!m_workerGroup.isTerminated()) { + if (!workerGroup.isTerminated()) { LOG.warn("Forcing shutdown of worker event loop..."); - m_workerGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS); + workerGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS); } - if (!m_bossGroup.isTerminated()) { + if (!bossGroup.isTerminated()) { LOG.warn("Forcing shutdown of boss event loop..."); - m_bossGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS); + bossGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS); } - MessageMetrics metrics = m_metricsCollector.computeMetrics(); - BytesMetrics bytesMetrics = m_bytesMetricsCollector.computeMetrics(); + MessageMetrics metrics = metricsCollector.computeMetrics(); + BytesMetrics bytesMetrics = bytesMetricsCollector.computeMetrics(); LOG.info("Metrics messages[read={}, write={}] bytes[read={}, write={}]", metrics.messagesRead(), - metrics.messagesWrote(), bytesMetrics.readBytes(), bytesMetrics.wroteBytes()); + metrics.messagesWrote(), bytesMetrics.readBytes(), bytesMetrics.wroteBytes()); } private ChannelHandler createSslHandler(SocketChannel channel, SslContext sslContext, boolean needsClientAuth) { diff --git a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java index 57f1ec49..d60c9c2a 100644 --- a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java @@ -21,6 +21,7 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.util.concurrent.EventExecutor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -49,7 +50,7 @@ public class AutoFlushHandler extends ChannelDuplexHandler { } @Override - public void handlerAdded(ChannelHandlerContext ctx) throws Exception { + 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. @@ -68,9 +69,9 @@ public class AutoFlushHandler extends ChannelDuplexHandler { @Override public void channelRegistered(ChannelHandlerContext ctx) throws Exception { // Initialize early if channel is active already. - if (ctx.channel().isActive()) { - initialize(ctx); - } +// if (ctx.channel().isActive()) { +// initialize(ctx); +// } super.channelRegistered(ctx); } @@ -79,7 +80,7 @@ public class AutoFlushHandler extends ChannelDuplexHandler { // 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); +// initialize(ctx); super.channelActive(ctx); } @@ -89,24 +90,6 @@ public class AutoFlushHandler extends ChannelDuplexHandler { 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 @@ -140,13 +123,12 @@ public class AutoFlushHandler extends ChannelDuplexHandler { /** * Is called when the write timeout expire. * - * @param ctx - * the channel context. + * @param ctx the channel context. */ private void channelIdle(ChannelHandlerContext ctx) { // ctx.fireUserEventTriggered(evt); - if (LOG.isDebugEnabled()) { - LOG.debug("Flushing idle Netty channel {} Cid: {}", ctx.channel(), NettyUtils.clientID(ctx.channel())); + if (LOG.isTraceEnabled()) { + LOG.trace("Flushing idle Netty channel {} Cid: {}", ctx.channel(), NettyUtils.clientID(ctx.channel())); } ctx.channel().flush(); }