Embedded pub and sub client into same process of the plain test broker to use same nanosecs

broker/src/test/resources/log4j.properties

@@ -45,12 +45,12 @@ log4j.appender.file.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %L
 #   Message Logger Configuration   #
 #####################################
 log4j.appender.messagelog=org.apache.log4j.RollingFileAppender
-log4j.appender.messagelog.Threshold=INFO
+log4j.appender.messagelog.Threshold=WARN
 log4j.appender.messagelog.File=moquette_messages.log
 log4j.appender.messagelog.MaxFileSize=100MB
 log4j.appender.messagelog.MaxBackupIndex=1
 log4j.appender.messagelog.layout=org.apache.log4j.PatternLayout
 log4j.appender.messagelog.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %L %x - %m%n
 
-log4j.category.messageLogger=INFO, messagelog
+log4j.category.messageLogger=WARN, messagelog
-log4j.additivity.messageLogger=false
+#log4j.additivity.messageLogger=false

netty_parser/pom.xml

@@ -64,6 +64,33 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.hdrhistogram</groupId>
+            <artifactId>HdrHistogram</artifactId>
+            <version>2.1.2</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.1.0</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.eclipse.jetty.toolchain</groupId>
+            <artifactId>jetty-perf-helper</artifactId>
+            <version>1.0.5</version>
+            <scope>test</scope>
+        </dependency>
+        <!--<dependency>-->
+            <!--<groupId>org.eclipse.paho</groupId>-->
+            <!--<artifactId>org.eclipse.paho.client.mqttv3</artifactId>-->
+            <!--<version>1.0.2</version>-->
+            <!--<scope>test</scope>-->
+        <!--</dependency>-->
+
     </dependencies>
     
     <build>

netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java

@@ -0,0 +1,124 @@
+package io.moquette.parser.netty.performance;
+
+
+import org.eclipse.jetty.toolchain.perf.PlatformTimer;
+import org.eclipse.paho.client.mqttv3.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CountDownLatch;
+
+class  BenchmarkPublisher {
+
+    private static final Logger LOG = LoggerFactory.getLogger(BenchmarkPublisher.class);
+
+    static class PublishCallback implements IMqttActionListener {
+        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
+            LOG.error("PUB: Publish failed: " + exception);
+            exception.printStackTrace();
+            System.exit(2);
+        }
+
+        public void onSuccess(IMqttToken asyncActionToken) {
+        }
+    }
+
+    static class ExitTopicCallback implements IMqttActionListener {
+
+        private final long startedTime;
+        private final long numToSend;
+        private final IMqttAsyncClient connection;
+        private final CountDownLatch stopLatch;
+
+        ExitTopicCallback(long startedTime, long numToSend, IMqttAsyncClient connection, CountDownLatch stopLatch) {
+            this.startedTime = startedTime;
+            this.numToSend = numToSend;
+            this.connection = connection;
+            this.stopLatch = stopLatch;
+        }
+
+        public void onSuccess(IMqttToken asyncActionToken) {
+            long stopTime = System.currentTimeMillis();
+            long spentTime = stopTime - startedTime;
+            LOG.info("PUB: published in {} ms", spentTime);
+
+            try {
+                this.connection.disconnect();
+            } catch (MqttException mex) {
+                LOG.error("Disconnect error", mex);
+            }
+            double msgPerSec = (numToSend / spentTime) * 1000;
+            LOG.info( "PUB: speed {} msg/sec", msgPerSec);
+            this.stopLatch.countDown();
+        }
+
+        public void onFailure(IMqttToken asyncActionToken, Throwable th) {
+            LOG.error("PUB: Publish failed: " + th);
+            System.exit(2);
+        }
+    }
+
+    private int qos = 0;
+    private boolean retain = false;
+    private final int numToSend;
+    private final int messagesPerSecond;
+    private final String dialog_id;
+    private final IMqttAsyncClient client;
+    private CountDownLatch m_latch;
+
+    public BenchmarkPublisher(MqttAsyncClient client, int numToSend, int messagesPerSecond, String dialog_id) {
+        this.client = client;
+        this.numToSend = numToSend;
+        this.messagesPerSecond = messagesPerSecond;
+        this.dialog_id = dialog_id;
+    }
+
+    public void connect() throws MqttException {
+        MqttConnectOptions connectOptions = new MqttConnectOptions();
+        connectOptions.setCleanSession(true);
+        this.client.connect(connectOptions, null, new IMqttActionListener() {
+            public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
+                LOG.error("PUB: connect fail", exception);
+                System.exit(2);
+            }
+
+            public void onSuccess(IMqttToken asyncActionToken) {
+                LOG.info("PUB: Successfully connected to server");
+            }
+        });
+
+        this.m_latch = new CountDownLatch(1);
+    }
+
+    public void firePublishes() throws MqttException {
+        long pauseMicroseconds = (1 / this.messagesPerSecond) * 1000 * 1000;
+        LOG.info("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds);
+
+        LOG.info("PUB: publishing..");
+        CountDownLatch m_latch = new CountDownLatch(1);
+        final long startTime = System.currentTimeMillis();
+
+        //initialize the timer
+        PlatformTimer timer = PlatformTimer.detect();
+        IMqttActionListener pubCallback = new PublishCallback();
+//        println ""
+        for (int i=0; i < numToSend; i++) {
+            long nanos = System.nanoTime();
+            byte[] message = ("Hello world!!-" + nanos).getBytes();
+            this.client.publish("/topic" + dialog_id, message, qos, retain, null, pubCallback);
+//            print '.'
+            timer.sleep(pauseMicroseconds);
+        }
+
+        IMqttActionListener exitCallback = new ExitTopicCallback(startTime, numToSend, this.client, m_latch);
+        long nanosExit = System.nanoTime();
+        byte[] exitMessage = ("Hello world!!-" + nanosExit).getBytes();
+        this.client.publish("/exit" + dialog_id, exitMessage, qos, retain, null, exitCallback);
+    }
+
+
+    public void waitFinish() throws InterruptedException {
+        m_latch.await();
+    }
+
+}

netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java

@@ -0,0 +1,144 @@
+package io.moquette.parser.netty.performance;
+
+import org.HdrHistogram.Histogram;
+import org.eclipse.paho.client.mqttv3.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CountDownLatch;
+
+class BenchmarkSubscriber {
+
+    private static final Logger LOG = LoggerFactory.getLogger(BenchmarkSubscriber.class);
+
+    class SubscriberCallback implements MqttCallback {
+
+        public void messageArrived(String topic, MqttMessage message) throws Exception {
+            //used to catch the first message published
+            if (!alreadyStarted) {
+                startTime = System.currentTimeMillis();
+                alreadyStarted = true;
+            }
+
+            //println "Received a publish on topic ${topic}"
+            if (topic.startsWith("/exit")) {
+                LOG.info("SUB: ok Exit!");
+                exit = true;
+                LOG.debug("Before disconnect");
+//                client.disconnect(1000) //wait max 1 sec
+                //client.disconnectForcibly()
+                //client.close()
+                LOG.debug("After disconnect");
+                long stopTime = System.currentTimeMillis();
+                long spentTime = stopTime - startTime;
+                LOG.info( "SUB: {} received", topic);
+                LOG.info( "SUB: subscriber disconnected, received {} messages in {} ms", numReceived, spentTime);
+                double msgPerSec = (numReceived / spentTime) * 1000;
+                LOG.info( "SUB: Speed: {} msg/sec", msgPerSec);
+                LOG.info( "SUB: Latency diagram [microsecs]");
+                histogram.outputPercentileDistribution(System.out, 1000.0); //nanos/1000
+                m_latch.countDown();
+            } else {
+                String payload = new String(message.getPayload());
+                long sentTime = Long.parseLong(payload.split("-")[1]);
+                long delay = System.nanoTime() - sentTime;
+                histogram.recordValue(delay);
+//                print '+'
+                numReceived++;
+//        if ((numReceived % 10000) == 0) {
+//            print '.'
+//        }
+            }
+
+        }
+
+        public void connectionLost(Throwable cause) {
+            LOG.error("Client Subscriber lost connection, caused by ", cause);
+            try {
+                client.disconnect();
+            } catch (MqttException mex) {
+                LOG.error(null, mex);
+            }
+            LOG.error("SUB: Something failed during message reception");
+            m_latch.countDown();
+        }
+
+        public void deliveryComplete(IMqttDeliveryToken token) {
+
+        }
+
+    }
+
+    private int numReceived = 0;
+    private long startTime = System.currentTimeMillis();
+    private boolean exit = false;
+    private CountDownLatch m_latch = new CountDownLatch(1);
+    private boolean alreadyStarted = false;
+    private Histogram histogram = new Histogram(5);
+
+    private final IMqttAsyncClient client;
+    private final String dialog_id;
+
+    public BenchmarkSubscriber(MqttAsyncClient client, String dialog_id) {
+        this.client = client;
+        this.dialog_id = dialog_id;
+    }
+
+    public void connect() throws MqttException {
+        MqttConnectOptions connectOptions = new MqttConnectOptions();
+        connectOptions.setCleanSession(true);
+        this.client.setCallback(new SubscriberCallback());
+        this.client.connect(connectOptions, null, new IMqttActionListener() {
+
+            public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
+                LOG.error( "SUB: connect fail", exception);
+                System.exit(2);
+            }
+
+            public void onSuccess(IMqttToken asyncActionToken) {
+                try {
+                    client.subscribe("/topic" + dialog_id, 1, null, new IMqttActionListener() {
+                        public void onFailure(IMqttToken asyncActionToken2, Throwable exception) {
+                            try {
+                                client.disconnect();
+                            } catch (MqttException mex) {
+                                LOG.error(null, mex);
+                            }
+                        }
+
+                        public void onSuccess(IMqttToken asyncActionToken2) {
+                            LOG.info("SUB: subscribed to /topic{} qos: 1", dialog_id);
+                        }
+                    });
+                } catch (MqttException e) {
+                    e.printStackTrace();
+                }
+
+                try {
+                    client.subscribe("/exit" + dialog_id, 1, null, new IMqttActionListener() {
+                        public void onFailure(IMqttToken asyncActionToken2, Throwable exception) {
+                            try {
+                                client.disconnect();
+                            } catch (MqttException e) {
+                                e.printStackTrace();
+                            }
+                            m_latch.countDown();
+                        }
+
+                        public void onSuccess(IMqttToken asyncActionToken2) {
+                            LOG.info("SUB: subscribed to /exit{} qos: 1", dialog_id);
+                        }
+                    });
+                } catch (MqttException e) {
+                    e.printStackTrace();
+                }
+            }
+        });
+    }
+
+
+    public void waitFinish() throws InterruptedException {
+        this.m_latch.await();
+    }
+
+}

netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopHandler.java

@@ -2,10 +2,14 @@ package io.moquette.parser.netty.performance;
 
 import io.moquette.parser.proto.messages.*;
 import static io.moquette.parser.proto.messages.AbstractMessage.*;
+
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.moquette.parser.proto.Utils;
+import io.netty.util.AttributeKey;
+import org.HdrHistogram.Histogram;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -19,6 +23,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
 
     private static final Logger LOG = LoggerFactory.getLogger(LoopMQTTHandler.class);
     private ProtocolDecodingServer.SharedState m_state;
+    Histogram processingTime = new Histogram(5);
 
     public LoopMQTTHandler(ProtocolDecodingServer.SharedState state) {
         this.m_state = state;
@@ -27,19 +32,26 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object message) {
         AbstractMessage msg = (AbstractMessage) message;
-        LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
+        String clientID = clientID(ctx.channel());
+
         try {
             switch (msg.getMessageType()) {
                 case CONNECT:
-                    handleConnect(ctx, (ConnectMessage) message);
+                    ConnectMessage connect = (ConnectMessage) message;
+                    clientID = connect.getClientID();
+                    LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
+                    handleConnect(ctx, connect);
                     return;
                 case SUBSCRIBE:
+                    LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
                     handleSubscribe(ctx, (SubscribeMessage) msg);
                     return;
                 case PUBLISH:
+                    LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
                     handlePublish(ctx, (PublishMessage) msg);
                     return;
                 case DISCONNECT:
+                    LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
                     ctx.close();
 //                case PUBACK:
 //                    NettyChannel channel;
@@ -56,6 +68,8 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
                     PingRespMessage pingResp = new PingRespMessage();
                     ctx.writeAndFlush(pingResp);
                     break;
+                default:
+                    LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
             }
         } catch (Exception ex) {
             LOG.error("Bad error in processing the message", ex);
@@ -68,6 +82,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
             return;
         }
 
+        long start = System.nanoTime();
         LOG.debug("push forward message the topic {}", msg.getTopicName());
         if (LOG.isDebugEnabled()) {
             LOG.debug("content <{}>", payload2Str(msg.getPayload()));
@@ -80,6 +95,15 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
         pubMessage.setPayload(msg.getPayload());
 
         m_state.getSubscriberCh().writeAndFlush(pubMessage);
+        /*Channel subscriberCh = m_state.getSubscriberCh();
+        if (subscriberCh.isWritable()) {
+            subscriberCh.write(pubMessage);
+        } else {
+            subscriberCh.writeAndFlush(pubMessage);
+        }*/
+        long stop = System.nanoTime();
+        processingTime.recordValue(stop - start);
+        LOG.info("Request processed in {} ns, matching {}", stop - start, payload2Str(msg.getPayload()));
     }
 
     private void handleSubscribe(ChannelHandlerContext ctx, SubscribeMessage msg) {
@@ -101,10 +125,11 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
 
     private void handleConnect(ChannelHandlerContext ctx, ConnectMessage msg) {
         String clientID = msg.getClientID();
+        clientID(ctx.channel(), clientID);
         if (clientID.toLowerCase().startsWith("sub")) {
-            m_state.setSubscriberCh(ctx);
+            m_state.setSubscriberCh(ctx.channel());
         } else if (clientID.toLowerCase().startsWith("pub")) {
-            m_state.setPublisherCh(ctx);
+            m_state.setPublisherCh(ctx.channel());
         } else {
             //we don't admit other names
             ConnAckMessage koResp = new ConnAckMessage();
@@ -120,12 +145,35 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        LOG.info("Received channel inactive");
 //        NettyChannel channel = m_channelMapper.get(ctx);
 //        String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
 //        m_messaging.lostConnection(channel, clientID);
-//        ctx.close(/*false*/);
+        ctx.channel().close(/*false*/);
+        System.out.println("Processing histogram (microsecs)");
+        this.processingTime.outputPercentileDistribution(System.out, 1000.0);
 //        synchronized (m_channelMapper) {
 //            m_channelMapper.remove(ctx);
 //        }
     }
+
+    @Override
+    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+        if (ctx.channel().isWritable()) {
+            String clientID = clientID(ctx.channel());
+            LOG.info("Channel for client <{}> is writable again", clientID);
+            ctx.channel().flush();
+        }
+        ctx.fireChannelWritabilityChanged();
+    }
+
+    private static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf("ClientID");
+
+    private static void clientID(Channel channel, String clientID) {
+        channel.attr(ATTR_KEY_CLIENTID).set(clientID);
+    }
+
+    private static String clientID(Channel channel) {
+        return (String) channel.attr(ATTR_KEY_CLIENTID).get();
+    }
 }

netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java

@@ -11,6 +11,9 @@ import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,8 +46,8 @@ public class ProtocolDecodingServer {
     EventLoopGroup m_workerGroup;
 
     static final class SharedState {
-        private volatile ChannelHandlerContext subscriberCh;
+        private volatile Channel subscriberCh;
-        private volatile ChannelHandlerContext publisherCh;
+        private volatile Channel publisherCh;
         private volatile boolean forwardPublish = false;
 
         public boolean isForwardable() {
@@ -55,15 +58,15 @@ public class ProtocolDecodingServer {
             this.forwardPublish = forwardPublish;
         }
 
-        public void setSubscriberCh(ChannelHandlerContext subscriberCh) {
+        public void setSubscriberCh(Channel subscriberCh) {
             this.subscriberCh = subscriberCh;
         }
 
-        public ChannelHandlerContext getSubscriberCh() {
+        public Channel getSubscriberCh() {
             return subscriberCh;
         }
 
-        public void setPublisherCh(ChannelHandlerContext publisherCh) {
+        public void setPublisherCh(Channel publisherCh) {
             this.publisherCh = publisherCh;
         }
     }
@@ -124,11 +127,13 @@ public class ProtocolDecodingServer {
         System.out.println("Server stopped");
     }
 
-    public static final void main(String[] args) {
+    public static void main(String[] args) throws MqttException, InterruptedException {
         final ProtocolDecodingServer server = new ProtocolDecodingServer();
         server.init();
 
         System.out.println("Loop server started");
+        startClientsTesting();
+        System.out.println("Started clients (sub/pub) for testing");
         //Bind  a shutdown hook
         Runtime.getRuntime().addShutdownHook(new Thread() {
             @Override
@@ -137,4 +142,27 @@ public class ProtocolDecodingServer {
             }
         });
     }
+
+    private static void startClientsTesting() throws MqttException, InterruptedException {
+        String host = "localhost";
+        int numToSend = 10;
+        int messagesPerSecond = 10000;
+        String dialog_id = "test1";
+
+        String tmpDir = System.getProperty("java.io.tmpdir");
+        MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
+        MqttAsyncClient pub = new MqttAsyncClient("tcp://" + host +":1883", "PublisherClient"+dialog_id, dataStore);
+        MqttDefaultFilePersistence dataStoreSub = new MqttDefaultFilePersistence(tmpDir);
+        MqttAsyncClient sub = new MqttAsyncClient("tcp://" + host +":1883", "SubscriberClient"+dialog_id, dataStoreSub);
+
+        BenchmarkSubscriber suscriberBench = new BenchmarkSubscriber(sub, dialog_id);
+        suscriberBench.connect();
+
+        BenchmarkPublisher publisherBench = new BenchmarkPublisher(pub, numToSend, messagesPerSecond, dialog_id);
+        publisherBench.connect();
+        publisherBench.firePublishes();
+
+        suscriberBench.waitFinish();
+        publisherBench.waitFinish();
+    }
 }

netty_parser/src/test/resources/log4j.properties

@@ -3,12 +3,12 @@
 log4j.rootLogger=INFO, stdout
 
 log4j.logger.io.moquette=WARN
-log4j.logger.io.netty.util=WRN
+log4j.logger.io.netty.util=WARN
-log4j.logger.io.moquette.parser.netty.performance=WARN
+log4j.logger.io.moquette.parser.netty.performance=INFO
 
 # stdout appender is set to be a ConsoleAppender.
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
-log4j.appender.stdout.Threshold=WARN
+log4j.appender.stdout.Threshold=INFO
 
 
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
@@ -22,6 +22,3 @@ log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1}
 #log4j.appender.file.MaxBackupIndex=1
 #log4j.appender.file.layout=org.apache.log4j.PatternLayout
 #log4j.appender.file.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %L %x - %m%n
-
-
-

tools_scripts/benchmark/paho_benchmaker/BenchmarkPublisher.groovy

@@ -86,11 +86,12 @@ class BenchmarkPublisher {
         //initialize the timer
         PlatformTimer timer = PlatformTimer.detect()
         def pubCallback = new PublishCallback()
+        println ""
         (1..numToSend).each {
             long nanos = System.nanoTime()
             byte[] message = "Hello world!!-${nanos}".bytes
             this.client.publish("/topic" + dialog_id, message, qos, retain, null, pubCallback)
-            print ".\n"
+            print '.'
             timer.sleep(pauseMicroseconds)
         }
 

tools_scripts/benchmark/paho_benchmaker/BenchmarkSubscriber.groovy

@@ -13,6 +13,10 @@ import org.HdrHistogram.Histogram
 
 class BenchmarkSubscriber {
     class SubscriberCallback implements MqttCallback {
+
+//        private final CountDownLatch m_latch
+//        SubscriberCallback()
+
         public void messageArrived(String topic, MqttMessage message) throws Exception {
             //used to catch the first message published
             if (!alreadyStarted) {
@@ -24,14 +28,19 @@ class BenchmarkSubscriber {
             if (topic.toString().startsWith("/exit")) {
                 println "SUB: ok Exit!"
                 exit = true
-                connection.disconnect()
+                println "Before disconnect"
+                println "Before disconnect clienst $client"
+//                client.disconnect(1000) //wait max 1 sec
+                //client.disconnectForcibly()
+                //client.close()
+                println "After disconnect"
                 long stopTime = System.currentTimeMillis()
                 long spentTime = stopTime - startTime
                 println "SUB: " + topic.toString() + " received"
                 println "SUB: subscriber disconnected, received ${numReceived} messages in ${spentTime} ms"
                 double msgPerSec = (numReceived / spentTime) * 1000
                 println "SUB: Speed: $msgPerSec msg/sec"
-                println "SUB: Latency diagram"
+                println "SUB: Latency diagram [microsecs]"
                 histogram.outputPercentileDistribution(System.out, 1000.0); //nanos/1000
                 m_latch.countDown()
             } else {
@@ -87,7 +96,7 @@ class BenchmarkSubscriber {
             public void onSuccess(IMqttToken asyncActionToken) {
                 this.client.subscribe("/topic" + dialog_id, 1, null, new IMqttActionListener() {
                     public void onFailure(IMqttToken asyncActionToken2, Throwable exception) {
-                        this.connection.disconnect()
+                        this.client.disconnect()
                     }
 
                     public void onSuccess(IMqttToken asyncActionToken2) {
@@ -97,7 +106,7 @@ class BenchmarkSubscriber {
 
                 this.client.subscribe("/exit" + dialog_id, 1, null, new IMqttActionListener() {
                     public void onFailure(IMqttToken asyncActionToken2, Throwable exception) {
-                        this.connection.disconnect()
+                        this.client.disconnect()
                         m_latch.countDown()
                     }