Implemented Publisher with Paho API
This commit is contained in:
andsel
@@ -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()
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<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" + dialog_id, QoS.AT_LEAST_ONCE)]
|
||||
connection.subscribe(topics, new Callback<byte[]>() {
|
||||
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<byte[]>() {
|
||||
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()
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 <host> <num messages to sent> <frequency>[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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user