diff --git a/parser/pom.xml b/parser/pom.xml index 5a4ec1ff..9cac38fa 100644 --- a/parser/pom.xml +++ b/parser/pom.xml @@ -64,6 +64,20 @@ 1.9.0 test + + + org.fusesource.mqtt-client + mqtt-client + + 1.3 + test + + + + org.fusesource.hawtbuf + hawtbuf + 1.9 + @@ -107,10 +121,21 @@ - + --> + + diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index 966507ab..de8107a2 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -7,6 +7,7 @@ import java.util.concurrent.TimeUnit; import org.dna.mqtt.moquette.client.Client; import org.dna.mqtt.moquette.client.IPublishCallback; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.fusesource.mqtt.client.*; import org.junit.After; import static org.junit.Assert.*; import org.junit.Before; @@ -41,6 +42,25 @@ public class ServerIntegrationTest { } } + @Test + public void testSubscribe_FSClient() throws Exception { +// startServer(); + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", Server.PORT); + mqtt.setClientId("TestClient"); + FutureConnection connection = mqtt.futureConnection(); + connection.connect().await(); + + Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)}; + Future futSub = connection.subscribe(topics); + + connection.publish("/topic", "Test my payload".getBytes(), QoS.AT_MOST_ONCE, false).await(); + + byte[] qoses = futSub.await(); + assertEquals(1, qoses.length); + connection.disconnect().await(); + } + @Test public void testSubscribe() throws IOException, InterruptedException {