From fffa9afbfab6893e14227125b987944b469cdda3 Mon Sep 17 00:00:00 2001 From: andrea Date: Thu, 5 Feb 2015 21:08:18 +0100 Subject: [PATCH] Divided integration and benchmark scripts --- .../benchmarkReportGenerator.groovy | 0 .../{ => benchmark}/benchmarker.groovy | 0 .../fuse_asynch_publisher.groovy | 0 .../fuse_asynch_subscriber.groovy | 0 .../{ => benchmark}/fuse_publisher.groovy | 0 .../{ => benchmark}/fuse_subscriber.groovy | 0 .../{ => benchmark}/publisher_client.groovy | 0 .../{ => benchmark}/subscriber_client.groovy | 0 .../integration/brokerEmbeddedTest.groovy | 16 +++ .../integration/bunch_publish_issue12.groovy | 24 ++++ .../integration/hawtdb_multindex.groovy | 81 ++++++++++++ tools_scripts/integration/hawtdb_play.groovy | 65 ++++++++++ .../integration/hawtdb_raw_access.groovy | 45 +++++++ tools_scripts/integration/index.html | 104 +++++++++++++++ .../integration/issue15_receiver.groovy | 48 +++++++ .../integration/issue15_sender.groovy | 27 ++++ .../issue4_publish2Subscribers.groovy | 42 ++++++ .../issue_12_multiple_republish_qos2.groovy | 89 +++++++++++++ .../{ => integration}/list-broker-topics.rb | 0 .../{ => integration}/paho_test_issue8.groovy | 0 .../integration/twoClientsWithSameID.groovy | 56 ++++++++ .../integration/webSocketClient.groovy | 121 ++++++++++++++++++ .../integration/web_socket_test.html | 101 +++++++++++++++ 23 files changed, 819 insertions(+) rename tools_scripts/{ => benchmark}/benchmarkReportGenerator.groovy (100%) rename tools_scripts/{ => benchmark}/benchmarker.groovy (100%) rename tools_scripts/{ => benchmark}/fuse_asynch_publisher.groovy (100%) rename tools_scripts/{ => benchmark}/fuse_asynch_subscriber.groovy (100%) rename tools_scripts/{ => benchmark}/fuse_publisher.groovy (100%) rename tools_scripts/{ => benchmark}/fuse_subscriber.groovy (100%) rename tools_scripts/{ => benchmark}/publisher_client.groovy (100%) rename tools_scripts/{ => benchmark}/subscriber_client.groovy (100%) create mode 100644 tools_scripts/integration/brokerEmbeddedTest.groovy create mode 100644 tools_scripts/integration/bunch_publish_issue12.groovy create mode 100644 tools_scripts/integration/hawtdb_multindex.groovy create mode 100644 tools_scripts/integration/hawtdb_play.groovy create mode 100644 tools_scripts/integration/hawtdb_raw_access.groovy create mode 100644 tools_scripts/integration/index.html create mode 100644 tools_scripts/integration/issue15_receiver.groovy create mode 100644 tools_scripts/integration/issue15_sender.groovy create mode 100644 tools_scripts/integration/issue4_publish2Subscribers.groovy create mode 100644 tools_scripts/integration/issue_12_multiple_republish_qos2.groovy rename tools_scripts/{ => integration}/list-broker-topics.rb (100%) rename tools_scripts/{ => integration}/paho_test_issue8.groovy (100%) create mode 100644 tools_scripts/integration/twoClientsWithSameID.groovy create mode 100644 tools_scripts/integration/webSocketClient.groovy create mode 100644 tools_scripts/integration/web_socket_test.html diff --git a/tools_scripts/benchmarkReportGenerator.groovy b/tools_scripts/benchmark/benchmarkReportGenerator.groovy similarity index 100% rename from tools_scripts/benchmarkReportGenerator.groovy rename to tools_scripts/benchmark/benchmarkReportGenerator.groovy diff --git a/tools_scripts/benchmarker.groovy b/tools_scripts/benchmark/benchmarker.groovy similarity index 100% rename from tools_scripts/benchmarker.groovy rename to tools_scripts/benchmark/benchmarker.groovy diff --git a/tools_scripts/fuse_asynch_publisher.groovy b/tools_scripts/benchmark/fuse_asynch_publisher.groovy similarity index 100% rename from tools_scripts/fuse_asynch_publisher.groovy rename to tools_scripts/benchmark/fuse_asynch_publisher.groovy diff --git a/tools_scripts/fuse_asynch_subscriber.groovy b/tools_scripts/benchmark/fuse_asynch_subscriber.groovy similarity index 100% rename from tools_scripts/fuse_asynch_subscriber.groovy rename to tools_scripts/benchmark/fuse_asynch_subscriber.groovy diff --git a/tools_scripts/fuse_publisher.groovy b/tools_scripts/benchmark/fuse_publisher.groovy similarity index 100% rename from tools_scripts/fuse_publisher.groovy rename to tools_scripts/benchmark/fuse_publisher.groovy diff --git a/tools_scripts/fuse_subscriber.groovy b/tools_scripts/benchmark/fuse_subscriber.groovy similarity index 100% rename from tools_scripts/fuse_subscriber.groovy rename to tools_scripts/benchmark/fuse_subscriber.groovy diff --git a/tools_scripts/publisher_client.groovy b/tools_scripts/benchmark/publisher_client.groovy similarity index 100% rename from tools_scripts/publisher_client.groovy rename to tools_scripts/benchmark/publisher_client.groovy diff --git a/tools_scripts/subscriber_client.groovy b/tools_scripts/benchmark/subscriber_client.groovy similarity index 100% rename from tools_scripts/subscriber_client.groovy rename to tools_scripts/benchmark/subscriber_client.groovy diff --git a/tools_scripts/integration/brokerEmbeddedTest.groovy b/tools_scripts/integration/brokerEmbeddedTest.groovy new file mode 100644 index 00000000..da08f1a9 --- /dev/null +++ b/tools_scripts/integration/brokerEmbeddedTest.groovy @@ -0,0 +1,16 @@ +@GrabResolver(name='moquette-bintray', root='http://dl.bintray.com/andsel/maven/') +@Grab(group='org.dna.mqtt', module='moquette-broker', version='0.6') + +import org.eclipse.moquette.server.Server + +println "Starting broker in embedded mode" +Server server = new Server() +//Properties props = [port: 1883, host: "0.0.0.0", 'password_file': '../broker/config/password_file.conf'] as Properties +Properties props = new Properties() +props.setProperty('port', '1883') +props.setProperty('host', "0.0.0.0") +props.setProperty('password_file', '../broker/config/password_file.conf') +println "starting proprs $props" +server.startServer(props) +println "Stopping broker.." +server.stopServer() \ No newline at end of file diff --git a/tools_scripts/integration/bunch_publish_issue12.groovy b/tools_scripts/integration/bunch_publish_issue12.groovy new file mode 100644 index 00000000..4770431b --- /dev/null +++ b/tools_scripts/integration/bunch_publish_issue12.groovy @@ -0,0 +1,24 @@ +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.MqttDefaultFilePersistence + +String tmpDir = System.getProperty("java.io.tmpdir") +MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir) + +MqttClient client = new MqttClient("tcp://localhost:1883", "SampleClient", dataStore) +client.connect() +println "connected" +(1..10).each { + //MqttMessage message = new MqttMessage("Hello world!! $it".toString().bytes) + String s = "Hello World!! $it" + MqttMessage message = new MqttMessage(s.bytes) + message.qos = 2 + message.retained = true + print "publishing.." + client.getTopic("log").publish(message) + println "published" +} + +client.disconnect() +println "disconnected" \ No newline at end of file diff --git a/tools_scripts/integration/hawtdb_multindex.groovy b/tools_scripts/integration/hawtdb_multindex.groovy new file mode 100644 index 00000000..44264415 --- /dev/null +++ b/tools_scripts/integration/hawtdb_multindex.groovy @@ -0,0 +1,81 @@ +@Grab(group='org.fusesource.hawtdb', module='hawtdb', version='1.6') + +import org.fusesource.hawtbuf.codec.StringCodec +import org.fusesource.hawtdb.api.IndexFactory +import org.fusesource.hawtdb.api.HashIndexFactory +import org.fusesource.hawtdb.api.BTreeIndexFactory +import org.fusesource.hawtdb.api.PageFile +import org.fusesource.hawtdb.api.PageFileFactory +import org.fusesource.hawtdb.api.MultiIndexFactory +import org.fusesource.hawtdb.api.SortedIndex + + + +File tmpFile = new File(System.getProperty("user.home") + File.separator + "test_hawtdb.dat") +//tmpFile.createNewFile() + +def writeSomeData(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile); + + BTreeIndexFactory indexFactory = new BTreeIndexFactory() + indexFactory.setKeyCodec(StringCodec.INSTANCE) + + BTreeIndexFactory indexFactory2 = new BTreeIndexFactory() + + + SortedIndex writeIndex = multiIndexFactory.openOrCreate("subscriptions", indexFactory) + SortedIndex retainedIndex = multiIndexFactory.openOrCreate("retained", indexFactory2) + println "Hawt opened, try to read" + println "Size: ${writeIndex.size()}" + + println "write something" + writeIndex.put("titolo", "capucetto rosso") + retainedIndex.put("autore", "leggenda popolare") + pageFile.flush() + + println "check after write content: ${writeIndex.get("titolo")}, size ${writeIndex.size()}" + + //pageFile.close() + factory.close() +} + + +def readAllFileContents(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + + MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile); + + IndexFactory indexFactory = new BTreeIndexFactory() + BTreeIndexFactory indexFactory2 = new BTreeIndexFactory() + indexFactory.setKeyCodec(StringCodec.INSTANCE) + + SortedIndex readingIndex = multiIndexFactory.openOrCreate("subscriptions", indexFactory) + SortedIndex retainedIndex = multiIndexFactory.openOrCreate("retained", indexFactory2) + + println "read something, size ${readingIndex.size()}" + for (def entry : readingIndex) { + println "read k, v -> ${entry.key}, ${entry.value} " + } + + for (def entry : retainedIndex) { + println "read k, v -> ${entry.key}, ${entry.value} " + } + + factory.close() +} + + +writeSomeData(tmpFile) +readAllFileContents(tmpFile) + + + + + diff --git a/tools_scripts/integration/hawtdb_play.groovy b/tools_scripts/integration/hawtdb_play.groovy new file mode 100644 index 00000000..17efe18e --- /dev/null +++ b/tools_scripts/integration/hawtdb_play.groovy @@ -0,0 +1,65 @@ +@Grab(group='org.fusesource.hawtdb', module='hawtdb', version='1.6') + +import org.fusesource.hawtbuf.codec.StringCodec +import org.fusesource.hawtdb.api.BTreeIndexFactory +import org.fusesource.hawtdb.api.PageFile +import org.fusesource.hawtdb.api.PageFileFactory +import org.fusesource.hawtdb.api.SortedIndex + + + +File tmpFile = new File(System.getProperty("user.home") + File.separator + "test_hawtdb.dat") +//tmpFile.createNewFile() + +def writeSomeData(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + + BTreeIndexFactory indexFactory = new BTreeIndexFactory() + indexFactory.setKeyCodec(StringCodec.INSTANCE) + + SortedIndex writeIndex = indexFactory.openOrCreate(pageFile) + println "Hawt opened, try to read" + println "Size: ${writeIndex.size()}" + + println "write something" + writeIndex.put("titolo", "capucetto rosso") + pageFile.flush() + + println "check after write content: ${writeIndex.get("titolo")}, size ${writeIndex.size()}" + + //pageFile.close() + factory.close() +} + + +def readAllFileContents(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + + BTreeIndexFactory indexFactory = new BTreeIndexFactory() + indexFactory.setKeyCodec(StringCodec.INSTANCE) + + SortedIndex readingIndex = indexFactory.openOrCreate(pageFile) + + println "read something, size ${readingIndex.size()}" + for (def entry : readingIndex) { + println "read k, v -> ${entry.key}, ${entry.value} " + } + + factory.close() +} + + +writeSomeData(tmpFile) +readAllFileContents(tmpFile) + + + + + + diff --git a/tools_scripts/integration/hawtdb_raw_access.groovy b/tools_scripts/integration/hawtdb_raw_access.groovy new file mode 100644 index 00000000..75579f46 --- /dev/null +++ b/tools_scripts/integration/hawtdb_raw_access.groovy @@ -0,0 +1,45 @@ +@Grab(group='org.fusesource.hawtdb', module='hawtdb', version='1.6') + +import org.fusesource.hawtbuf.codec.StringCodec +import org.fusesource.hawtdb.api.BTreeIndexFactory +import org.fusesource.hawtdb.api.PageFile +import org.fusesource.hawtdb.api.PageFileFactory +import org.fusesource.hawtdb.api.SortedIndex +import org.fusesource.hawtbuf.Buffer + + + +File tmpFile = new File(System.getProperty("user.home") + File.separator + "test_hawtdb.dat") +//tmpFile.createNewFile() + +def writeSomeData(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + int pageIdx = pageFile.alloc() + pageFile.write(pageIdx, new Buffer("Test".getBytes())) + println "writed at page index ${pageIdx}" + pageFile.flush() + pageFile.close() +} + + +def readTest(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + println "page size is: " + pageFile.pageSize + Buffer b = new Buffer(512) + pageFile.read(0, b) + println "content try read " + b + + pageFile.close() +} + + +writeSomeData(tmpFile) +readTest(tmpFile) + + diff --git a/tools_scripts/integration/index.html b/tools_scripts/integration/index.html new file mode 100644 index 00000000..ca365799 --- /dev/null +++ b/tools_scripts/integration/index.html @@ -0,0 +1,104 @@ + + + + + Moquette performances + + + + + + + + + + + + + + +
+ + + + diff --git a/tools_scripts/integration/issue15_receiver.groovy b/tools_scripts/integration/issue15_receiver.groovy new file mode 100644 index 00000000..a2900e2b --- /dev/null +++ b/tools_scripts/integration/issue15_receiver.groovy @@ -0,0 +1,48 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import java.net.URISyntaxException + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +class Reciever { + MQTT mqtt = new MQTT() + BlockingConnection connection + + public void connectAndSubscribe(boolean cleanSession) { + mqtt.setCleanSession(cleanSession) + mqtt.setHost("localhost", 1883) + + mqtt.setClientId("reciever") + connection = mqtt.blockingConnection() + connection.connect() + Topic[] topics = [new Topic("/topic", QoS.AT_LEAST_ONCE)] + byte[] qoses = connection.subscribe(topics) + } +} + + +Reciever r = new Reciever() +r.connectAndSubscribe(false) +Message message +3.times { i -> + message = r.connection.receive() + println "Topic: ${message.topic}, payload: ${new String(message.payload)}" + message.ack() +} +r.connection.kill() +//r.connection.disconnect() + +Thread.sleep(3000) + +r.connectAndSubscribe(false) +3.times { i -> + message = r.connection.receive() + println "Topic: ${message.topic}, payload: ${new String(message.payload)}" + message.ack() +} +r.connection.disconnect() + diff --git a/tools_scripts/integration/issue15_sender.groovy b/tools_scripts/integration/issue15_sender.groovy new file mode 100644 index 00000000..f79d4a36 --- /dev/null +++ b/tools_scripts/integration/issue15_sender.groovy @@ -0,0 +1,27 @@ +//@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/') +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import java.net.URISyntaxException + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.QoS + + +MQTT mqtt = new MQTT() +//mqtt.setHost("test.mosquitto.org", 1883); +mqtt.setHost("localhost", 1883) + +BlockingConnection connection = mqtt.blockingConnection() +connection.connect() + +//for (int i = 0; i < 1000; i++) { +(1..1000).each{ i -> +// LOG.info("Publishing"); + String payload = "Hello world MQTT!!${i}" + connection.publish("/topic", payload.getBytes(), QoS.AT_LEAST_ONCE, false) + println payload + Thread.sleep 1000 +} + +connection.disconnect() diff --git a/tools_scripts/integration/issue4_publish2Subscribers.groovy b/tools_scripts/integration/issue4_publish2Subscribers.groovy new file mode 100644 index 00000000..cd3b8fdb --- /dev/null +++ b/tools_scripts/integration/issue4_publish2Subscribers.groovy @@ -0,0 +1,42 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +String host = args[0] +MQTT mqtt = new MQTT() +mqtt.setHost(host, 1883) +mqtt.setCleanSession(true) +mqtt.setClientId("SubscriberClient") + +BlockingConnection subscriber = mqtt.blockingConnection() +subscriber.connect() +println "Connected to ${host}" + +Topic[] topics = [new Topic("/topic", QoS.AT_MOST_ONCE)] +byte[] qoses = subscriber.subscribe(topics) +println "Subscribed to ${topics}" + +//start a publisher +MQTT mqtt2 = new MQTT() +mqtt2.setHost(host, 1883) +BlockingConnection publisher = mqtt2.blockingConnection() +publisher.connect() +publisher.publish("/topic", 'Hello world!!'.bytes, QoS.AT_MOST_ONCE, false) + +//check subscriber received +Message message = subscriber.receive() +assert "/topic" == message.topic +assert "Hello world!!" == new String(message.payload) +message.ack() + +println "shutdown subscriber" +subscriber.disconnect() + +println "shutdown publisher" +publisher.disconnect() + +println "Kill your broker and start it again, then trivialPublish.groovy" \ No newline at end of file diff --git a/tools_scripts/integration/issue_12_multiple_republish_qos2.groovy b/tools_scripts/integration/issue_12_multiple_republish_qos2.groovy new file mode 100644 index 00000000..99700437 --- /dev/null +++ b/tools_scripts/integration/issue_12_multiple_republish_qos2.groovy @@ -0,0 +1,89 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +//test case moquette 0.7 update for 2015-1-26 +//client eclipse.paht.javaclient version :1.0.1 +String host = args.size() > 0 ? args[0] : 'localhost' + +MQTT mqttPub = new MQTT() +mqttPub.setHost(host, 1883) +mqttPub.setCleanSession(false) +mqttPub.setClientId("PublisherClient") + +MQTT mqttSub = new MQTT() +mqttSub.setHost(host, 1883) +mqttSub.setCleanSession(false) +mqttSub.setClientId("FlapSubscriberClient") + +BlockingConnection publisher = mqttPub.blockingConnection() +publisher.connect() +println "Publisher connected to ${host}" + +BlockingConnection subscriber = mqttSub.blockingConnection() +subscriber.connect() +println "${subscriber} connected to ${host}" +Topic[] topics = [new Topic("/topic", QoS.AT_MOST_ONCE)] +byte[] qoses = subscriber.subscribe(topics) +println "${subscriber} Subscribed to ${topics}" + +subscriber.disconnect() +println "${subscriber} disconnects immediately" + +println "${publisher} publishes QoS 2 messages on /topic" +publisher.publish("/topic", 'Hello world!!'.bytes, QoS.EXACTLY_ONCE, false) + + +subscriber = mqttSub.blockingConnection() +subscriber.connect() +println "${subscriber} reconnects to ${host}" + +Message message = subscriber.receive() +assert "/topic" == message.topic +String msgBody = new String(message.payload) +assert "Hello world!!" == msgBody +message.ack() +println "${subscriber} received on topic ${message.topic} the message \n\t ${msgBody}" +subscriber.disconnect() +println "${subscriber} disconnects again" + +subscriber = mqttSub.blockingConnection() +subscriber.connect() +println "${subscriber} reconnects again to ${host}" + +println "Wait receive something..." +message = subscriber.receive() +msgBody = new String(message.payload) +assert "Hello world!!" == msgBody +message.ack() +println "${subscriber} received again on topic ${message.topic} the message \n\t ${msgBody}" + +//aclient (clear session false) +//bclient (clear session false) +// +//a -> connection Server +//b -> connection Server +//b -> subscriptions (qos =0) +//b -> distconnection or lostconnection +// +//a ->publish ("abc/topic","test1",qos=2,false); +// +//... +//a ->publish ("abc/topic","test10",qos=2,false); +// +//b -> connection Server +//received message test1 +//... +//received message test10 +// +//b-> distconnection Server or loast connection +//b->connnection Server +//received message test1 +//... +//received message test10 +// +//why repeat received message diff --git a/tools_scripts/list-broker-topics.rb b/tools_scripts/integration/list-broker-topics.rb similarity index 100% rename from tools_scripts/list-broker-topics.rb rename to tools_scripts/integration/list-broker-topics.rb diff --git a/tools_scripts/paho_test_issue8.groovy b/tools_scripts/integration/paho_test_issue8.groovy similarity index 100% rename from tools_scripts/paho_test_issue8.groovy rename to tools_scripts/integration/paho_test_issue8.groovy diff --git a/tools_scripts/integration/twoClientsWithSameID.groovy b/tools_scripts/integration/twoClientsWithSameID.groovy new file mode 100644 index 00000000..3269df0e --- /dev/null +++ b/tools_scripts/integration/twoClientsWithSameID.groovy @@ -0,0 +1,56 @@ +@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.0.1', ext='jar') + + +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 + +String host = args[0] + +String tmpDir = System.getProperty("java.io.tmpdir") +MqttDefaultFilePersistence dataStore1 = new MqttDefaultFilePersistence(tmpDir + "/client1") +MqttClient client1 = new MqttClient("tcp://${host}:1883", "ConnectedClient", dataStore1) +client1.connect() +println "Client1 connected" +sleep(3000) + +//now connect a second client but with the same ID +MqttDefaultFilePersistence dataStore2 = new MqttDefaultFilePersistence(tmpDir + "/client2") +MqttClient client2 = new MqttClient("tcp://${host}:1883", "ConnectedClient", dataStore2) +client2.connect() +println "Client2 connected" +try { + client1.subscribe("topic", 2) + println "BAD! Client1 subscribed to topic but its session has to be dropped out by client2" +} catch (Exception ex) { + println "Ok, client1 has been dropped out" +} +sleep(3000) + +//now connect a second client but with the same ID +MqttDefaultFilePersistence dataStorePublisher = new MqttDefaultFilePersistence(tmpDir + "/publisher") +MqttClient publisher = new MqttClient("tcp://${host}:1883", "Publisher", dataStorePublisher) +publisher.connect() +println "publisher connected" +client2.subscribe("topic", 2) +println "Client2 subscribed to topic" +sleep(3000) + +print "Publisher is going to publish.." +publisher.publish('topic', 'Hello world!!'.bytes, 2, false) +println "Ok mate, done!" + +try { + client1.disconnect() + println "BAD! Client1 has been able to call disconnect, but it has to be already dropped out" +} catch (Exception ex) { + println "Ok, client1 can't disconnect because it was already disconnected" +} + +client2.disconnect() +println "Client2 disconnected" + +publisher.disconnect() +println "publisher disconnected" \ No newline at end of file diff --git a/tools_scripts/integration/webSocketClient.groovy b/tools_scripts/integration/webSocketClient.groovy new file mode 100644 index 00000000..2f992023 --- /dev/null +++ b/tools_scripts/integration/webSocketClient.groovy @@ -0,0 +1,121 @@ +@Grab(group='org.eclipse.jetty.websocket', module='websocket-client', version='9.2.0.M1', ext='jar') +@Grab(group='org.slf4j', module='slf4j-api', version='1.7.7') +@Grab(group='org.slf4j', module='slf4j-simple', version='1.7.7') + +import java.nio.ByteBuffer; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.eclipse.jetty.websocket.api.Session; +import org.eclipse.jetty.websocket.api.StatusCode; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError; +import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage; +import org.eclipse.jetty.websocket.api.annotations.WebSocket; +import org.eclipse.jetty.websocket.client.WebSocketClient; +import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +@WebSocket +public class MQTTWebSocket { + + private static final Logger LOG = LoggerFactory.getLogger(MQTTWebSocket.class); + + private Session session; + + private final CountDownLatch connectSentLatch; + + public MQTTWebSocket() { + this.connectSentLatch = new CountDownLatch(1); + } + + @OnWebSocketClose + public void onClose(int statusCode, String reason) { + System.out.printf("onClose - Connection closed: %d - %s%n", statusCode, reason); + this.session = null; + } + + @OnWebSocketConnect + public void onConnect(Session session) throws IOException { + System.out.println("onConnect - Got connect"); + this.session = session; + + //Trivial MQTT Connect message +// assertEquals(12, out.readByte()); //remaining length +// verifyString("MQIsdp", out); +// assertEquals(0x03, out.readByte()); //protocol version +// assertEquals(0x32, out.readByte()); //flags +// assertEquals(2, out.readByte()); //keepAliveTimer msb +// assertEquals(0, out.readByte()); //keepAliveTimer lsb + + byte[] rawMsg = [ 0x10, //message type + 0x11, //remaining length 17 bytes + 0x00, 0x06, //MQIspd length + 'M' as char, 'Q' as char, 'I' as char, 's' as char,'d' as char, 'p' as char, + 0x03, //protocol version + 0x32, //flags + 0x00, 0x0A,// keepAlive Timer, 10 seconds + 0x00, 0x03, 0x41, 0x42, 0x43 // A, B, C as client ID + ] as byte[] + ByteBuffer msg = ByteBuffer.wrap(rawMsg); + + //DBG + println "onConnect - send bytes " + rawMsg.encodeHex().toString() + //DBG + +// try { + session.getRemote().sendBytes(msg); + println("onConnect - Sent connect bytes"); + this.connectSentLatch.countDown(); + //session.close(StatusCode.NORMAL, "I'm done"); + println("onConnect - After connect"); + + println "send disconnect" + disconnect() + } + + public boolean awaitConnected(int duration, TimeUnit unit) throws InterruptedException { + return this.connectSentLatch.await(duration, unit); + } + + /** + * Send a disconnect message to the server + * */ + public void disconnect() { + byte[] rawMsg = [0xE0, 0x00] as byte[] + println "disconnect - send bytes " + rawMsg.encodeHex().toString() + ByteBuffer msg = ByteBuffer.wrap(rawMsg); + this.session.getRemote().sendBytes(msg) + } + + @OnWebSocketMessage + public void onMessage(Session session, byte[] buf, int offset, int length) { + System.out.printf("onMessage - Got msg: %s%n", buf); + } + + @OnWebSocketError + public void onError(Session session, Throwable cause) { + System.out.printf("onError - Got exception: %s%n", cause); + } +} + + +String destUri = "ws://localhost:8080/mqtt" +//String destUri = "ws://test.mosquitto.org:8080/" +WebSocketClient client = new WebSocketClient() +MQTTWebSocket socket = new MQTTWebSocket(); +client.start(); +URI echoUri = new URI(destUri); +ClientUpgradeRequest request = new ClientUpgradeRequest(); +client.connect(socket, echoUri, request); +println "main - Connecting to : ${echoUri}" +boolean connected = socket.awaitConnected(4, TimeUnit.SECONDS); +println "main - After wait for awaitConnected was ${connected}" + +//println "send disconnect" +//socket.disconnect() diff --git a/tools_scripts/integration/web_socket_test.html b/tools_scripts/integration/web_socket_test.html new file mode 100644 index 00000000..05c36704 --- /dev/null +++ b/tools_scripts/integration/web_socket_test.html @@ -0,0 +1,101 @@ + + + + + + + + + + + + +

Example Web Messaging web page.

+
+
+ Connect + Make a connection to the server, and set up a call back used if a + message arrives for this client. +
+ + +
+ +
+ Subscribe + Make a subscription to topic "/World". +
+
+ +
+ Send + Create a Message object containing the word "Hello" and then publish it at + the server. +
+ +
+ +
+ Receive + A copy of the published Message is received in the callback we created earlier. + +
+ +
+ Disconnect + Now disconnect this client from the server. +
+
+
+ +