diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index 2e9e03f9..45181780 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -34,8 +34,7 @@ public class HawtDBStorageService implements IStorageService { //bind clientID+MsgID -> evt message published private SortedIndex m_inflightStore; - //persistent Map of clientID, list of Subscriptions - //TODO move to Set to avoid double subscriptions + //persistent Map of clientID, set of Subscriptions private SortedIndex> m_persistentSubscriptions; public HawtDBStorageService() { diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 626572a9..c3f2a122 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -125,6 +125,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { public void republishStored(String clientID) { //create the event to push + LOG.debug("republishStored invoked to publish soterd messages for clientID " + clientID); disruptorPublish(new RepublishEvent(clientID)); } @@ -176,8 +177,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { } public void onEvent(ValueEvent t, long l, boolean bln) throws Exception { - LOG.debug("onEvent processing event"); MessagingEvent evt = t.getEvent(); + LOG.debug("onEvent processing messaging event " + evt); if (evt instanceof PublishEvent) { processPublish((PublishEvent) evt); } else if (evt instanceof StopEvent) { @@ -307,6 +308,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { } } + subscriptions.connect(msg.getClientID()); + //handle clean session flag if (msg.isCleanSession()) { //remove all prev subscriptions @@ -323,7 +326,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } protected void processPublish(PublishEvent evt) { - LOG.debug("processPublish invoked"); + LOG.debug("processPublish invoked with " + evt); final String topic = evt.getTopic(); final QOSType qos = evt.getQos(); final byte[] message = evt.getMessage(); @@ -346,9 +349,12 @@ public class SimpleMessaging implements IMessaging, EventHandler { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it if (!sub.isCleanSession() && !sub.isActive()) { - m_storageService.storePublishForFuture(evt); + //clone the event with matching clientID + PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), evt.getMessageID(), null); + m_storageService.storePublishForFuture(newPublishEvt); + } else { + notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); } - notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); } } @@ -396,8 +402,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { //TODO by now it handles only QoS 0 messages for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(QOSType.MOST_ONE); + ackMessage.addType(QOSType.LEAST_ONE); } - LOG.info("replying with SubAct to MSG ID {0}", msg.getMessageID()); + LOG.info("replying with SubAct to MSG ID " + msg.getMessageID()); session.write(ackMessage); } @@ -447,8 +454,10 @@ public class SimpleMessaging implements IMessaging, EventHandler { } private void processRepublish(RepublishEvent evt) throws InterruptedException { + LOG.debug("processRepublish invoked"); List publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID()); if (publishedEvents == null) { + LOG.debug("processRepublish, no stored publish events"); return; } @@ -470,7 +479,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { pubMessage.setMessageID(evt.getMessageID()); } - LOG.debug("notify invoked"); try { assert m_clientIDs != null; LOG.debug("clientIDs are " + m_clientIDs); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java new file mode 100644 index 00000000..515b7408 --- /dev/null +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -0,0 +1,135 @@ +package org.dna.mqtt.moquette.server; + +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 org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; + +import static org.junit.Assert.*; + + +/** + * Test the broker using the widely used FuseSource libraries + * */ +public class ServerIntegrationFuseTest { + + boolean received; + Server server; + + protected void startServer() throws IOException { + server = new Server(); + server.startServer(); + } + + @Before + public void setUp() throws IOException { + startServer(); + } + + @After + public void tearDown() { + server.stopServer(); + File dbFile = new File(Server.STORAGE_FILE_PATH); + if (dbFile.exists()) { + dbFile.delete(); + } + } + + @Ignore + public void testSubscribe() 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(); + } + + @Ignore + public void testPublishWithQoS1() throws Exception { + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", Server.PORT); + mqtt.setClientId("TestClient"); + FutureConnection connection = mqtt.futureConnection(); + connection.connect().await(); + + //subscribe to a QoS 1 topic + Topic[] topics = {new Topic("/topic", QoS.AT_LEAST_ONCE)}; + Future subscribeFuture = connection.subscribe(topics); + byte[] qoses = subscribeFuture.await(); + + Future receive = connection.receive(); + + //publish a QoS 1 message + Future f3 = connection.publish("/topic", "Hello MQTT".getBytes(), QoS.AT_LEAST_ONCE, false); + + //verify the reception + Message message = receive.await(); + message.ack(); + + assertEquals("Hello MQTT", new String(message.getPayload())); + + connection.disconnect().await(); + } + + @Test + public void testPublishWithQoS1_notCleanSession() throws Exception { + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", Server.PORT); + mqtt.setClientId("TestClient"); + mqtt.setCleanSession(false); + FutureConnection connection = mqtt.futureConnection() ; + connection.connect().await(); + + //subscribe to a QoS 1 topic + Topic[] topics = {new Topic("/topic", QoS.AT_LEAST_ONCE)}; + Future subscribeFuture = connection.subscribe(topics); + byte[] qoses = subscribeFuture.await(); + + connection.disconnect().await(); + + //publish a QoS 1 message another client publish a message on the topic + publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), QoS.AT_LEAST_ONCE); + + + //Reconnect + connection = mqtt.futureConnection(); + connection.connect().await(); + Future receive = connection.receive(); + + //verify the reception + Message message = receive.await(); + message.ack(); + + assertEquals("Hello MQTT", new String(message.getPayload())); + + connection.disconnect().await(); + } + + private void publishFromAnotherClient(String topic, byte[] payload, QoS qos) throws Exception { + MQTT mqttPub = new MQTT(); + mqttPub.setHost("localhost", Server.PORT); + mqttPub.setClientId("TestClientPUB"); + FutureConnection connectionPub = mqttPub.futureConnection(); + connectionPub.connect().await(); + Future futurePublish = connectionPub.publish(topic, payload, qos, false); + futurePublish.await(); + connectionPub.disconnect().await(); + } +}