wildfire chat server source initial submit

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

@@ -0,0 +1,191 @@
+/*
+ * 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.util.concurrent.EventExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+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 Logger LOG = LoggerFactory.getLogger(AutoFlushHandler.class);
+    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
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Initializing autoflush handler. MqttClientId = {}.", NettyUtils.clientID(ctx.channel()));
+        }
+        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.
+     *
+     * @param ctx
+     *            the channel context.
+     * @throws Exception
+     *             in case of any IO error.
+     */
+    protected void channelIdle(ChannelHandlerContext ctx/* , IdleStateEvent evt */) throws Exception {
+        // ctx.fireUserEventTriggered(evt);
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Flushing idle Netty channel. MqttClientId = {}.", NettyUtils.clientID(ctx.channel()));
+        }
+        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/BugSnagErrorsHandler.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2012-2017 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 com.bugsnag.Bugsnag;
+import io.moquette.server.config.IConfig;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import static io.moquette.BrokerConstants.BUGSNAG_TOKEN_PROPERTY_NAME;
+
+@Sharable
+public class BugSnagErrorsHandler extends ChannelInboundHandlerAdapter {
+
+    private Bugsnag bugsnag;
+
+    public void init(IConfig props) {
+        final String token = props.getProperty(BUGSNAG_TOKEN_PROPERTY_NAME);
+        this.bugsnag = new Bugsnag(token);
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        bugsnag.notify(cause);
+        ctx.fireExceptionCaught(cause);
+    }
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) 2012-2017 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+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 MoquetteIdleTimeoutHandler extends ChannelDuplexHandler {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MoquetteIdleTimeoutHandler.class);
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleStateEvent) {
+            IdleState e = ((IdleStateEvent) evt).state();
+            if (e == IdleState.READER_IDLE) {
+                LOG.info("Firing channel inactive event. MqttClientId = {}.", NettyUtils.clientID(ctx.channel()));
+                // fire a channelInactive to trigger publish of Will
+                ctx.fireChannelInactive();
+                ctx.close();
+            }
+        } else {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug(
+                        "Firing Netty event. MqttClientId = {}, eventClass = {}.",
+                        NettyUtils.clientID(ctx.channel()),
+                        evt.getClass().getName());
+            }
+            super.userEventTriggered(ctx, evt);
+        }
+    }
+}

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

@@ -0,0 +1,403 @@
+/*
+ * Copyright (c) 2012-2017 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.moquette.BrokerConstants;
+import io.moquette.server.ServerAcceptor;
+import io.moquette.server.config.IConfig;
+import io.moquette.server.netty.metrics.*;
+import io.moquette.spi.impl.ProtocolProcessor;
+import io.moquette.spi.security.ISslContextCreator;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.*;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpRequestDecoder;
+import io.netty.handler.codec.http.HttpResponseEncoder;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import io.netty.handler.codec.mqtt.MqttDecoder;
+import io.netty.handler.codec.mqtt.MqttEncoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.concurrent.Future;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import static io.moquette.BrokerConstants.*;
+import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
+
+public class NettyAcceptor implements ServerAcceptor {
+
+    private static final String MQTT_SUBPROTOCOL_CSV_LIST = "mqtt, mqttv3.1, mqttv3.1.1";
+
+    static class WebSocketFrameToByteBufDecoder extends MessageToMessageDecoder<BinaryWebSocketFrame> {
+
+        @Override
+        protected void decode(ChannelHandlerContext chc, BinaryWebSocketFrame frame, List<Object> out)
+                throws Exception {
+            // convert the frame to a ByteBuf
+            ByteBuf bb = frame.content();
+            // System.out.println("WebSocketFrameToByteBufDecoder decode - " +
+            // ByteBufUtil.hexDump(bb));
+            bb.retain();
+            out.add(bb);
+        }
+    }
+
+    static class ByteBufToWebSocketFrameEncoder extends MessageToMessageEncoder<ByteBuf> {
+
+        @Override
+        protected void encode(ChannelHandlerContext chc, ByteBuf bb, List<Object> out) throws Exception {
+            // convert the ByteBuf to a WebSocketFrame
+            BinaryWebSocketFrame result = new BinaryWebSocketFrame();
+            // System.out.println("ByteBufToWebSocketFrameEncoder encode - " +
+            // ByteBufUtil.hexDump(bb));
+            result.content().writeBytes(bb);
+            out.add(result);
+        }
+    }
+
+    abstract class PipelineInitializer {
+
+        abstract void init(ChannelPipeline pipeline) throws Exception;
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(NettyAcceptor.class);
+
+    EventLoopGroup m_bossGroup;
+    EventLoopGroup m_workerGroup;
+    BytesMetricsCollector m_bytesMetricsCollector = new BytesMetricsCollector();
+    MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector();
+    private Optional<? extends ChannelInboundHandler> metrics;
+    private Optional<? extends ChannelInboundHandler> errorsCather;
+
+    private int nettySoBacklog;
+    private boolean nettySoReuseaddr;
+    private boolean nettyTcpNodelay;
+    private boolean nettySoKeepalive;
+    private int nettyChannelTimeoutSeconds;
+
+    private Class<? extends ServerSocketChannel> channelClass;
+
+    @Override
+    public void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator)
+            throws IOException {
+        LOG.info("Initializing Netty acceptor...");
+
+        nettySoBacklog = Integer.parseInt(props.getProperty(BrokerConstants.NETTY_SO_BACKLOG_PROPERTY_NAME, "128"));
+        nettySoReuseaddr = Boolean
+                .parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_REUSEADDR_PROPERTY_NAME, "true"));
+        nettyTcpNodelay = Boolean
+                .parseBoolean(props.getProperty(BrokerConstants.NETTY_TCP_NODELAY_PROPERTY_NAME, "true"));
+        nettySoKeepalive = Boolean
+                .parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_KEEPALIVE_PROPERTY_NAME, "true"));
+        nettyChannelTimeoutSeconds = Integer
+                .parseInt(props.getProperty(BrokerConstants.NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME, "10"));
+
+        boolean epoll = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, "false"));
+        if (epoll) {
+            // 由于目前只支持TCP MQTT， 所以bossGroup的线程数配置为1
+            LOG.info("Netty is using Epoll");
+            m_bossGroup = new EpollEventLoopGroup(1);
+            m_workerGroup = new EpollEventLoopGroup();
+            channelClass = EpollServerSocketChannel.class;
+        } else {
+            LOG.info("Netty is using NIO");
+            m_bossGroup = new NioEventLoopGroup(1);
+            m_workerGroup = new NioEventLoopGroup();
+            channelClass = NioServerSocketChannel.class;
+        }
+
+        final NettyMQTTHandler mqttHandler = new NettyMQTTHandler(processor);
+
+        final boolean useFineMetrics = Boolean.parseBoolean(props.getProperty(METRICS_ENABLE_PROPERTY_NAME, "false"));
+        if (useFineMetrics) {
+            DropWizardMetricsHandler metricsHandler = new DropWizardMetricsHandler();
+            metricsHandler.init(props);
+            this.metrics = Optional.of(metricsHandler);
+        } else {
+            this.metrics = Optional.empty();
+        }
+
+        final boolean useBugSnag = Boolean.parseBoolean(props.getProperty(BUGSNAG_ENABLE_PROPERTY_NAME, "false"));
+        if (useBugSnag) {
+            BugSnagErrorsHandler bugSnagHandler = new BugSnagErrorsHandler();
+            bugSnagHandler.init(props);
+            this.errorsCather = Optional.of(bugSnagHandler);
+        } else {
+            this.errorsCather = Optional.empty();
+        }
+        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) {
+            SSLContext sslContext = sslCtxCreator.initSSLContext();
+            if (sslContext == null) {
+                LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
+                return;
+            }
+            initializeSSLTCPTransport(mqttHandler, props, sslContext);
+            initializeWSSTransport(mqttHandler, props, sslContext);
+        }
+    }
+
+    private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) {
+        LOG.info("Initializing server. Protocol={}", protocol);
+        ServerBootstrap b = new ServerBootstrap();
+        b.group(m_bossGroup, m_workerGroup).channel(channelClass)
+                .childHandler(new ChannelInitializer<SocketChannel>() {
+
+                    @Override
+                    public void initChannel(SocketChannel ch) throws Exception {
+                        ChannelPipeline pipeline = ch.pipeline();
+                        try {
+                            pipeliner.init(pipeline);
+                        } catch (Throwable th) {
+                            LOG.error("Severe error during pipeline creation", th);
+                            throw th;
+                        }
+                    }
+                })
+            .option(ChannelOption.SO_BACKLOG, 10240) // 服务端可连接队列大小
+            .option(ChannelOption.SO_BACKLOG, nettySoBacklog)
+            .option(ChannelOption.SO_REUSEADDR, nettySoReuseaddr)
+            .childOption(ChannelOption.TCP_NODELAY, nettyTcpNodelay)
+            .childOption(ChannelOption.SO_KEEPALIVE, nettySoKeepalive);
+        try {
+            LOG.info("Binding server. host={}, port={}", host, port);
+            // Bind and start to accept incoming connections.
+            ChannelFuture f = b.bind(host, port);
+            LOG.info("Server has been bound. host={}, port={}", host, port);
+            f.sync().addListener(FIRE_EXCEPTION_ON_FAILURE);
+        } catch (InterruptedException ex) {
+            LOG.error("An interruptedException was caught while initializing server. Protocol={}", protocol, ex);
+        }
+    }
+
+    private void initializePlainTCPTransport(final NettyMQTTHandler handler,
+                                             IConfig props) throws IOException {
+        LOG.info("Configuring TCP MQTT transport");
+        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+        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);
+            return;
+        }
+        int port = Integer.parseInt(tcpPortProp);
+        initFactory(host, port, "TCP MQTT", new PipelineInitializer() {
+
+            @Override
+            void init(ChannelPipeline 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("decoder", new MqttDecoder());
+                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);
+            }
+        });
+    }
+
+    private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
+        LOG.info("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);
+            return;
+        }
+        int port = Integer.parseInt(webSocketPortProp);
+
+        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+
+        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+        initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
+
+            @Override
+            void init(ChannelPipeline pipeline) {
+                pipeline.addLast(new HttpServerCodec());
+                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
+                pipeline.addLast("webSocketHandler",
+                        new WebSocketServerProtocolHandler("/org/fusesource/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("decoder", new MqttDecoder());
+                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+                pipeline.addLast("handler", handler);
+            }
+        });
+    }
+
+    private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext)
+            throws IOException {
+        LOG.info("Configuring SSL MQTT transport");
+        String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+        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);
+            return;
+        }
+
+        int sslPort = Integer.parseInt(sslPortProp);
+        LOG.info("Starting SSL on port {}", sslPort);
+
+        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+        String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
+        final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
+        initFactory(host, sslPort, "SSL MQTT", new PipelineInitializer() {
+
+            @Override
+            void init(ChannelPipeline pipeline) throws Exception {
+                pipeline.addLast("ssl", createSslHandler(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("decoder", new MqttDecoder());
+                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+                pipeline.addLast("handler", handler);
+            }
+        });
+    }
+
+    private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext)
+            throws IOException {
+        LOG.info("Configuring secure websocket MQTT transport");
+        String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+        if (DISABLED_PORT_BIND.equals(sslPortProp)) {
+            // Do nothing no SSL configured
+            LOG.info("Property {} has been set to {}. Secure websocket MQTT will be disabled",
+                BrokerConstants.WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+            return;
+        }
+        int sslPort = Integer.parseInt(sslPortProp);
+        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+        String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
+        final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
+        initFactory(host, sslPort, "Secure websocket", new PipelineInitializer() {
+
+            @Override
+            void init(ChannelPipeline pipeline) throws Exception {
+                pipeline.addLast("ssl", createSslHandler(sslContext, needsClientAuth));
+                pipeline.addLast("httpEncoder", new HttpResponseEncoder());
+                pipeline.addLast("httpDecoder", new HttpRequestDecoder());
+                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
+                pipeline.addLast("webSocketHandler",
+                        new WebSocketServerProtocolHandler("/org/fusesource/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("decoder", new MqttDecoder());
+                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+                pipeline.addLast("handler", handler);
+            }
+        });
+    }
+
+    public void close() {
+        LOG.info("Closing Netty acceptor...");
+        if (m_workerGroup == null || m_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();
+
+        /*
+         * We shouldn't raise an IllegalStateException if we are interrupted. If we did so, the
+         * broker is not shut down properly.
+         */
+        LOG.info("Waiting for worker and boss event loop groups to terminate...");
+        try {
+            workerWaiter.await(10, TimeUnit.SECONDS);
+            bossWaiter.await(10, TimeUnit.SECONDS);
+        } catch (InterruptedException iex) {
+            LOG.warn("An InterruptedException was caught while waiting for event loops to terminate...");
+        }
+
+        if (!m_workerGroup.isTerminated()) {
+            LOG.warn("Forcing shutdown of worker event loop...");
+            m_workerGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS);
+        }
+
+        if (!m_bossGroup.isTerminated()) {
+            LOG.warn("Forcing shutdown of boss event loop...");
+            m_bossGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS);
+        }
+
+        LOG.info("Collecting message metrics...");
+        MessageMetrics metrics = m_metricsCollector.computeMetrics();
+        LOG.info("Metrics have been collected. Read messages={}, written messages={}", metrics.messagesRead(),
+            metrics.messagesWrote());
+
+        LOG.info("Collecting bytes metrics...");
+        BytesMetrics bytesMetrics = m_bytesMetricsCollector.computeMetrics();
+        LOG.info("Bytes metrics have been collected. Read bytes={}, written bytes={}", bytesMetrics.readBytes(),
+            bytesMetrics.wroteBytes());
+    }
+
+    private ChannelHandler createSslHandler(SSLContext sslContext, boolean needsClientAuth) {
+        SSLEngine sslEngine = sslContext.createSSLEngine();
+        sslEngine.setUseClientMode(false);
+        if (needsClientAuth) {
+            sslEngine.setNeedClientAuth(true);
+        }
+        return new SslHandler(sslEngine);
+    }
+}

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

@@ -0,0 +1,129 @@
+/*
+ * Copyright (c) 2012-2017 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.moquette.spi.impl.ProtocolProcessor;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.mqtt.*;
+import io.netty.util.ReferenceCountUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
+
+@Sharable
+public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
+
+    private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
+    private final ProtocolProcessor m_processor;
+
+    public NettyMQTTHandler(ProtocolProcessor processor) {
+        m_processor = processor;
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object message) {
+        try {
+            if (!(message instanceof MqttMessage)) {
+                LOG.error("Unknown mqtt message type {}, {}", message.getClass().getName(), message);
+                return;
+            }
+            MqttMessage msg = (MqttMessage) message;
+            MqttMessageType messageType = msg.fixedHeader().messageType();
+            LOG.info("Processing MQTT message, type={}", messageType);
+
+            switch (messageType) {
+                case CONNECT:
+                    m_processor.processConnect(ctx.channel(), (MqttConnectMessage) msg);
+                    break;
+                case SUBSCRIBE:
+                    m_processor.processSubscribe(ctx.channel(), (MqttSubscribeMessage) msg);
+                    break;
+                case UNSUBSCRIBE:
+                    m_processor.processUnsubscribe(ctx.channel(), (MqttUnsubscribeMessage) msg);
+                    break;
+                case PUBLISH:
+                    m_processor.processPublish(ctx.channel(), (MqttPublishMessage) msg);
+                    break;
+                case PUBREC:
+                    m_processor.processPubRec(ctx.channel(), msg);
+                    break;
+                case PUBCOMP:
+                    m_processor.processPubComp(ctx.channel(), msg);
+                    break;
+                case PUBREL:
+                    m_processor.processPubRel(ctx.channel(), msg);
+                    break;
+                case DISCONNECT:
+                    m_processor.processDisconnect(ctx.channel(), msg.fixedHeader().isDup());
+                    break;
+                case PUBACK:
+                    m_processor.processPubAck(ctx.channel(), (MqttPubAckMessage) msg);
+                    break;
+                case PINGREQ:
+                    MqttFixedHeader pingHeader = new MqttFixedHeader(
+                            MqttMessageType.PINGRESP,
+                            false,
+                            AT_MOST_ONCE,
+                            false,
+                            0);
+                    MqttMessage pingResp = new MqttMessage(pingHeader);
+                    ctx.writeAndFlush(pingResp);
+                    break;
+                default:
+                    LOG.error("Unkonwn MessageType:{}", messageType);
+                    break;
+            }
+        } catch (Throwable ex) {
+            LOG.error("Exception was caught while processing MQTT message, " + ex.getCause(), ex);
+            ctx.fireExceptionCaught(ex);
+            ctx.close();
+        } finally {
+            ReferenceCountUtil.release(message);
+        }
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        String clientID = NettyUtils.clientID(ctx.channel());
+        if (clientID != null && !clientID.isEmpty()) {
+            LOG.info("Notifying connection lost event. MqttClientId = {}.", clientID);
+            m_processor.processConnectionLost(clientID, ctx.channel());
+        }
+        ctx.close();
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        LOG.error("An unexpected exception was caught while processing MQTT message. Closing Netty channel. CId={}, " +
+            "cause={}, errorMessage={}", NettyUtils.clientID(ctx.channel()), cause.getCause(), cause.getMessage());
+        for (StackTraceElement ste : cause.getStackTrace()) {
+            LOG.error(ste.toString());
+        }
+        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/server/netty/NettyUtils.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright (c) 2012-2017 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.moquette.server.Constants;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+
+/**
+ * Some Netty's channels utilities.
+ */
+public final class NettyUtils {
+
+    public static final String ATTR_USERNAME = "username";
+    public static final String ATTR_SESSION_STOLEN = "sessionStolen";
+    public static final String ATTR_CHANNEL_STATUS = "channelStatus";
+
+    private static final AttributeKey<Object> ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE);
+    private static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION);
+    private static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID);
+    private static final AttributeKey<Object> ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME);
+
+    public static Object getAttribute(ChannelHandlerContext ctx, AttributeKey<Object> key) {
+        Attribute<Object> attr = ctx.channel().attr(key);
+        return attr.get();
+    }
+
+    public static void keepAlive(Channel channel, int keepAlive) {
+        channel.attr(NettyUtils.ATTR_KEY_KEEPALIVE).set(keepAlive);
+    }
+
+    public static void cleanSession(Channel channel, boolean cleanSession) {
+        channel.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(cleanSession);
+    }
+
+    public static boolean cleanSession(Channel channel) {
+        return (Boolean) channel.attr(NettyUtils.ATTR_KEY_CLEANSESSION).get();
+    }
+
+    public static void clientID(Channel channel, String clientID) {
+        channel.attr(NettyUtils.ATTR_KEY_CLIENTID).set(clientID);
+    }
+
+    public static String clientID(Channel channel) {
+        return (String) channel.attr(NettyUtils.ATTR_KEY_CLIENTID).get();
+    }
+
+    public static void userName(Channel channel, String username) {
+        channel.attr(NettyUtils.ATTR_KEY_USERNAME).set(username);
+    }
+
+    public static String userName(Channel channel) {
+        return (String) channel.attr(NettyUtils.ATTR_KEY_USERNAME).get();
+    }
+
+    private NettyUtils() {
+    }
+}

broker/src/main/java/io/moquette/server/netty/metrics/BytesMetrics.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2012-2017 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.metrics;
+
+public class BytesMetrics {
+
+    private long m_readBytes;
+    private long m_wroteBytes;
+
+    void incrementRead(long numBytes) {
+        m_readBytes += numBytes;
+    }
+
+    void incrementWrote(long numBytes) {
+        m_wroteBytes += numBytes;
+    }
+
+    public long readBytes() {
+        return m_readBytes;
+    }
+
+    public long wroteBytes() {
+        return m_wroteBytes;
+    }
+}

broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsCollector.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) 2012-2017 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.metrics;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Collects all the metrics from the various pipeline.
+ */
+public class BytesMetricsCollector {
+
+    private AtomicLong readBytes = new AtomicLong();
+    private AtomicLong wroteBytes = new AtomicLong();
+
+    public BytesMetrics computeMetrics() {
+        BytesMetrics allMetrics = new BytesMetrics();
+        allMetrics.incrementRead(readBytes.get());
+        allMetrics.incrementWrote(wroteBytes.get());
+        return allMetrics;
+    }
+
+    public void sumReadBytes(long count) {
+        readBytes.getAndAdd(count);
+    }
+
+    public void sumWroteBytes(long count) {
+        wroteBytes.getAndAdd(count);
+    }
+}

broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2012-2017 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.metrics;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+
+public class BytesMetricsHandler extends ChannelDuplexHandler {
+
+    private static final AttributeKey<BytesMetrics> ATTR_KEY_METRICS = AttributeKey.valueOf("BytesMetrics");
+
+    private BytesMetricsCollector m_collector;
+
+    public BytesMetricsHandler(BytesMetricsCollector collector) {
+        m_collector = collector;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        Attribute<BytesMetrics> attr = ctx.channel().attr(ATTR_KEY_METRICS);
+        attr.set(new BytesMetrics());
+
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
+        metrics.incrementRead(((ByteBuf) msg).readableBytes());
+        ctx.fireChannelRead(msg);
+    }
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
+        metrics.incrementWrote(((ByteBuf) msg).writableBytes());
+        ctx.write(msg, promise);
+    }
+
+    @Override
+    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+        BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
+        m_collector.sumReadBytes(metrics.readBytes());
+        m_collector.sumWroteBytes(metrics.wroteBytes());
+        super.close(ctx, promise);
+    }
+
+    public static BytesMetrics getBytesMetrics(Channel channel) {
+        return channel.attr(ATTR_KEY_METRICS).get();
+    }
+}

broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright (c) 2012-2017 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.metrics;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+import com.librato.metrics.reporter.Librato;
+import io.moquette.server.config.IConfig;
+import io.moquette.server.netty.NettyUtils;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessageType;
+
+import java.util.concurrent.TimeUnit;
+
+import static io.moquette.BrokerConstants.*;
+import static io.netty.channel.ChannelHandler.Sharable;
+
+/**
+ * Pipeline handler use to track some MQTT metrics.
+ */
+@Sharable
+public final class DropWizardMetricsHandler extends ChannelInboundHandlerAdapter {
+    private MetricRegistry metrics;
+    private Meter publishesMetrics;
+    private Meter subscribeMetrics;
+    private Counter connectedClientsMetrics;
+
+    public void init(IConfig props) {
+        this.metrics = new MetricRegistry();
+        this.publishesMetrics = metrics.meter("publish.requests");
+        this.subscribeMetrics = metrics.meter("subscribe.requests");
+        this.connectedClientsMetrics = metrics.counter("connect.num_clients");
+//        ConsoleReporter reporter = ConsoleReporter.forRegistry(metrics)
+//            .convertRatesTo(TimeUnit.SECONDS)
+//            .convertDurationsTo(TimeUnit.MILLISECONDS)
+//            .build();
+//        reporter.start(1, TimeUnit.MINUTES);
+        final String email = props.getProperty(METRICS_LIBRATO_EMAIL_PROPERTY_NAME);
+        final String token = props.getProperty(METRICS_LIBRATO_TOKEN_PROPERTY_NAME);
+        final String source = props.getProperty(METRICS_LIBRATO_SOURCE_PROPERTY_NAME);
+
+        Librato.reporter(this.metrics, email, token)
+            .setSource(source)
+            .start(10, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object message) {
+        MqttMessage msg = (MqttMessage) message;
+        MqttMessageType messageType = msg.fixedHeader().messageType();
+        switch (messageType) {
+            case PUBLISH:
+                this.publishesMetrics.mark();
+                break;
+            case SUBSCRIBE:
+                this.subscribeMetrics.mark();
+                break;
+            case CONNECT:
+                this.connectedClientsMetrics.inc();
+                break;
+            case DISCONNECT:
+                this.connectedClientsMetrics.dec();
+                break;
+            default:
+                break;
+        }
+        ctx.fireChannelRead(message);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        String clientID = NettyUtils.clientID(ctx.channel());
+        if (clientID != null && !clientID.isEmpty()) {
+            this.connectedClientsMetrics.dec();
+        }
+        ctx.fireChannelInactive();
+    }
+
+}

broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2012-2017 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.metrics;
+
+import io.moquette.server.netty.NettyUtils;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.mqtt.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static io.moquette.spi.impl.Utils.messageId;
+
+/**
+ *
+ * @author andrea
+ */
+@Sharable
+public class MQTTMessageLogger extends ChannelDuplexHandler {
+
+    private static final Logger LOG = LoggerFactory.getLogger("messageLogger");
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object message) {
+        logMQTTMessage(ctx, message, "C->B");
+        ctx.fireChannelRead(message);
+    }
+
+    private void logMQTTMessage(ChannelHandlerContext ctx, Object message, String direction) {
+        if (!(message instanceof MqttMessage)) {
+            return;
+        }
+        MqttMessage msg = (MqttMessage) message;
+        String clientID = NettyUtils.clientID(ctx.channel());
+        MqttMessageType messageType = msg.fixedHeader().messageType();
+        switch (messageType) {
+            case CONNECT:
+            case CONNACK:
+            case PINGREQ:
+            case PINGRESP:
+            case DISCONNECT:
+                LOG.info("{} {} <{}>", direction, messageType, clientID);
+                break;
+            case SUBSCRIBE:
+                MqttSubscribeMessage subscribe = (MqttSubscribeMessage) msg;
+                LOG.info("{} SUBSCRIBE <{}> to topics {}", direction, clientID,
+                    subscribe.payload().topicSubscriptions());
+                break;
+            case UNSUBSCRIBE:
+                MqttUnsubscribeMessage unsubscribe = (MqttUnsubscribeMessage) msg;
+                LOG.info("{} UNSUBSCRIBE <{}> to topics <{}>", direction, clientID, unsubscribe.payload().topics());
+                break;
+            case PUBLISH:
+                MqttPublishMessage publish = (MqttPublishMessage) msg;
+                LOG.info("{} PUBLISH <{}> to topics <{}>", direction, clientID, publish.variableHeader().topicName());
+                break;
+            case PUBREC:
+            case PUBCOMP:
+            case PUBREL:
+            case PUBACK:
+            case UNSUBACK:
+                LOG.info("{} {} <{}> packetID <{}>", direction, messageType, clientID, messageId(msg));
+                break;
+            case SUBACK:
+                MqttSubAckMessage suback = (MqttSubAckMessage) msg;
+                final List<Integer> grantedQoSLevels = suback.payload().grantedQoSLevels();
+                LOG.info("{} SUBACK <{}> packetID <{}>, grantedQoses {}", direction, clientID, messageId(msg),
+                    grantedQoSLevels);
+                break;
+        }
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        String clientID = NettyUtils.clientID(ctx.channel());
+        if (clientID != null && !clientID.isEmpty()) {
+            LOG.info("Channel closed <{}>", clientID);
+        }
+        ctx.fireChannelInactive();
+    }
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        logMQTTMessage(ctx, msg, "C<-B");
+        ctx.write(msg, promise);
+    }
+}

broker/src/main/java/io/moquette/server/netty/metrics/MessageMetrics.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2012-2017 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.metrics;
+
+public class MessageMetrics {
+
+    private long m_messagesRead;
+    private long m_messageWrote;
+
+    void incrementRead(long numMessages) {
+        m_messagesRead += numMessages;
+    }
+
+    void incrementWrote(long numMessages) {
+        m_messageWrote += numMessages;
+    }
+
+    public long messagesRead() {
+        return m_messagesRead;
+    }
+
+    public long messagesWrote() {
+        return m_messageWrote;
+    }
+}

broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsCollector.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) 2012-2017 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.metrics;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Collects all the metrics from the various pipeline.
+ */
+public class MessageMetricsCollector {
+
+    private AtomicLong readMsgs = new AtomicLong();
+    private AtomicLong wroteMsgs = new AtomicLong();
+
+    public MessageMetrics computeMetrics() {
+        MessageMetrics allMetrics = new MessageMetrics();
+        allMetrics.incrementRead(readMsgs.get());
+        allMetrics.incrementWrote(wroteMsgs.get());
+        return allMetrics;
+    }
+
+    public void sumReadMessages(long count) {
+        readMsgs.getAndAdd(count);
+    }
+
+    public void sumWroteMessages(long count) {
+        wroteMsgs.getAndAdd(count);
+    }
+}

broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2012-2017 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.metrics;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+
+public class MessageMetricsHandler extends ChannelDuplexHandler {
+
+    private static final AttributeKey<MessageMetrics> ATTR_KEY_METRICS = AttributeKey.valueOf("MessageMetrics");
+
+    private MessageMetricsCollector m_collector;
+
+    public MessageMetricsHandler(MessageMetricsCollector collector) {
+        m_collector = collector;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        Attribute<MessageMetrics> attr = ctx.channel().attr(ATTR_KEY_METRICS);
+        attr.set(new MessageMetrics());
+
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
+        metrics.incrementRead(1);
+        ctx.fireChannelRead(msg);
+    }
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
+        metrics.incrementWrote(1);
+        ctx.write(msg, promise);
+    }
+
+    @Override
+    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+        MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
+        m_collector.sumReadMessages(metrics.messagesRead());
+        m_collector.sumWroteMessages(metrics.messagesWrote());
+        super.close(ctx, promise);
+    }
+
+    public static MessageMetrics getMessageMetrics(Channel channel) {
+        return channel.attr(ATTR_KEY_METRICS).get();
+    }
+}