diff --git a/tools_scripts/benchmark/BenchmarkPublisher.groovy b/tools_scripts/benchmark/BenchmarkPublisher.groovy new file mode 100644 index 00000000..2a41e57c --- /dev/null +++ b/tools_scripts/benchmark/BenchmarkPublisher.groovy @@ -0,0 +1,129 @@ +import java.net.URISyntaxException + +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.nio.ByteBuffer +import java.util.concurrent.CountDownLatch +import org.fusesource.hawtdispatch.Task +import org.eclipse.jetty.toolchain.perf.PlatformTimer + +class BenchmarkPublisher { + static class PublishCallback implements Callback { + public void onSuccess(Void value) { + //ok published + //increment counter + } + + public void onFailure(Throwable th) { + println "PUB: Publish failed: " + th + th.printStackTrace() + System.exit(2) + } + } + + static class ExitTopicCallback implements Callback { + + private final long startedTime + private final long numToSend + private final CallbackConnection connection + private final CountDownLatch stopLatch + + ExitTopicCallback(long startedTime, long numToSend, CallbackConnection connection, CountDownLatch stopLatch) { + this.startedTime = startedTime + this.numToSend = numToSend + this.connection = connection + this.stopLatch = stopLatch + } + + public void onSuccess(Void value2) { + long stopTime = System.currentTimeMillis() + long spentTime = stopTime - startedTime + println "PUB: published in ${spentTime} ms" + + this.connection.disconnect(null) + double msgPerSec = (numToSend / spentTime) * 1000 + println "PUB: speed $msgPerSec msg/sec" + this.stopLatch.countDown() + } + + public void onFailure(Throwable th) { + println "PUB: Publish failed: " + th + System.exit(2); + } + } + + private QoS qos = QoS.AT_MOST_ONCE + private boolean retain = false + private int numToSend + private int messagesPerSecond + private String dialog_id + private MQTT mqtt + private CallbackConnection connection + private CountDownLatch m_latch + + def connect() { + this.connection = mqtt.callbackConnection() + //connection.resume() + connection.connect(new Callback() { + public void onFailure(Throwable th) { + println "PUB: connect fail" + System.exit(2); + } + + // Once we connect.. + public void onSuccess(Void v) { + println "PUB: Succesfully 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" + //byte[] message = 'Hello world!!'.bytes + + 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 { + connection.getDispatchQueue().execute(new Task() { + public void run() { + long nanos = System.nanoTime() + byte[] message = "Hello world!!-${nanos}".bytes + connection.publish("/topic" + dialog_id, message, qos, retain, pubCallback) + } + }) + timer.sleep(pauseMicroseconds) + } + + def exitCallback = new ExitTopicCallback(startTime, numToSend, connection, m_latch) + connection.getDispatchQueue().execute(new Task() { + public void run() { + long nanos = System.nanoTime() + byte[] message = "Now exit!!-${nanos}".bytes + connection.publish("/exit" + dialog_id, message, qos, retain, exitCallback) + } + }) + } + + + def waitFinish() { + //wait finish... + m_latch.await() + } + +} \ No newline at end of file diff --git a/tools_scripts/benchmark/BenchmarkSubscriber.groovy b/tools_scripts/benchmark/BenchmarkSubscriber.groovy new file mode 100644 index 00000000..6185170d --- /dev/null +++ b/tools_scripts/benchmark/BenchmarkSubscriber.groovy @@ -0,0 +1,128 @@ +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); + m_latch.countDown() + } else { + String message = payload as String + long sentTime = message.split('-')[1] as long + long delay = System.nanoTime() - sentTime + histogram.recordValue(delay) + 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/fuse_asynch_benchmarker.groovy b/tools_scripts/benchmark/fuse_asynch_benchmarker.groovy new file mode 100644 index 00000000..03df858f --- /dev/null +++ b/tools_scripts/benchmark/fuse_asynch_benchmarker.groovy @@ -0,0 +1,59 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') +@Grab(group='org.eclipse.jetty.toolchain', module='jetty-perf-helper', version='1.0.5') + +import java.net.URISyntaxException + +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.nio.ByteBuffer +import java.util.concurrent.CountDownLatch +import org.fusesource.hawtdispatch.Task +import org.eclipse.jetty.toolchain.perf.PlatformTimer + +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 +} + +String host = args[0] +int numToSend = args[1] as int +int messagesPerSecond = args[2] as int +String dialog_id = args.size() > 3 ? args[3] : "" + +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.setHost(host, 1883) +sub.setClientId("SubscriberClient${dialog_id}") + +BenchmarkSubscriber suscriberBench = new BenchmarkSubscriber(mqtt: sub, dialog_id: dialog_id) +suscriberBench.connect() + +BenchmarkPublisher publisherBench = new BenchmarkPublisher(mqtt: pub, numToSend: numToSend, messagesPerSecond: messagesPerSecond, dialog_id: dialog_id) +publisherBench.connect() +publisherBench.firePublishes() + +suscriberBench.waitFinish() +publisherBench.waitFinish() + +