diff --git a/tools_scripts/fuse_asynch_publisher.groovy b/tools_scripts/fuse_asynch_publisher.groovy new file mode 100644 index 00000000..89ac19aa --- /dev/null +++ b/tools_scripts/fuse_asynch_publisher.groovy @@ -0,0 +1,111 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') + +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.util.concurrent.CountDownLatch +import org.fusesource.hawtdispatch.Task + +if (args.size() < 2) { + println "Usage publisher " + return +} + +String host = args[0] +int numToSend = args[1] as int +MQTT mqtt = new MQTT() +//mqtt.setHost("test.mosquitto.org", 1883); +mqtt.setHost(host, 1883) +mqtt.setCleanSession(true) +mqtt.setHost(host, 1883) +mqtt.setClientId("PublisherClient") + +long startTime = System.currentTimeMillis() +CountDownLatch m_latch = new CountDownLatch(1) + +final CallbackConnection connection = mqtt.callbackConnection() +//connection.resume() +connection.connect(new Callback() { + public void onFailure(Throwable th) { + println "connect fail" + System.exit(2); + } + + // Once we connect.. + public void onSuccess(Void v) { + + } +}) + +String topic = "/topic" +byte[] message = 'Hello world!!'.bytes +QoS qos = QoS.AT_MOST_ONCE +boolean retain = false + +print "publishing.." +startTime = System.currentTimeMillis() + +new Task() { + long sent = 0; + public void run() { + final Task publish = this + + connection.publish(topic, message, qos, retain, new Callback() { + public void onSuccess(Void value) { + sent ++; + + if( sent < numToSend ) { + connection.getDispatchQueue().execute(publish); + } else { + connection.getDispatchQueue().execute(new Runnable() { + public void run() { + connection.publish("/exit", message, qos, retain, new Callback() { + public void onSuccess(Void value2) { + long stopTime = System.currentTimeMillis() + long spentTime = stopTime -startTime + println "published in ${spentTime} ms" + + connection.disconnect(null) + double msgPerSec = (numToSend / spentTime) * 1000 + println "speed $msgPerSec msg/sec" + + connection.disconnect(new Callback() { + public void onSuccess(Void valueSuccess2) { + m_latch.countDown(); + } + public void onFailure(Throwable th2) { + m_latch.countDown(); + } + }); + } + + public void onFailure(Throwable th) { + println "Publish failed: " + th + + System.exit(2); + } + }); + } + }); + + } + } + public void onFailure(Throwable th) { + println "Publish failed: " + th + + System.exit(2); + } + }); + } +}.run(); + +m_latch.await() \ No newline at end of file diff --git a/tools_scripts/fuse_asynch_subscriber.groovy b/tools_scripts/fuse_asynch_subscriber.groovy new file mode 100644 index 00000000..ec1cf442 --- /dev/null +++ b/tools_scripts/fuse_asynch_subscriber.groovy @@ -0,0 +1,117 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') + +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 + +if (args.size() != 1) { + println "Usage subscriber " + return +} + +String host = args[0] +MQTT mqtt = new MQTT() +//mqtt.setHost("test.mosquitto.org", 1883); +mqtt.setHost(host, 1883) +mqtt.setCleanSession(true) +mqtt.setHost(host, 1883) +mqtt.setClientId("SubscriberClient") + +int numReceived = 0 +long startTime = System.currentTimeMillis() +boolean exit = false +CountDownLatch m_latch = new CountDownLatch(1) + +final CallbackConnection connection = mqtt.callbackConnection() +connection.listener(new Listener() { + + public void onDisconnected() { + println "Listener received disconnection" + } + public void onConnected() { + println "Listener received a connected to ${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() + //println "Received a publish on topic ${topic}" + if (topic.toString() == "/exit") { + println "Ok Exit!" + exit = true + connection.disconnect(null) + long stopTime = System.currentTimeMillis() + long spentTime = stopTime - startTime + println "/exit received" + println "subscriber disconnected, received ${numReceived} messages in ${spentTime} ms" + double msgPerSec = (numReceived / spentTime) * 1000 + println "speed $msgPerSec msg/sec" + m_latch.countDown() + } else { + numReceived++ +// if ((numReceived % 10000) == 0) { +// print '.' +// } + } + + } + public void onFailure(Throwable value) { + connection.disconnect(null) + println "Some thing failed $value" + m_latch.countDown() + } +}) + +//Topic[] topics = [new Topic("/topic", QoS.AT_MOST_ONCE), new Topic("/exit", QoS.AT_MOST_ONCE)] +//byte[] qoses = connection.subscribe(topics) +//println "subscribed to /topic qos: ${qoses[0]}" +//println "subscribed to /exit qos: ${qoses[1]}" + +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", QoS.AT_LEAST_ONCE)] + connection.subscribe(topics, new Callback() { + public void onSuccess(byte[] qoses) { + // The result of the subcribe request. + println "subscribed to /topic ${qoses}" + } + public void onFailure(Throwable value) { + connection.disconnect(null); // subscribe failed. + } + }); + + + // Subscribe to /topic + Topic[] exitTopic = [new Topic("/exit", QoS.AT_LEAST_ONCE)] + connection.subscribe(exitTopic, new Callback() { + public void onSuccess(byte[] qoses) { + // The result of the subcribe request. + println "subscribed to /exit ${qoses}" + } + public void onFailure(Throwable value) { + connection.disconnect(null); // subscribe failed. + m_latch.countDown() + } + }); + } +}) + + +m_latch.await() \ No newline at end of file diff --git a/tools_scripts/fuse_publisher.groovy b/tools_scripts/fuse_publisher.groovy index 7d53ee29..81cd94b9 100644 --- a/tools_scripts/fuse_publisher.groovy +++ b/tools_scripts/fuse_publisher.groovy @@ -1,4 +1,4 @@ -@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') import java.net.URISyntaxException @@ -6,7 +6,13 @@ import org.fusesource.mqtt.client.BlockingConnection import org.fusesource.mqtt.client.MQTT import org.fusesource.mqtt.client.QoS +if (args.size() < 2) { + println "Usage publisher " + return +} + String host = args[0] +int numToSend = args[1] as int MQTT mqtt = new MQTT() //mqtt.setHost("test.mosquitto.org", 1883); @@ -16,7 +22,7 @@ connection.connect() print "publishing.." long startTime = System.currentTimeMillis() -(1..100000).each { +(1..numToSend).each { connection.publish("/topic", 'Hello world!!'.bytes, QoS.AT_MOST_ONCE, false) } connection.publish('/exit', 'Exit'.bytes, QoS.AT_MOST_ONCE, false) @@ -24,4 +30,6 @@ long stopTime = System.currentTimeMillis() long spentTime = stopTime -startTime println "published in ${spentTime} ms" connection.disconnect() +double msgPerSec = (numToSend / spentTime) * 1000 +println "speed $msgPerSec msg/sec" println "disconnected" \ No newline at end of file diff --git a/tools_scripts/fuse_subscriber.groovy b/tools_scripts/fuse_subscriber.groovy index 2fbc055d..a1e85e6c 100644 --- a/tools_scripts/fuse_subscriber.groovy +++ b/tools_scripts/fuse_subscriber.groovy @@ -1,4 +1,4 @@ -@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') import java.net.URISyntaxException @@ -10,6 +10,11 @@ import org.fusesource.mqtt.client.Topic import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit +if (args.size() != 1) { + println "Usage subscriber " + return +} + String host = args[0] MQTT mqtt = new MQTT() //mqtt.setHost("test.mosquitto.org", 1883); @@ -48,4 +53,6 @@ long stopTime = System.currentTimeMillis() long spentTime = stopTime - startTime println "/exit received" println "subscriber disconnected, received ${numReceived} messages in ${spentTime} ms" +double msgPerSec = (numReceived / spentTime) * 1000 +println "speed $msgPerSec msg/sec" diff --git a/tools_scripts/publisher_client.groovy b/tools_scripts/publisher_client.groovy index 8f4797ca..b9c5a1ed 100644 --- a/tools_scripts/publisher_client.groovy +++ b/tools_scripts/publisher_client.groovy @@ -5,19 +5,33 @@ import org.eclipse.paho.client.mqttv3.MqttClient import org.eclipse.paho.client.mqttv3.MqttMessage import org.eclipse.paho.client.mqttv3.MqttException import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence +import org.eclipse.paho.client.mqttv3.MqttConnectOptions + +if (args.size() < 2) { + println "Usage publisher " + return +} String host = args[0] +int numToSend = args[1] as int String tmpDir = System.getProperty("java.io.tmpdir") MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir) MqttClient client = new MqttClient("tcp://${host}:1883", "PublisherClient", dataStore) -client.connect() +//Only for Apollo --- +//MqttConnectOptions connOptions = new MqttConnectOptions() +//connOptions.userName ='admin' +//connOptions.password = 'password' as char[] +//Only for Apollo --- +//client.connect(connOptions) +client.connect(connOptions) MqttMessage message = new MqttMessage('Hello world!!'.bytes) message.setQos(0) -print "publishing.." +print "publishing..${numToSend} messages, " long startTime = System.currentTimeMillis() -(1..1000000).each { +//(1..1000000).each { +(1..numToSend).each { client.publish('topic', 'Hello world!!'.bytes, 0, false) } client.publish('/exit', 'Exit'.bytes, 0, false) diff --git a/tools_scripts/subscriber_client.groovy b/tools_scripts/subscriber_client.groovy index aaecc5f4..6e507058 100644 --- a/tools_scripts/subscriber_client.groovy +++ b/tools_scripts/subscriber_client.groovy @@ -7,6 +7,7 @@ import org.eclipse.paho.client.mqttv3.MqttException import org.eclipse.paho.client.mqttv3.MqttCallback import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence +import org.eclipse.paho.client.mqttv3.MqttConnectOptions import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit @@ -41,7 +42,7 @@ class SubscriberCallback implements MqttCallback { client.disconnect() m_latch.countDown() } else { - println "-received ${message} on ${topic} with QoS ${message.qos}" + //println "-received ${message} on ${topic} with QoS ${message.qos}" m_numReceived++ } } @@ -65,6 +66,12 @@ int rnd = (Math.random() * 100) as int MqttClient client = new MqttClient("tcp://${host}:1883", "SubscriberClient${rnd}", dataStore) def callback = new SubscriberCallback() client.callback = callback +//Only for Apollo --- +//MqttConnectOptions connOptions = new MqttConnectOptions() +//connOptions.userName ='admin' +//connOptions.password = 'password' as char[] +//client.connect(connOptions) +//Only for Apollo --- client.connect() client.subscribe("topic", 0) println "subscribed to topic"