Parameteried the topics used for benchmark tests

tools_scripts/benchmark/fuse_asynch_publisher.groovy

@@ -20,8 +20,8 @@ import org.fusesource.hawtdispatch.Task
 import org.eclipse.jetty.toolchain.perf.PlatformTimer
 
 if (args.size() < 3) {
-    println "Usage publisher <host> <num messages to sent> <frequency>[msg/sec]"
-    println "Ex localhost 10000 5000"
+    println "Usage publisher <host> <num messages to sent> <frequency>[msg/sec] :[dialog_id]"
+    println "Ex localhost 10000 5000 test1"
     println "should take 2 secs"
     return
 }
@@ -29,13 +29,14 @@ if (args.size() < 3) {
 String host = args[0]
 int numToSend = args[1] as int
 int messagesPerSecond = args[2] as int
+String dialog_id = args.size() > 3 ? args[3] : ""
 long pauseMicroseconds = (1 / messagesPerSecond) * 1000 * 1000
 println "Pause over the each message sent ${pauseMicroseconds} microsecs"
 
 MQTT mqtt = new MQTT()
 mqtt.setHost(host, 1883)
 mqtt.setCleanSession(true)
-mqtt.setClientId("PublisherClient")
+mqtt.setClientId("PublisherClient${dialog_id}")
 
 final CallbackConnection connection = mqtt.callbackConnection()
 //connection.resume()
@@ -110,7 +111,7 @@ def pubCallback = new PublishCallback()
         public void run() {
             long nanos = System.nanoTime()
             byte[] message = "Hello world!!-${nanos}".bytes
-            connection.publish("/topic", message, qos, retain, pubCallback)
+            connection.publish("/topic" + dialog_id, message, qos, retain, pubCallback)
         }
     })
     timer.sleep(pauseMicroseconds)
@@ -122,7 +123,7 @@ connection.getDispatchQueue().execute(new Task() {
     public void run() {
         long nanos = System.nanoTime()
         byte[] message = "Now exit!!-${nanos}".bytes
-        connection.publish("/exit", message, qos, retain, exitCallback)
+        connection.publish("/exit" + dialog_id, message, qos, retain, exitCallback)
     }
 })
 

tools_scripts/benchmark/fuse_asynch_subscriber.groovy

@@ -15,18 +15,20 @@ import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import org.HdrHistogram.Histogram
 
-if (args.size() != 1) {
-    println "Usage subscriber <host>"
+if (args.size() < 1) {
+    println "Usage subscriber <host> :[dialog_id]"
     return
 }
 
 String host = args[0]
+String dialog_id = args.size() > 1 ? args[1] : ""
+
 MQTT mqtt = new MQTT()
 //mqtt.setHost("test.mosquitto.org", 1883);
 mqtt.setHost(host, 1883)
 mqtt.setCleanSession(true)
 mqtt.setHost(host, 1883)
-mqtt.setClientId("SubscriberClient")
+mqtt.setClientId("SubscriberClient${dialog_id}")
 
 int numReceived = 0
 long startTime = System.currentTimeMillis()
@@ -58,13 +60,13 @@ connection.listener(new Listener() {
          }
 
         //println "Received a publish on topic ${topic}"
-        if (topic.toString() == "/exit") {
+        if (topic.toString().startsWith("/exit")) {
             println "Ok Exit!"
             exit = true
             connection.disconnect(null)
             long stopTime = System.currentTimeMillis()
             long spentTime = stopTime - startTime
-            println "/exit received"
+            println topic.toString() + " received"
             println "subscriber disconnected, received ${numReceived} messages in ${spentTime} ms"
             double msgPerSec = (numReceived / spentTime) * 1000
             println "Speed: $msgPerSec msg/sec"
@@ -99,11 +101,11 @@ connection.connect(new Callback<Void>() {
     // Once we connect..
     public void onSuccess(Void v) {
         // Subscribe to /topic
-        Topic[] topics = [new Topic("/topic", QoS.AT_LEAST_ONCE)]
+        Topic[] topics = [new Topic("/topic" + dialog_id, QoS.AT_LEAST_ONCE)]
         connection.subscribe(topics, new Callback<byte[]>() {
             public void onSuccess(byte[] qoses) {
                 // The result of the subcribe request.
-                println "subscribed to /topic ${qoses}"
+                println "subscribed to /topic${dialog_id} ${qoses}"
             }
             public void onFailure(Throwable value) {
                 connection.disconnect(null); // subscribe failed.
@@ -112,11 +114,11 @@ connection.connect(new Callback<Void>() {
 
 
         // Subscribe to /topic
-        Topic[] exitTopic = [new Topic("/exit", QoS.AT_LEAST_ONCE)]
+        Topic[] exitTopic = [new Topic("/exit" + dialog_id, QoS.AT_LEAST_ONCE)]
         connection.subscribe(exitTopic, new Callback<byte[]>() {
             public void onSuccess(byte[] qoses) {
                 // The result of the subcribe request.
-                println "subscribed to /exit ${qoses}"
+                println "subscribed to /exit${dialog_id} ${qoses}"
             }
             public void onFailure(Throwable value) {
                 connection.disconnect(null); // subscribe failed.