From c4f186b3c8be3f750c9a17ddd3b54f6031f405e6 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Mon, 10 Sep 2012 23:30:40 +0200 Subject: [PATCH] Added rough implementation for QoS 2 publish --- .../messaging/spi/IStorageService.java | 6 ++ .../spi/impl/HawtDBStorageService.java | 25 ++++- .../messaging/spi/impl/SimpleMessaging.java | 91 +++++++++++++++---- .../spi/impl/DummyStorageService.java | 12 +++ .../server/ServerIntegrationFuseTest.java | 24 +++++ 5 files changed, 140 insertions(+), 18 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java index 590cdd40..16b346af 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -39,4 +39,10 @@ public interface IStorageService { List retrieveAllSubscriptions(); void close(); + + void persistQoS2Message(String publishKey, PublishEvent evt); + + void removeQoS2Message(String publishKey); + + PublishEvent retrieveQoS2Message(String publishKey); } 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 4675b9d0..88e1386e 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 @@ -52,6 +52,8 @@ public class HawtDBStorageService implements IStorageService { private SortedIndex m_retainedStore; //bind clientID+MsgID -> evt message published private SortedIndex m_inflightStore; + //bind clientID+MsgID -> evt message published + private SortedIndex m_qos2Store; //persistent Map of clientID, set of Subscriptions private SortedIndex> m_persistentSubscriptions; @@ -81,6 +83,7 @@ public class HawtDBStorageService implements IStorageService { initPersistentMessageStore(); initInflightMessageStore(); initPersistentSubscriptions(); + initPersistentQoS2MessageStore(); } private void initRetainedStore() { @@ -88,7 +91,6 @@ public class HawtDBStorageService implements IStorageService { indexFactory.setKeyCodec(StringCodec.INSTANCE); m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", indexFactory); - } @@ -117,6 +119,13 @@ public class HawtDBStorageService implements IStorageService { m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); } + private void initPersistentQoS2MessageStore() { + BTreeIndexFactory indexFactory = new BTreeIndexFactory(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_qos2Store = (SortedIndex) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory); + } + public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { if (message.length == 0) { //clean the message from topic @@ -202,4 +211,18 @@ public class HawtDBStorageService implements IStorageService { LOG.error(null, ex); } } + + /*-------- QoS 2 storage management --------------*/ + public void persistQoS2Message(String publishKey, PublishEvent evt) { + LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt)); + m_qos2Store.put(publishKey, evt); + } + + public void removeQoS2Message(String publishKey) { + m_qos2Store.remove(publishKey); + } + + public PublishEvent retrieveQoS2Message(String publishKey) { + return m_qos2Store.get(publishKey); + } } 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 23f50bd7..9336c689 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 @@ -19,6 +19,7 @@ import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; +import org.dna.mqtt.moquette.proto.PubCompMessage; import org.dna.mqtt.moquette.proto.messages.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.ConnectionDescriptor; @@ -178,6 +179,10 @@ public class SimpleMessaging implements IMessaging, EventHandler { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); + } else if (message instanceof PubRelMessage) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + int messageID = ((PubRelMessage) message).getMessageID(); + processPubRel(clientID, messageID); } else { throw new RuntimeException("Illegal message received " + message); } @@ -274,6 +279,30 @@ public class SimpleMessaging implements IMessaging, EventHandler { session.write(okResp); } + /** + * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish + * to all interested subscribers. + * */ + protected void processPubRel(String clientID, int messageID) { + String publishKey = String.format("%s%d", clientID, messageID); + PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); + + final String topic = evt.getTopic(); + final QOSType qos = evt.getQos(); + final byte[] message = evt.getMessage(); + boolean retain = evt.isRetain(); + + publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + + m_storageService.removeQoS2Message(publishKey); + + if (retain) { + m_storageService.storeRetained(topic, message, qos); + } + + sendPubComp(clientID, messageID); + } + protected void processPublish(PublishEvent evt) { LOG.debug("processPublish invoked with " + evt); final String topic = evt.getTopic(); @@ -286,24 +315,14 @@ public class SimpleMessaging implements IMessaging, EventHandler { //store the temporary message publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); m_storageService.addInFlight(evt, publishKey); + } else if (qos == QOSType.EXACTLY_ONCE) { + publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); + //store the message in temp store + m_storageService.persistQoS2Message(publishKey, evt); + sendPubRec(evt.getClientID(), evt.getMessageID()); } - - for (final Subscription sub : subscriptions.matches(topic)) { - if (qos == QOSType.MOST_ONE) { - //QoS 0 - notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); - } else { - //QoS 1 or 2 - //if the target subscription is not clean session and is not connected => store it - if (!sub.isCleanSession() && !sub.isActive()) { - //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.getClientId(), topic, qos, message, false)); - } - } - } + + publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); if (qos == QOSType.LEAST_ONE) { assert publishKey != null; @@ -316,6 +335,28 @@ public class SimpleMessaging implements IMessaging, EventHandler { } } + /** + * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 + * */ + private void publish2Subscribers(String topic, QOSType qos, byte[] message, boolean retain, Integer messageID) { + for (final Subscription sub : subscriptions.matches(topic)) { + if (qos == QOSType.MOST_ONE) { + //QoS 0 + notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); + } else { + //QoS 1 or 2 + //if the target subscription is not clean session and is not connected => store it + if (!sub.isCleanSession() && !sub.isActive()) { + //clone the event with matching clientID + PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); + m_storageService.storePublishForFuture(newPublishEvt); + } else { + notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); + } + } + } + } + private void subscribeSingleTopic(Subscription newSubscription, final String topic) { subscriptions.add(newSubscription); @@ -455,4 +496,20 @@ public class SimpleMessaging implements IMessaging, EventHandler { LOG.error(null, t); } } + + private void sendPubRec(String clientID, int messageID) { + LOG.debug(String.format("sendPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + PubRecMessage pubRecMessage = new PubRecMessage(); + pubRecMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubRecMessage); + } + + private void sendPubComp(String clientID, int messageID) { + LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); + PubCompMessage pubCompMessage = new PubCompMessage(); + pubCompMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubCompMessage); + } } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java index fd0de890..e4849ce8 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java @@ -60,4 +60,16 @@ public class DummyStorageService implements IStorageService { public void close() { //To change body of implemented methods use File | Settings | File Templates. } + + public void persistQoS2Message(String publishKey, PublishEvent evt) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void removeQoS2Message(String publishKey) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public PublishEvent retrieveQoS2Message(String publishKey) { + return null; //To change body of implemented methods use File | Settings | File Templates. + } } 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 index 5b2d75df..1789eb4c 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -121,4 +121,28 @@ public class ServerIntegrationFuseTest { futurePublish.await(); connectionPub.disconnect().await(); } + + + @Test + public void testPublishWithQoS2() throws Exception { + m_mqtt.setClientId("TestClient"); + m_connection = m_mqtt.futureConnection(); + m_connection.connect().await(); + + //subscribe to a QoS 2 topic + Topic[] topics = {new Topic("/topic", QoS.EXACTLY_ONCE)}; + Future subscribeFuture = m_connection.subscribe(topics); + byte[] qoses = subscribeFuture.await(); + + Future receive = m_connection.receive(); + + //publish a QoS 2 message + Future f3 = m_connection.publish("/topic", "Hello MQTT".getBytes(), QoS.EXACTLY_ONCE, false); + + //verify the reception + Message message = receive.await(); + message.ack(); + + assertEquals("Hello MQTT", new String(message.getPayload())); + } }