diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties
index 7913dd43..1de48f19 100644
--- a/broker/src/test/resources/log4j.properties
+++ b/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.additivity.messageLogger=false
+log4j.category.messageLogger=WARN, messagelog
+#log4j.additivity.messageLogger=false
diff --git a/netty_parser/pom.xml b/netty_parser/pom.xml
index fe2e7036..250a9268 100644
--- a/netty_parser/pom.xml
+++ b/netty_parser/pom.xml
@@ -64,6 +64,33 @@
test
+
+ org.hdrhistogram
+ HdrHistogram
+ 2.1.2
+ test
+
+
+
+ org.eclipse.paho
+ org.eclipse.paho.client.mqttv3
+ 1.1.0
+ test
+
+
+
+ org.eclipse.jetty.toolchain
+ jetty-perf-helper
+ 1.0.5
+ test
+
+
+
+
+
+
+
+
diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java
new file mode 100644
index 00000000..8f8f4bb6
--- /dev/null
+++ b/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();
+ }
+
+}
diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java
new file mode 100644
index 00000000..f1d77050
--- /dev/null
+++ b/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();
+ }
+
+}
\ No newline at end of file
diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopHandler.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopHandler.java
index afe3c0d4..b80bcbc0 100644
--- a/netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopHandler.java
+++ b/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