diff --git a/tools_scripts/benchmark/paho_benchmaker/BenchmarkPublisher.groovy b/tools_scripts/benchmark/paho_benchmaker/BenchmarkPublisher.groovy new file mode 100644 index 00000000..01b1c77f --- /dev/null +++ b/tools_scripts/benchmark/paho_benchmaker/BenchmarkPublisher.groovy @@ -0,0 +1,109 @@ + +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 java.util.concurrent.CountDownLatch +import org.eclipse.jetty.toolchain.perf.PlatformTimer + +class BenchmarkPublisher { + static class PublishCallback implements IMqttActionListener { + public void onFailure(IMqttToken asyncActionToken, Throwable exception) { + println "PUB: Publish failed: " + exception + th.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 + println "PUB: published in ${spentTime} ms" + + this.connection.disconnect() + double msgPerSec = (numToSend / spentTime) * 1000 + println "PUB: speed $msgPerSec msg/sec" + this.stopLatch.countDown() + } + + public void onFailure(IMqttToken asyncActionToken, Throwable th) { + println "PUB: Publish failed: " + th + System.exit(2); + } + } + + private int qos = 0 + private boolean retain = false + private int numToSend + private int messagesPerSecond + private String dialog_id + private IMqttAsyncClient client + private CountDownLatch m_latch + + def connect() { + MqttConnectOptions connectOptions = new MqttConnectOptions() + connectOptions.cleanSession = true + this.client.connect(connectOptions, null, new IMqttActionListener() { + public void onFailure(IMqttToken asyncActionToken, Throwable exception) { + println "PUB: connect fail" + println "Excp: ${exception}" + System.exit(2) + } + + public void onSuccess(IMqttToken asyncActionToken) { + println "PUB: Successfully connected to server" + } + }) + + this.m_latch = new CountDownLatch(1) + } + + def firePublishes() { + long pauseMicroseconds = (1 / this.messagesPerSecond) * 1000 * 1000 + println "PUB: Pause over the each message sent ${pauseMicroseconds} microsecs" + + print "PUB: publishing.." + CountDownLatch m_latch = new CountDownLatch(1) + final long startTime = System.currentTimeMillis() + + //initialize the timer + PlatformTimer timer = PlatformTimer.detect() + def pubCallback = new PublishCallback() + (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" + timer.sleep(pauseMicroseconds) + } + + def exitCallback = new ExitTopicCallback(startTime, numToSend, this.client, m_latch) + long nanosExit = System.nanoTime() + byte[] exitMessage = "Hello world!!-${nanosExit}".bytes + this.client.publish("/exit" + dialog_id, exitMessage, qos, retain, null, exitCallback) + } + + + def waitFinish() { + m_latch.await() + } + +} \ No newline at end of file diff --git a/tools_scripts/benchmark/paho_benchmaker/BenchmarkSubscriber.groovy b/tools_scripts/benchmark/paho_benchmaker/BenchmarkSubscriber.groovy new file mode 100644 index 00000000..c1d145ac --- /dev/null +++ b/tools_scripts/benchmark/paho_benchmaker/BenchmarkSubscriber.groovy @@ -0,0 +1,129 @@ +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 java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import org.HdrHistogram.Histogram + +class BenchmarkSubscriber { + 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 String dialog_id + +// BenchmarkSubscriber(MQTT mqtt) { +// this.mqtt = mqtt +// } + def connect() { + this.connection = mqtt.callbackConnection() + connection.listener(new Listener() { + + 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 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 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 onFailure(Throwable value) { + connection.disconnect(null); // subscribe failed. + m_latch.countDown() + } + }); + } + }) + } + + + def waitFinish() { + this.m_latch.await() + } + +} \ No newline at end of file diff --git a/tools_scripts/benchmark/paho_benchmaker/benchmarker.groovy b/tools_scripts/benchmark/paho_benchmaker/benchmarker.groovy new file mode 100644 index 00000000..98905136 --- /dev/null +++ b/tools_scripts/benchmark/paho_benchmaker/benchmarker.groovy @@ -0,0 +1,55 @@ +//@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/') +@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.1.0') +@Grab(group='org.eclipse.jetty.toolchain', module='jetty-perf-helper', version='1.0.5') + + +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence +import org.eclipse.paho.client.mqttv3.MqttAsyncClient + +//import BenchmarkSubscriber +import BenchmarkPublisher + +if (args.size() < 3) { + println "Usage benchmarker [msg/sec] :[dialog_id]" + println "Ex localhost 10000 5000 test1" + println "should take 2 secs" + return +} + +println "*** Histogram measures are in microseconds ***" + +String host = args[0] +int numToSend = args[1] as int +int messagesPerSecond = args[2] as int +String dialog_id = args.size() > 3 ? args[3] : "" + +String tmpDir = System.getProperty("java.io.tmpdir") +MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir) +MqttAsyncClient pub = new MqttAsyncClient("tcp://${host}:1883", "PublisherClient${dialog_id}", dataStore) +//MqttConnectOptions pubConnOptions = new MqttConnectOptions() +//pubConnOptions.cleanSession = true +//pub.connect(pubConnOptions).waitForCompletion(1000) + +//MQTT pub = new MQTT() +//pub.setHost(host, 1883) +//pub.setCleanSession(true) +//pub.setClientId("PublisherClient${dialog_id}") +// +//MQTT sub = new MQTT() +////mqtt.setHost("test.mosquitto.org", 1883); +//sub.setHost(host, 1883) +//sub.setCleanSession(true) +//sub.setClientId("SubscriberClient${dialog_id}") + +//BenchmarkSubscriber suscriberBench = new BenchmarkSubscriber(mqtt: 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() +publisherBench.waitFinish() + +