From 68a4093e1afd61e10eef9dab540d0ed717c72e36 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 21 Dec 2016 23:17:22 +0100 Subject: [PATCH] Subscriber for benchmarking using Paho API --- .../paho_benchmaker/BenchmarkPublisher.groovy | 1 - .../BenchmarkSubscriber.groovy | 180 ++++++++---------- .../paho_benchmaker/benchmarker.groovy | 9 +- 3 files changed, 90 insertions(+), 100 deletions(-) diff --git a/tools_scripts/benchmark/paho_benchmaker/BenchmarkPublisher.groovy b/tools_scripts/benchmark/paho_benchmaker/BenchmarkPublisher.groovy index 01b1c77f..e4758de7 100644 --- a/tools_scripts/benchmark/paho_benchmaker/BenchmarkPublisher.groovy +++ b/tools_scripts/benchmark/paho_benchmaker/BenchmarkPublisher.groovy @@ -1,4 +1,3 @@ - import org.eclipse.paho.client.mqttv3.IMqttActionListener import org.eclipse.paho.client.mqttv3.IMqttToken import org.eclipse.paho.client.mqttv3.MqttConnectOptions diff --git a/tools_scripts/benchmark/paho_benchmaker/BenchmarkSubscriber.groovy b/tools_scripts/benchmark/paho_benchmaker/BenchmarkSubscriber.groovy index c1d145ac..aa8c4dd2 100644 --- a/tools_scripts/benchmark/paho_benchmaker/BenchmarkSubscriber.groovy +++ b/tools_scripts/benchmark/paho_benchmaker/BenchmarkSubscriber.groovy @@ -1,122 +1,110 @@ -import org.fusesource.mqtt.client.BlockingConnection -import org.fusesource.mqtt.client.CallbackConnection -import org.fusesource.mqtt.client.MQTT -import org.fusesource.mqtt.client.Listener -import org.fusesource.mqtt.client.Message -import org.fusesource.mqtt.client.QoS -import org.fusesource.mqtt.client.Topic -import org.fusesource.mqtt.client.Callback -import org.fusesource.hawtbuf.Buffer -import org.fusesource.hawtbuf.UTF8Buffer +import org.eclipse.paho.client.mqttv3.IMqttActionListener +import org.eclipse.paho.client.mqttv3.IMqttToken +import org.eclipse.paho.client.mqttv3.MqttConnectOptions +import org.eclipse.paho.client.mqttv3.IMqttAsyncClient +import org.eclipse.paho.client.mqttv3.MqttMessage +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken +import org.eclipse.paho.client.mqttv3.MqttCallback + + import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit import org.HdrHistogram.Histogram class BenchmarkSubscriber { + 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.toString().startsWith("/exit")) { + println "SUB: ok Exit!" + exit = true + connection.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" + histogram.outputPercentileDistribution(System.out, 1000.0); //nanos/1000 + m_latch.countDown() + } else { + String payload = new String(message.payload) + long sentTime = payload.split('-')[1] as long + long delay = System.nanoTime() - sentTime + histogram.recordValue(delay) + print '+' + numReceived++ +// if ((numReceived % 10000) == 0) { +// print '.' +// } + } + + } + + public void connectionLost(Throwable cause) { + println "Client ${clientCounter} lost connection, caused by ${cause}" + client.disconnect() + println "SUB: Something failed during message reception" + th.printStackTrace() + 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 CallbackConnection connection - private MQTT mqtt + private IMqttAsyncClient client private String dialog_id -// BenchmarkSubscriber(MQTT mqtt) { -// this.mqtt = mqtt -// } def connect() { - this.connection = mqtt.callbackConnection() - connection.listener(new Listener() { + MqttConnectOptions connectOptions = new MqttConnectOptions() + connectOptions.cleanSession = true + this.client.callback = new SubscriberCallback() + this.client.connect(connectOptions, null, new IMqttActionListener() { - public void onDisconnected() { - println "SUB: listener received disconnection" - } - public void onConnected() { - println "SUB: listener received a connected to ${mqtt.host}" -// startTime = System.currentTimeMillis() + public void onFailure(IMqttToken asyncActionToken, Throwable exception) { + println "SUB: connect fail" + println "Excp: ${exception}" + System.exit(2) } - public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) { - // You can now process a received message from a topic. - // Once process execute the ack runnable. - ack.run() - - //used to catch the first message published - if (!alreadyStarted) { - startTime = System.currentTimeMillis() - alreadyStarted = true - } - - //println "Received a publish on topic ${topic}" - if (topic.toString().startsWith("/exit")) { - println "SUB: ok Exit!" - exit = true - connection.disconnect(null) - 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" - histogram.outputPercentileDistribution(System.out, 1000.0); //nanos/1000 - m_latch.countDown() - } else { - String message = payload as String - long sentTime = message.split('-')[1] as long - long delay = System.nanoTime() - sentTime - histogram.recordValue(delay) - print '+' - numReceived++ -// if ((numReceived % 10000) == 0) { -// print '.' -// } - } - - } - public void onFailure(Throwable th) { - connection.disconnect(null) - println "SUB: Something failed during message reception" - th.printStackTrace() - m_latch.countDown() - } - }) - - connection.connect(new Callback() { - public void onFailure(Throwable value) { - System.exit(2); - } - - // Once we connect.. - public void onSuccess(Void v) { - // Subscribe to /topic - Topic[] topics = [new Topic("/topic" + dialog_id, QoS.AT_LEAST_ONCE)] - connection.subscribe(topics, new Callback() { - public void onSuccess(byte[] qoses) { - // The result of the subcribe request. - println "SUB: subscribed to /topic${dialog_id} ${qoses}" + 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() } - public void onFailure(Throwable value) { - connection.disconnect(null); // subscribe failed. - } - }); - - // Subscribe to /topic - Topic[] exitTopic = [new Topic("/exit" + dialog_id, QoS.AT_LEAST_ONCE)] - connection.subscribe(exitTopic, new Callback() { - public void onSuccess(byte[] qoses) { - // The result of the subcribe request. - println "SUB: subscribed to /exit${dialog_id} ${qoses}" + public void onSuccess(IMqttToken asyncActionToken2) { + println "SUB: subscribed to /topic${dialog_id} qos: 1" } - public void onFailure(Throwable value) { - connection.disconnect(null); // subscribe failed. + }) + + this.client.subscribe("/exit" + dialog_id, 1, null, new IMqttActionListener() { + public void onFailure(IMqttToken asyncActionToken2, Throwable exception) { + this.connection.disconnect() m_latch.countDown() } - }); + + public void onSuccess(IMqttToken asyncActionToken2) { + println "SUB: subscribed to /exit${dialog_id} qos: 1" + } + }) } }) } diff --git a/tools_scripts/benchmark/paho_benchmaker/benchmarker.groovy b/tools_scripts/benchmark/paho_benchmaker/benchmarker.groovy index 98905136..79f7ac13 100644 --- a/tools_scripts/benchmark/paho_benchmaker/benchmarker.groovy +++ b/tools_scripts/benchmark/paho_benchmaker/benchmarker.groovy @@ -41,15 +41,18 @@ MqttAsyncClient pub = new MqttAsyncClient("tcp://${host}:1883", "PublisherClient //sub.setCleanSession(true) //sub.setClientId("SubscriberClient${dialog_id}") -//BenchmarkSubscriber suscriberBench = new BenchmarkSubscriber(mqtt: sub, dialog_id: dialog_id) -//suscriberBench.connect() +MqttDefaultFilePersistence dataStoreSub = new MqttDefaultFilePersistence(tmpDir) +MqttAsyncClient sub = new MqttAsyncClient("tcp://${host}:1883", "SubscriberClient${dialog_id}", dataStoreSub) + +BenchmarkSubscriber suscriberBench = new BenchmarkSubscriber(client: sub, dialog_id: dialog_id) +suscriberBench.connect() BenchmarkPublisher publisherBench = new BenchmarkPublisher(client: pub, numToSend: numToSend, messagesPerSecond: messagesPerSecond, dialog_id: dialog_id) publisherBench.connect() publisherBench.firePublishes() -//suscriberBench.waitFinish() +suscriberBench.waitFinish() publisherBench.waitFinish()