anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added mqtt pub/sub synch(Paho lib) Future based and Callback/continuation based (Fuse lib)

Browse Source
This commit is contained in:
andrea
2015-02-05 20:57:56 +01:00
parent 9562f56a3a
commit d731fddaf1
6 changed files with 271 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools_scripts/fuse_asynch_publisher.groovy
+111
View File
@@ -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 <host> <num messages to sent>"
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<Void>() {
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<Void>() {
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<Void>() {
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<Void>() {
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()
tools_scripts/fuse_asynch_subscriber.groovy
+117
View File
@@ -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 <host>"
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<Void>() {
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<byte[]>() {
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<byte[]>() {
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()
tools_scripts/fuse_publisher.groovy
+10 -2
View File
@@ -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 <host> <num messages to sent>"
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"
tools_scripts/fuse_subscriber.groovy
+8 -1
View File
@@ -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 <host>"
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"
tools_scripts/publisher_client.groovy
+17 -3
View File
@@ -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 <host> <num messages to sent>"
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)
tools_scripts/subscriber_client.groovy
+8 -1
View File
@@ -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"