From e5fbf5c655d0be7ed6a6e0f1c9c581c6ce8d9ed9 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 12:33:06 +0200 Subject: [PATCH] Added the storage of QoS1 messages for not clean sessions and the refire once the client reconnects --- .../moquette/messaging/spi/IMessaging.java | 2 + .../messaging/spi/impl/SimpleMessaging.java | 69 ++++++++++++++++--- .../spi/impl/events/RepublishEvent.java | 16 +++++ .../dna/mqtt/moquette/server/Constants.java | 2 +- .../dna/mqtt/moquette/server/MQTTHandler.java | 5 +- 5 files changed, 83 insertions(+), 11 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RepublishEvent.java diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 4f86fb44..97a1d6e6 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -47,4 +47,6 @@ public interface IMessaging { void disconnect(IoSession session); void refill(MessagingEvent evt) throws InterruptedException; + + void republishStored(String clientID); } 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 b0a4dde3..ff5d84dc 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 @@ -4,6 +4,7 @@ import java.io.File; import java.io.IOException; import java.io.Serializable; import java.text.ParseException; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.BlockingQueue; @@ -62,6 +63,8 @@ public class SimpleMessaging implements IMessaging, Runnable { private SortedIndex m_retainedStore; //bind clientID+MsgID -> evt message published private SortedIndex m_inflightStore; + //maps clientID to the list of pending messages stored + private SortedIndex> m_persistentMessageStore; public SimpleMessaging() { String storeFile = Server.STORAGE_FILE_PATH; @@ -82,6 +85,8 @@ public class SimpleMessaging implements IMessaging, Runnable { subscriptions.init(m_multiIndexFactory); initRetainedMessageStore(); + initInflightMessageStore(); + //init the message store for QoS 1/2 messages in clean sessions initPersistentMessageStore(); } @@ -95,14 +100,21 @@ public class SimpleMessaging implements IMessaging, Runnable { /** * Initialize the message store used to handle the temporary storage of QoS 1,2 - * messages. + * messages in flight. */ - private void initPersistentMessageStore() { + private void initInflightMessageStore() { BTreeIndexFactory indexFactory = new BTreeIndexFactory(); indexFactory.setKeyCodec(StringCodec.INSTANCE); m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); } + + private void initPersistentMessageStore() { + BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); + } public void run() { eventLoop(); @@ -110,9 +122,6 @@ public class SimpleMessaging implements IMessaging, Runnable { public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session) { - //TODO for this moment no qos > 0 trivial implementation -// QOSType defQos = QOSType.MOST_ONE; - try { m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, session)); } catch (InterruptedException ex) { @@ -161,6 +170,15 @@ public class SimpleMessaging implements IMessaging, Runnable { m_inboundQueue.put(evt); } + public void republishStored(String clientID) { + //create the event to push + try { + m_inboundQueue.put(new RepublishEvent(clientID)); + } catch (InterruptedException iex) { + LOG.error(null, iex); + } + } + /** * Verify if the 2 topics matching respecting the rules of MQTT Appendix A */ @@ -237,11 +255,12 @@ public class SimpleMessaging implements IMessaging, Runnable { } else if (evt instanceof CloseEvent) { processClose(); } else if (evt instanceof DisconnectEvent) { - m_outboundQueue.put(evt); + processDisconnect((DisconnectEvent) evt); } else if (evt instanceof CleanInFlightEvent) { //remove the message from inflight storage m_inflightStore.remove(((CleanInFlightEvent)evt).getMsgId()); - + } else if (evt instanceof RepublishEvent) { + processRepublish((RepublishEvent) evt); } } catch (InterruptedException ex) { processClose(); @@ -249,7 +268,7 @@ public class SimpleMessaging implements IMessaging, Runnable { } } } - + protected void processPublish(PublishEvent evt) throws InterruptedException { LOG.debug("processPublish invoked"); final String topic = evt.getTopic(); @@ -273,6 +292,17 @@ public class SimpleMessaging implements IMessaging, Runnable { } else { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it + if (!sub.isCleanSession() && !sub.isActive()) { + List storedEvents; + String clientID = evt.getClientID(); + if (!m_persistentMessageStore.containsKey(clientID)) { + storedEvents = new ArrayList(); + } else { + storedEvents = m_persistentMessageStore.get(clientID); + } + storedEvents.add(evt); + m_persistentMessageStore.put(clientID, storedEvents); + } m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID())); } } @@ -326,6 +356,17 @@ public class SimpleMessaging implements IMessaging, Runnable { protected void processRemoveAllSubscriptions(RemoveAllSubscriptionsEvent evt) { LOG.debug("processRemoveAllSubscriptions invoked"); subscriptions.removeForClient(evt.getClientID()); + + //remove also the messages stored of type QoS1/2 + m_persistentMessageStore.remove(evt.getClientID()); + } + + private void processDisconnect(DisconnectEvent evt) throws InterruptedException { + m_outboundQueue.put(evt); + + //de-activate the subscriptions for this ClientID + String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); + subscriptions.disconnect(clientID); } private void processClose() { @@ -336,4 +377,16 @@ public class SimpleMessaging implements IMessaging, Runnable { LOG.error(null, ex); } } + + private void processRepublish(RepublishEvent evt) throws InterruptedException { + List publishedEvents = m_persistentMessageStore.get(evt.getClientID()); + if (publishedEvents == null) { + return; + } + + for (PublishEvent pubEvt : publishedEvents) { + m_outboundQueue.put(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), + pubEvt.getMessage(), false, pubEvt.getMessageID())); + } + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RepublishEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RepublishEvent.java new file mode 100644 index 00000000..a410e8c4 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RepublishEvent.java @@ -0,0 +1,16 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * Used to re-fire stored QoS1 QoS2 events once a client reconnects. + */ +public class RepublishEvent extends MessagingEvent { + private String m_clientID; + + public RepublishEvent(String clientID) { + this.m_clientID = clientID; + } + + public String getClientID() { + return m_clientID; + } +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Constants.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Constants.java index 1fa8809b..0a2ea7fd 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Constants.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Constants.java @@ -4,6 +4,6 @@ package org.dna.mqtt.moquette.server; * Server constants keeper */ public class Constants { - protected static final String ATTR_CLIENTID = "ClientID"; + public static final String ATTR_CLIENTID = "ClientID"; public static final String CLEAN_SESSION = "cleanSession"; } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index cd2f5126..1d64d78c 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -130,13 +130,14 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { } } - //TODO for QoS1 and QoS remember to store all the messages arrived to - //the dormat subscriptionsonce the QoS1 and QoS2 //handle clean session flag if (msg.isCleanSession()) { //remove all prev subscriptions //cleanup topic subscriptions m_messaging.removeSubscriptions(msg.getClientID()); + } else { + //force the republish of stored QoS1 and QoS2 + m_messaging.republishStored(msg.getClientID()); } ConnAckMessage okResp = new ConnAckMessage();