From 90259d414861b8b2188ec8372c1c284b0decd669 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 31 Aug 2012 16:56:46 +0200 Subject: [PATCH] First handling of QoS 1 level publish --- .../moquette/messaging/spi/IMessaging.java | 5 +- .../messaging/spi/impl/SimpleMessaging.java | 67 +++++++++++++------ .../spi/impl/events/CleanInFlightEvent.java | 17 +++++ .../spi/impl/events/PublishEvent.java | 21 +++++- .../dna/mqtt/moquette/server/MQTTHandler.java | 5 +- .../dna/mqtt/moquette/server/Notifier.java | 9 ++- .../org/dna/mqtt/moquette/server/Server.java | 3 +- 7 files changed, 100 insertions(+), 27 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CleanInFlightEvent.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 bf5623ee..fe762252 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 @@ -1,6 +1,7 @@ package org.dna.mqtt.moquette.messaging.spi; import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** @@ -16,7 +17,7 @@ public interface IMessaging { * notification also after all the current subscriber to the given topic has * been notified. */ - void publish(String topic, byte[] message, QOSType qos, boolean retain); + void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID); /** * Subscribe a client to a specified topic with a defined level @@ -38,4 +39,6 @@ public interface IMessaging { void unsubscribe(String topic, String clientID); void disconnect(IoSession session); + + void refill(MessagingEvent evt) throws InterruptedException; } 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 75d4c254..44a69fb4 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 @@ -12,14 +12,7 @@ import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; -import org.dna.mqtt.moquette.messaging.spi.impl.events.CloseEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.UnsubscribeEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; @@ -66,6 +59,8 @@ public class SimpleMessaging implements IMessaging, Runnable { private MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; private SortedIndex m_retainedStore; + //bind clientID+MsgID -> evt message published + private SortedIndex m_inflightStore; public SimpleMessaging() { String storeFile = Server.STORAGE_FILE_PATH; @@ -86,6 +81,7 @@ public class SimpleMessaging implements IMessaging, Runnable { subscriptions.init(m_multiIndexFactory); initRetainedMessageStore(); + initPersistentMessageStore(); } private void initRetainedMessageStore() { @@ -96,17 +92,28 @@ public class SimpleMessaging implements IMessaging, Runnable { m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", indexFactory); } + /** + * Initialize the message store used to handle the temporary storage of QoS 1,2 + * messages. + */ + private void initPersistentMessageStore() { + BTreeIndexFactory indexFactory = new BTreeIndexFactory(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); + } + public void run() { eventLoop(); } - public void publish(String topic, byte[] message, QOSType qos, boolean retain) { + public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID) { //TODO for this moment no qos > 0 trivial implementation - QOSType defQos = QOSType.MOST_ONE; +// QOSType defQos = QOSType.MOST_ONE; try { - m_inboundQueue.put(new PublishEvent(topic, defQos, message, retain)); + m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID)); } catch (InterruptedException ex) { LOG.error(null, ex); } @@ -139,6 +146,11 @@ public class SimpleMessaging implements IMessaging, Runnable { LOG.error(null, ex); } } + + //method used by hte Notifier to re-put an event on the inbound queue + public void refill(MessagingEvent evt) throws InterruptedException { + m_inboundQueue.put(evt); + } /** @@ -198,8 +210,8 @@ public class SimpleMessaging implements IMessaging, Runnable { public BlockingQueue getNotifyEventQueue() { return m_outboundQueue; } - - + + protected void eventLoop() { LOG.debug("Started event loop"); boolean interrupted = false; @@ -218,6 +230,9 @@ public class SimpleMessaging implements IMessaging, Runnable { processClose(); } else if (evt instanceof DisconnectEvent) { m_outboundQueue.put(evt); + } else if (evt instanceof CleanInFlightEvent) { + //remove the message from inflight storage + m_inflightStore.remove(((CleanInFlightEvent)evt).getMsgId()); } } catch (InterruptedException ex) { processClose(); @@ -226,20 +241,30 @@ public class SimpleMessaging implements IMessaging, Runnable { } } - protected void processPublish(PublishEvent evt) { + protected void processPublish(PublishEvent evt) throws InterruptedException { LOG.debug("processPublish invoked"); final String topic = evt.getTopic(); final QOSType qos = evt.getQos(); final byte[] message = evt.getMessage(); boolean retain = evt.isRetain(); + + CleanInFlightEvent cleanEvt = null; + + if (qos == QOSType.LEAST_ONE) { + //store the temporary message + String publishKey = evt.getClientID() + evt.getMessageID(); + m_inflightStore.put(publishKey, evt); + cleanEvt = new CleanInFlightEvent(publishKey); + } for (final Subscription sub : subscriptions.matches(topic)) { - try { - m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); - } catch (InterruptedException ex) { - LOG.error(null, ex); - } + m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); } + + if (cleanEvt != null) { + m_outboundQueue.put(cleanEvt); + } + if (retain) { if (message.length == 0) { //clean the message from topic @@ -250,8 +275,8 @@ public class SimpleMessaging implements IMessaging, Runnable { } } } - - + + protected void processSubscribe(SubscribeEvent evt)/* throws InterruptedException*/ { LOG.debug("processSubscribe invoked"); Subscription newSubscription = evt.getSubscription(); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CleanInFlightEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CleanInFlightEvent.java new file mode 100644 index 00000000..b56ea194 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CleanInFlightEvent.java @@ -0,0 +1,17 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * Message to remove the message of type QoS 1 or 2 from in flight store + */ +public class CleanInFlightEvent extends MessagingEvent { + + private String m_msgId; + + public CleanInFlightEvent(String msgId) { + m_msgId = msgId; + } + + public String getMsgId() { + return m_msgId; + } +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index 355a56a3..86a9954b 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -11,12 +11,23 @@ public class PublishEvent extends MessagingEvent { QOSType m_qos; byte[] m_message; boolean m_retain; + String m_clientID; + //Optional attribute, available only fo QoS 1 and 2 + String m_msgID; - public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain) { + public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, + String clientID) { m_topic = topic; m_qos = qos; m_message = message; m_retain = retain; + m_clientID = clientID; + } + + public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, + String clientID, String msgID) { + this(topic, qos, message, retain, clientID); + m_msgID = msgID; } public String getTopic() { @@ -34,4 +45,12 @@ public class PublishEvent extends MessagingEvent { public boolean isRetain() { return m_retain; } + + public String getClientID() { + return m_clientID; + } + + public String getMessageID() { + return m_msgID; + } } 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 367cbef0..8ae838ba 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 @@ -120,7 +120,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { if (msg.isWillFlag()) { QOSType willQos = QOSType.values()[msg.getWillQos()]; m_messaging.publish(msg.getWillTopic(), msg.getWillMessage().getBytes(), - willQos, msg.isWillRetain()); + willQos, msg.isWillRetain(), msg.getClientID()); } //handle user authentication @@ -184,8 +184,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { } protected void handlePublish(IoSession session, PublishMessage message) { + String clientID = (String) session.getAttribute(ATTR_CLIENTID); m_messaging.publish(message.getTopicName(), message.getPayload(), - message.getQos(), message.isRetainFlag()); + message.getQos(), message.isRetainFlag(), clientID); } protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) { diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java index 676184b9..850b58b2 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java @@ -1,7 +1,10 @@ package org.dna.mqtt.moquette.server; import java.util.concurrent.BlockingQueue; + +import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.INotifier; +import org.dna.mqtt.moquette.messaging.spi.impl.events.CleanInFlightEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; @@ -18,10 +21,12 @@ class Notifier implements Runnable { BlockingQueue m_queue; INotifier m_notifier; + IMessaging m_messaging; - Notifier(BlockingQueue queue, INotifier notifier) { + Notifier(BlockingQueue queue, INotifier notifier, IMessaging messaging) { m_queue = queue; m_notifier = notifier; + m_messaging = messaging; } public void run() { @@ -37,6 +42,8 @@ class Notifier implements Runnable { } else if (evt instanceof DisconnectEvent) { DisconnectEvent evtD = (DisconnectEvent) evt; m_notifier.disconnect(evtD.getSession()); + } else if (evt instanceof CleanInFlightEvent) { + m_messaging.refill(evt); } } catch (InterruptedException ex) { interrupted = true; diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index 91b2a64d..e1ebefc7 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -94,7 +94,8 @@ public class Server { BlockingQueue notifyQueue = messaging.getNotifyEventQueue(); for (int i = 0; i < NOTIFIER_POOL_SIZE; i++) { - m_notifierPool.submit(new Notifier(notifyQueue, handler)); + Notifier notifier = new Notifier(notifyQueue, handler, messaging); + m_notifierPool.submit(notifier); } m_acceptor.setHandler(handler);