Switched to netty 4.1.6 to include Netty's official mqtt codec to make some performance tests

perf/pom.xml

@@ -54,6 +54,11 @@
             <artifactId>slf4j-log4j12</artifactId>
             <version>1.6.4</version>
         </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-codec-mqtt</artifactId>
+            <version>${netty.version}</version>
+        </dependency>
 
         <dependency>
             <groupId>org.hdrhistogram</groupId>

perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java

@@ -91,7 +91,7 @@ class  BenchmarkPublisher {
     }
 
     public void firePublishes() throws MqttException {
-        long pauseMicroseconds = (1 / this.messagesPerSecond) * 1000 * 1000;
+        long pauseMicroseconds = (int)((1.0 / messagesPerSecond) * 1000 * 1000);
         LOG.info("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds);
 
         LOG.info("PUB: publishing..");

perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java

@@ -0,0 +1,80 @@
+package io.moquette.parser.netty.performance;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandler;
+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 io.netty.handler.codec.mqtt.MqttPublishMessage;
+import org.HdrHistogram.Histogram;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+@ChannelHandler.Sharable
+class NettyPublishReceiverHandler extends ChannelInboundHandlerAdapter {
+
+    private static final Logger LOG = LoggerFactory.getLogger(NettyPublishReceiverHandler.class);
+    Histogram forthNetworkTime = new Histogram(5);
+
+    public NettyPublishReceiverHandler() {
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object message) {
+        MqttMessage msg = (MqttMessage) message;
+        MqttMessageType type = msg.fixedHeader().messageType();
+
+        try {
+            switch (type) {
+                case PUBLISH:
+                    LOG.info("Received a message of type {}", type);
+                    handlePublish((MqttPublishMessage) msg);
+                    return;
+                default:
+                    LOG.info("Received a message of type {}",type);
+            }
+        } catch (Exception ex) {
+            LOG.error("Bad error in processing the message", ex);
+        }
+    }
+
+    private void handlePublish(MqttPublishMessage msg) {
+        long start = System.nanoTime();
+        LOG.debug("push forward message the topic {}", msg.variableHeader().topicName());
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("content <{}>", payload2Str(msg.content()));
+        }
+        String decodedPayload = payload2Str(msg.content());
+        long sentTime = Long.parseLong(decodedPayload.split("-")[1]);
+        forthNetworkTime.recordValue(start - sentTime);
+
+        long stop = System.nanoTime();
+        LOG.info("Request processed in {} ns, matching {}", stop - start, decodedPayload);
+    }
+
+    static String payload2Str(ByteBuf content) {
+        byte[] rawBytes;
+        if (content.hasArray()) {
+            rawBytes = content.array();
+        } else {
+            int size = content.readableBytes();
+            rawBytes = new byte[size];
+            content.getBytes(content.readerIndex(), rawBytes);
+        }
+        return new String(rawBytes);
+    }
+
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        LOG.info("Received channel inactive");
+        ctx.channel().close();
+
+        System.out.println("Network time histogram (microsecs)");
+        this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0);
+    }
+
+}

perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java

@@ -7,11 +7,9 @@ import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.timeout.IdleState;
-import io.netty.handler.timeout.IdleStateEvent;
-import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
+import io.netty.handler.codec.mqtt.MqttDecoder;
+import io.netty.handler.codec.mqtt.MqttEncoder;
 import org.eclipse.paho.client.mqttv3.MqttException;
-import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,6 +42,9 @@ public class ProtocolPublishDecodingServer {
                             pipeline.addLast("decoder", new MQTTDecoder());
                             pipeline.addLast("encoder", new MQTTEncoder());
                             pipeline.addLast("handler", new PublishReceiverHandler());
+//                            pipeline.addLast("decoder", new MqttDecoder());
+//                            pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+//                            pipeline.addLast("handler", new NettyPublishReceiverHandler());
                         } catch (Throwable th) {
                             LOG.error("Severe error during pipeline creation", th);
                             throw th;
@@ -87,6 +88,8 @@ public class ProtocolPublishDecodingServer {
         publishBombing();
         publishBombing();
         publishBombing();
+        publishBombing();
+        publishBombing();
         //Bind  a shutdown hook
         Runtime.getRuntime().addShutdownHook(new Thread() {
             @Override
@@ -99,7 +102,7 @@ public class ProtocolPublishDecodingServer {
     private static void publishBombing() {
         LOG.info("Started publish loop");
         PublishBomber heavyPublisher = new PublishBomber("localhost", 1883);
-        heavyPublisher.publishLoop(10000, 10000);
+        heavyPublisher.publishLoop(25000, 100000);
         heavyPublisher.disconnect();
         LOG.info("Finished publish loop");
     }

perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java

@@ -57,8 +57,8 @@ class PublishBomber {
 
 
     public void publishLoop(int messagesPerSecond, int numToSend) {
-        long pauseMicroseconds = (1 / messagesPerSecond) * 1000 * 1000;
-        LOG.info("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds);
+        long pauseMicroseconds = (int)((1.0 / messagesPerSecond) * 1000 * 1000);
+        LOG.warn("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds);
 
         LOG.info("PUB: publishing..");
         final long startTime = System.currentTimeMillis();

pom.xml

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <netty.version>4.0.33.Final</netty.version>
+        <netty.version>4.1.6.Final</netty.version>
         <source.version>1.7</source.version>
         <target.version>1.7</target.version>
         <bintray.repo>andsel/maven</bintray.repo>