From 90259d414861b8b2188ec8372c1c284b0decd669 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 31 Aug 2012 16:56:46 +0200 Subject: [PATCH 01/13] 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); From 376140cab8f9be96be4b40c30c283cecf83f74fe Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 31 Aug 2012 17:07:49 +0200 Subject: [PATCH 02/13] Fixed API used by test calls --- .../mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java | 2 +- .../test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index fe19a9af..811ce739 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -57,7 +57,7 @@ public class SimpleMessagingTest { // messaging.setNotifier(notifier); //Exercise - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI"); messaging.processPublish(pubEvt); //Verify diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java index 22a59916..22c79781 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -86,7 +86,7 @@ public class MQTTHandlerTest { //Verify assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), - any(AbstractMessage.QOSType.class), anyBoolean()); + any(AbstractMessage.QOSType.class), anyBoolean(), eq("123")); } From 585008e29da845be1752c902e552872fc2a86dfa Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 31 Aug 2012 17:14:40 +0200 Subject: [PATCH 03/13] Fixed API used by test calls --- .../java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java | 2 +- .../java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java index 15274f94..3cc5a178 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java @@ -38,7 +38,7 @@ public class PubAckDecoderTest { assertNotNull(m_mockProtoDecoder.getMessage()); assertEquals(MessageDecoder.OK, res); - assertEquals(messageId, m_mockProtoDecoder.getMessage().getMessageID()); + assertEquals(messageId, m_mockProtoDecoder.getMessage().getMessageID().intValue()); assertEquals(AbstractMessage.PUBACK, m_mockProtoDecoder.getMessage().getMessageType()); } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java index 5139f3ac..5de67eb0 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java @@ -39,7 +39,7 @@ public class SubAckDecoderTest { //Verify assertEquals(MessageDecoderResult.OK, res); - assertEquals(0xAABB, m_mockProtoDecoder.getMessage().getMessageID()); + assertEquals(0xAABB, m_mockProtoDecoder.getMessage().getMessageID().intValue()); List qoses = m_mockProtoDecoder.getMessage().types(); assertEquals(3, qoses.size()); assertEquals(QOSType.LEAST_ONE, qoses.get(0)); @@ -64,7 +64,7 @@ public class SubAckDecoderTest { assertNotNull(m_mockProtoDecoder.getMessage()); assertEquals(MessageDecoder.OK, res); SubAckMessage message = (SubAckMessage) m_mockProtoDecoder.getMessage(); - assertEquals(0x0A, message.getMessageID()); + assertEquals(0x0A, message.getMessageID().intValue()); assertEquals(1, message.types().size()); assertEquals(AbstractMessage.QOSType.LEAST_ONE, message.types().get(0)); } From 60a79ba9b7b222547ebbc4cbef45c780eb89fac4 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 31 Aug 2012 17:16:01 +0200 Subject: [PATCH 04/13] Changed PublishMessage to be a able to handle the messages ID for QoS > 0 cases --- .../mqtt/moquette/proto/messages/MessageIDMessage.java | 8 ++++---- .../dna/mqtt/moquette/proto/messages/PublishMessage.java | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java index 7f7a8712..27752d44 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java @@ -9,14 +9,14 @@ package org.dna.mqtt.moquette.proto.messages; * @author andrea */ public abstract class MessageIDMessage extends AbstractMessage { - int m_messageID; + private Integer m_messageID; //could be null if Qos is == 0 - public int getMessageID() { + public Integer getMessageID() { return m_messageID; } - public void setMessageID(int messageID) { + public void setMessageID(Integer messageID) { this.m_messageID = messageID; } - + } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java index c8b33555..41742e2a 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java @@ -4,19 +4,19 @@ package org.dna.mqtt.moquette.proto.messages; * * @author andrea */ -public class PublishMessage extends AbstractMessage { +public class PublishMessage extends MessageIDMessage { private String m_topicName; - private Integer m_messageID; //could be null if Qos is == 0 +// private Integer m_messageID; //could be null if Qos is == 0 private byte[] m_payload; - public Integer getMessageID() { + /*public Integer getMessageID() { return m_messageID; } public void setMessageID(Integer messageID) { this.m_messageID = messageID; - } + }*/ public String getTopicName() { return m_topicName; From cb95ecc7cb2eafe7f4e146cff40ed14f23820e0b Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 31 Aug 2012 18:04:31 +0200 Subject: [PATCH 05/13] Added QoS 1 management to Client part --- .../org/dna/mqtt/moquette/client/Client.java | 64 ++++++++++++------- .../moquette/client/ClientMQTTHandler.java | 15 +++-- 2 files changed, 50 insertions(+), 29 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java index 10f5bf57..141b8d29 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java @@ -24,17 +24,7 @@ import org.dna.mqtt.moquette.ConnectionException; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.PublishException; import org.dna.mqtt.moquette.SubscribeException; -import org.dna.mqtt.moquette.proto.ConnAckDecoder; -import org.dna.mqtt.moquette.proto.ConnectEncoder; -import org.dna.mqtt.moquette.proto.DisconnectEncoder; -import org.dna.mqtt.moquette.proto.PingReqEncoder; -import org.dna.mqtt.moquette.proto.PingRespDecoder; -import org.dna.mqtt.moquette.proto.PublishDecoder; -import org.dna.mqtt.moquette.proto.PublishEncoder; -import org.dna.mqtt.moquette.proto.SubAckDecoder; -import org.dna.mqtt.moquette.proto.SubscribeEncoder; -import org.dna.mqtt.moquette.proto.UnsubAckDecoder; -import org.dna.mqtt.moquette.proto.UnsubscribeEncoder; +import org.dna.mqtt.moquette.proto.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; @@ -120,6 +110,7 @@ public final class Client { decoder.addMessageDecoder(new SubAckDecoder()); decoder.addMessageDecoder(new UnsubAckDecoder()); decoder.addMessageDecoder(new PublishDecoder()); + decoder.addMessageDecoder(new PubAckDecoder()); decoder.addMessageDecoder(new PingRespDecoder()); DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); @@ -230,27 +221,46 @@ public final class Client { public void publish(String topic, byte[] payload) throws PublishException { publish(topic, payload, false); } - + + /** + * Publish by default with QoS 0 + * */ public void publish(String topic, byte[] payload, boolean retain) throws PublishException { + publish(topic, payload, AbstractMessage.QOSType.MOST_ONE, retain); + } + + public void publish(String topic, byte[] payload, AbstractMessage.QOSType qos, boolean retain) throws PublishException { PublishMessage msg = new PublishMessage(); msg.setRetainFlag(retain); msg.setTopicName(topic); msg.setPayload(payload); - //Untill the server could handle all the Qos levels - msg.setQos(AbstractMessage.QOSType.MOST_ONE); + //Untill the server could handle all the Qos 2 level + if (qos != AbstractMessage.QOSType.MOST_ONE) { + msg.setQos(AbstractMessage.QOSType.LEAST_ONE); + int messageID = m_messageIDGenerator.next(); + msg.setMessageID(messageID); - WriteFuture wf = m_session.write(msg); - try { - wf.await(); - } catch (InterruptedException ex) { - LOG.debug(null, ex); - throw new PublishException(ex); - } + try { + manageSendQoS1(msg); + } catch (Throwable ex) { + throw new MQTTException(ex); + } + } else { + //QoS 0 case + msg.setQos(AbstractMessage.QOSType.MOST_ONE); + WriteFuture wf = m_session.write(msg); + try { + wf.await(); + } catch (InterruptedException ex) { + LOG.debug(null, ex); + throw new PublishException(ex); + } - Throwable ex = wf.getException(); - if (ex != null) { - throw new PublishException(ex); + Throwable ex = wf.getException(); + if (ex != null) { + throw new PublishException(ex); + } } updatePinger(); @@ -368,6 +378,12 @@ public final class Client { m_receivedSubAckMessageID = messageID; } + void publishAckCallback(Integer messageID) { + LOG.info("publishAckCallback invoked"); + m_subscribeBarrier.countDown(); + m_receivedSubAckMessageID = messageID; + } + protected void publishCallback(String topic, byte[] payload) { IPublishCallback callback = m_subscribersList.get(topic); if (callback == null) { diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java index 74d7723b..bdf5ffaa 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java @@ -2,12 +2,10 @@ package org.dna.mqtt.moquette.client; import org.apache.mina.core.service.IoHandlerAdapter; import org.apache.mina.core.session.IoSession; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.*; + import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; -import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; -import org.dna.mqtt.moquette.proto.messages.SubAckMessage; -import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -45,11 +43,18 @@ public class ClientMQTTHandler extends IoHandlerAdapter { case PUBLISH: handlePublish(session, (PublishMessage) msg); break; + case PUBACK: + handlePublishAck(session, (PubAckMessage) msg); + break; case PINGRESP: break; } } + private void handlePublishAck(IoSession session, PubAckMessage msg) { + m_callback.publishAckCallback(msg.getMessageID()); + } + private void handleConnectAck(IoSession session, ConnAckMessage connAckMessage) { m_callback.connectionAckCallback(connAckMessage.getReturnCode()); } From 65e34ddc40d4157418d7b8cc01c1db05fec611a6 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 31 Aug 2012 19:23:53 +0200 Subject: [PATCH 06/13] Added QoS 1 pub ack to the server part --- .../moquette/messaging/spi/IMessaging.java | 6 ++ .../moquette/messaging/spi/INotifier.java | 9 ++- .../messaging/spi/impl/SimpleMessaging.java | 21 ++++-- .../spi/impl/events/NotifyEvent.java | 13 ++++ .../spi/impl/events/PubAckEvent.java | 24 +++++++ .../spi/impl/events/PublishEvent.java | 10 +-- .../dna/mqtt/moquette/server/MQTTHandler.java | 65 ++++++++++++++----- .../dna/mqtt/moquette/server/Notifier.java | 9 ++- .../org/dna/mqtt/moquette/server/Server.java | 20 +----- .../server/ServerIntegrationTest.java | 25 +++++++ 10 files changed, 152 insertions(+), 50 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.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 fe762252..cacf9da1 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 @@ -2,6 +2,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.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** @@ -18,6 +19,11 @@ public interface IMessaging { * been notified. */ void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID); + + /** + * Used to publish messages with QoS > 0 + * */ + void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID); /** * Subscribe a client to a specified topic with a defined level diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java index 13685f38..5196ca2e 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java @@ -1,7 +1,10 @@ package org.dna.mqtt.moquette.messaging.spi; import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PubAckMessage; /** * @@ -9,8 +12,10 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; */ public interface INotifier { - public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained); + public void notify(NotifyEvent evt/*String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained*/); public void disconnect(IoSession session); - + + void sendPubAck(PubAckEvent evt); + } 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 44a69fb4..05c27b8c 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 @@ -119,6 +119,14 @@ public class SimpleMessaging implements IMessaging, Runnable { } } + public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID) { + try { + m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID)); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } + } + public void subscribe(String clientId, String topic, QOSType qos) { Subscription newSubscription = new Subscription(clientId, topic, qos); try { @@ -151,8 +159,7 @@ public class SimpleMessaging implements IMessaging, Runnable { public void refill(MessagingEvent evt) throws InterruptedException { m_inboundQueue.put(evt); } - - + /** * Verify if the 2 topics matching respecting the rules of MQTT Appendix A */ @@ -233,6 +240,7 @@ public class SimpleMessaging implements IMessaging, Runnable { } else if (evt instanceof CleanInFlightEvent) { //remove the message from inflight storage m_inflightStore.remove(((CleanInFlightEvent)evt).getMsgId()); + } } catch (InterruptedException ex) { processClose(); @@ -252,17 +260,22 @@ public class SimpleMessaging implements IMessaging, Runnable { if (qos == QOSType.LEAST_ONE) { //store the temporary message - String publishKey = evt.getClientID() + evt.getMessageID(); + String publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); m_inflightStore.put(publishKey, evt); cleanEvt = new CleanInFlightEvent(publishKey); } for (final Subscription sub : subscriptions.matches(topic)) { - m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); + if (qos == QOSType.MOST_ONE) { + m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); + } else { + m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID())); + } } if (cleanEvt != null) { m_outboundQueue.put(cleanEvt); + m_outboundQueue.put(new PubAckEvent(evt.getMessageID(), evt.getClientID())); } if (retain) { diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java index 4fe6dc2e..ddbb138a 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java @@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** + * TODO it's identical to PublishEvent * * @author andrea */ @@ -13,6 +14,9 @@ public class NotifyEvent extends MessagingEvent { private QOSType m_qos; private byte[] m_message; private boolean m_retaned; + + //Optional attribute, available only fo QoS 1 and 2 + int m_msgID; public NotifyEvent(String clientId, String topic, QOSType qos, byte[] message, boolean retained) { @@ -23,6 +27,11 @@ public class NotifyEvent extends MessagingEvent { m_retaned = retained; } + public NotifyEvent(String clientId, String topic, QOSType qos, byte[] message, boolean retained, int msgID) { + this(clientId, topic, qos, message, retained); + m_msgID = msgID; + } + public String getClientId() { return m_clientId; } @@ -42,5 +51,9 @@ public class NotifyEvent extends MessagingEvent { public boolean isRetained() { return m_retaned; } + + public int getMessageID() { + return m_msgID; + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java new file mode 100644 index 00000000..2197775d --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java @@ -0,0 +1,24 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * Used to send the ack message back to the client after a publish + */ +public class PubAckEvent extends MessagingEvent { + + int m_messageId; + + String m_clientID; + + public PubAckEvent(int messageID, String clientID) { + m_messageId = messageID ; + m_clientID = clientID; + } + + public int getMessageId() { + return m_messageId; + } + + public String getClientID() { + return m_clientID; + } +} 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 86a9954b..6dbf1e5a 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 @@ -2,18 +2,20 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import java.io.Serializable; + /** * * @author andrea */ -public class PublishEvent extends MessagingEvent { +public class PublishEvent extends MessagingEvent implements Serializable { String m_topic; QOSType m_qos; byte[] m_message; boolean m_retain; String m_clientID; //Optional attribute, available only fo QoS 1 and 2 - String m_msgID; + int m_msgID; public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, String clientID) { @@ -25,7 +27,7 @@ public class PublishEvent extends MessagingEvent { } public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, - String clientID, String msgID) { + String clientID, int msgID) { this(topic, qos, message, retain, clientID); m_msgID = msgID; } @@ -50,7 +52,7 @@ public class PublishEvent extends MessagingEvent { return m_clientID; } - public String getMessageID() { + public int 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 8ae838ba..d27cdb59 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 @@ -9,18 +9,12 @@ import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.INotifier; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; +import org.dna.mqtt.moquette.proto.messages.*; + import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; -import org.dna.mqtt.moquette.proto.messages.DisconnectMessage; -import org.dna.mqtt.moquette.proto.messages.PingRespMessage; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; -import org.dna.mqtt.moquette.proto.messages.SubAckMessage; -import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; -import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; -import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -179,14 +173,20 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { UnsubAckMessage ackMessage = new UnsubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); - LOG.info("replying with UnsubAct to MSG ID {0}", msg.getMessageID()); + LOG.info("replying with UnsubAck to MSG ID {0}", msg.getMessageID()); session.write(ackMessage); } 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(), clientID); + + if (message.getQos() == QOSType.MOST_ONE) { + m_messaging.publish(message.getTopicName(), message.getPayload(), + message.getQos(), message.isRetainFlag(), clientID); + } else { + m_messaging.publish(message.getTopicName(), message.getPayload(), + message.getQos(), message.isRetainFlag(), clientID, message.getMessageID()); + } } protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) { @@ -222,12 +222,19 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { m_authenticator = authenticator; } - public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained) { + + + public void notify(NotifyEvent evt) { + LOG.debug("notify invoked with event " + evt); + String clientId = evt.getClientId(); PublishMessage pubMessage = new PublishMessage(); - pubMessage.setRetainFlag(retained); - pubMessage.setTopicName(topic); - pubMessage.setQos(qOSType); - pubMessage.setPayload(payload); + pubMessage.setRetainFlag(evt.isRetained()); + pubMessage.setTopicName(evt.getTopic()); + pubMessage.setQos(evt.getQos()); + pubMessage.setPayload(evt.getMessage()); + if (pubMessage.getQos() != QOSType.MOST_ONE) { + pubMessage.setMessageID(evt.getMessageID()); + } LOG.debug("notify invoked"); m_clientIDsLock.lock(); @@ -254,4 +261,26 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { } session.close(true); } + + public void sendPubAck(PubAckEvent evt) { + LOG.debug("sendPubAck invoked"); + + String clientId = evt.getClientID(); + + PubAckMessage pubAckMessage = new PubAckMessage(); + pubAckMessage.setMessageID(evt.getMessageId()); + + m_clientIDsLock.lock(); + try { + assert m_clientIDs != null; + LOG.debug("clientIDs are " + m_clientIDs); + assert m_clientIDs.get(clientId) != null; + LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); + m_clientIDs.get(clientId).getSession().write(pubAckMessage); + }catch(Throwable t) { + LOG.error(null, t); + } finally { + m_clientIDsLock.unlock(); + } + } } 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 850b58b2..ed331543 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 @@ -4,10 +4,7 @@ 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; +import org.dna.mqtt.moquette.messaging.spi.impl.events.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,12 +35,14 @@ class Notifier implements Runnable { LOG.debug("Notifing event " + evt); if (evt instanceof NotifyEvent) { NotifyEvent evtN = (NotifyEvent) evt; - m_notifier.notify(evtN.getClientId(), evtN.getTopic(), evtN.getQos(), evtN.getMessage(), evtN.isRetained()); + m_notifier.notify(evtN); } else if (evt instanceof DisconnectEvent) { DisconnectEvent evtD = (DisconnectEvent) evt; m_notifier.disconnect(evtD.getSession()); } else if (evt instanceof CleanInFlightEvent) { m_messaging.refill(evt); + } else if (evt instanceof PubAckEvent) { + m_notifier.sendPubAck((PubAckEvent) 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 e1ebefc7..ca9cb9e8 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 @@ -15,23 +15,8 @@ import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder; import org.apache.mina.transport.socket.nio.NioSocketAcceptor; import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; -import org.dna.mqtt.moquette.proto.ConnAckEncoder; -import org.dna.mqtt.moquette.proto.ConnectDecoder; -import org.dna.mqtt.moquette.proto.DisconnectDecoder; -import org.dna.mqtt.moquette.proto.MQTTLoggingFilter; -import org.dna.mqtt.moquette.proto.PingReqDecoder; -import org.dna.mqtt.moquette.proto.PingRespEncoder; -import org.dna.mqtt.moquette.proto.PublishDecoder; -import org.dna.mqtt.moquette.proto.PublishEncoder; -import org.dna.mqtt.moquette.proto.SubAckEncoder; -import org.dna.mqtt.moquette.proto.SubscribeDecoder; -import org.dna.mqtt.moquette.proto.UnsubAckEncoder; -import org.dna.mqtt.moquette.proto.UnsubscribeDecoder; -import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; -import org.dna.mqtt.moquette.proto.messages.PingRespMessage; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; -import org.dna.mqtt.moquette.proto.messages.SubAckMessage; -import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; +import org.dna.mqtt.moquette.proto.*; +import org.dna.mqtt.moquette.proto.messages.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** @@ -73,6 +58,7 @@ public class Server { encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder()); encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder()); + encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder()); encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder()); 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 01b952cd..28aaa6df 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 @@ -6,6 +6,7 @@ import java.util.concurrent.CountDownLatch; 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.junit.After; import static org.junit.Assert.*; import org.junit.Before; @@ -313,5 +314,29 @@ public class ServerIntegrationTest { client.close(); client.shutdown(); } + + /*-------------- QoS 1 tests ----*/ + @Test + public void testPublishWithQoS1() throws IOException, InterruptedException { +// startServer(); + Client client = new Client("localhost", Server.PORT); +// Client client = new Client("test.mosquitto.org", 1883); + client.connect(); + + + client.subscribe("/topic", new IPublishCallback() { + + public void published(String topic, byte[] message) { + received = true; + } + }); + + client.publish("/topic", "Test my payload".getBytes(), AbstractMessage.QOSType.LEAST_ONE, false); + + client.close(); + client.shutdown(); + + assertTrue(received); + } } From 04872e07f748d8a6f1064f65a525e64444d26e93 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 10:10:48 +0200 Subject: [PATCH 07/13] Extracted MQTT session constants into separate class --- .../dna/mqtt/moquette/server/Constants.java | 9 +++++++ .../dna/mqtt/moquette/server/MQTTHandler.java | 25 +++++++++---------- 2 files changed, 21 insertions(+), 13 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/server/Constants.java 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 new file mode 100644 index 00000000..1fa8809b --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Constants.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.moquette.server; + +/** + * Server constants keeper + */ +public class Constants { + protected 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 d27cdb59..727f8672 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 @@ -24,7 +24,6 @@ import org.slf4j.LoggerFactory; */ public class MQTTHandler extends IoHandlerAdapter implements INotifier { - protected static final String ATTR_CLIENTID = "ClientID"; private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class); /** * Maps CLIENT_ID to the IoSession that represents the connection @@ -87,7 +86,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { if (m_clientIDs.containsKey(msg.getClientID())) { //clean the subscriptions if the old used a cleanSession = true IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession(); - boolean cleanSession = (Boolean) oldSession.getAttribute("cleanSession"); + boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION); if (cleanSession) { //cleanup topic subscriptions m_messaging.removeSubscriptions(msg.getClientID()); @@ -104,9 +103,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { int keepAlive = msg.getKeepAlive(); session.setAttribute("keepAlive", keepAlive); - session.setAttribute("cleanSession", msg.isCleanSession()); + session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); //used to track the client in the subscription and publishing phases. - session.setAttribute(ATTR_CLIENTID, msg.getClientID()); + session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID()); session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f)); @@ -114,7 +113,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(), msg.getClientID()); + willQos, msg.isWillRetain(), msg.getClientID(), session); } //handle user authentication @@ -148,7 +147,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { protected void handleSubscribe(IoSession session, SubscribeMessage msg) { LOG.debug("handleSubscribe, registering the subscriptions"); for (SubscribeMessage.Couple req : msg.subscriptions()) { - m_messaging.subscribe((String) session.getAttribute(ATTR_CLIENTID), + m_messaging.subscribe((String) session.getAttribute(Constants.ATTR_CLIENTID), req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()]); } @@ -167,7 +166,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) { LOG.info("unregistering the subscriptions"); for (String topic : msg.topics()) { - m_messaging.unsubscribe(topic, (String) session.getAttribute(ATTR_CLIENTID)); + m_messaging.unsubscribe(topic, (String) session.getAttribute(Constants.ATTR_CLIENTID)); } //ack the client UnsubAckMessage ackMessage = new UnsubAckMessage(); @@ -178,19 +177,19 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { } protected void handlePublish(IoSession session, PublishMessage message) { - String clientID = (String) session.getAttribute(ATTR_CLIENTID); + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); if (message.getQos() == QOSType.MOST_ONE) { m_messaging.publish(message.getTopicName(), message.getPayload(), - message.getQos(), message.isRetainFlag(), clientID); + message.getQos(), message.isRetainFlag(), clientID, session); } else { m_messaging.publish(message.getTopicName(), message.getPayload(), - message.getQos(), message.isRetainFlag(), clientID, message.getMessageID()); + message.getQos(), message.isRetainFlag(), clientID, message.getMessageID(), session); } } protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) { - String clientID = (String) session.getAttribute(ATTR_CLIENTID); + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); //remove from clientIDs // m_clientIDsLock.lock(); // try { @@ -198,7 +197,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { // } finally { // m_clientIDsLock.unlock(); // } - boolean cleanSession = (Boolean) session.getAttribute("cleanSession"); + boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); if (cleanSession) { //cleanup topic subscriptions m_messaging.removeSubscriptions(clientID); @@ -252,7 +251,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { } public void disconnect(IoSession session) { - String clientID = (String) session.getAttribute(ATTR_CLIENTID); + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); m_clientIDsLock.lock(); try { m_clientIDs.remove(clientID); From 3be2ccca8f18aea895492d169d6c14c9a1b21e9b Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 10:22:53 +0200 Subject: [PATCH 08/13] Added IoSession attribute to PublishEvent to avoid looking up --- .../mqtt/moquette/messaging/spi/IMessaging.java | 4 ++-- .../messaging/spi/impl/SimpleMessaging.java | 12 ++++++++---- .../messaging/spi/impl/events/PublishEvent.java | 14 +++++++++++--- .../messaging/spi/impl/SimpleMessagingTest.java | 2 +- .../dna/mqtt/moquette/server/MQTTHandlerTest.java | 4 ++-- 5 files changed, 24 insertions(+), 12 deletions(-) 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 cacf9da1..733f0615 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 @@ -18,12 +18,12 @@ 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, String clientID); + void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session); /** * Used to publish messages with QoS > 0 * */ - void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID); + void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session); /** * Subscribe a client to a specified topic with a defined level 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 05c27b8c..7e97b652 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 @@ -14,6 +14,7 @@ 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.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.server.Constants; import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; import org.fusesource.hawtdb.api.BTreeIndexFactory; @@ -108,20 +109,20 @@ public class SimpleMessaging implements IMessaging, Runnable { } - public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID) { + 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)); + m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, session)); } catch (InterruptedException ex) { LOG.error(null, ex); } } - public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID) { + public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session) { try { - m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID)); + m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID, session)); } catch (InterruptedException ex) { LOG.error(null, ex); } @@ -267,8 +268,11 @@ public class SimpleMessaging implements IMessaging, Runnable { for (final Subscription sub : subscriptions.matches(topic)) { if (qos == QOSType.MOST_ONE) { + //QoS 0 m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); } else { + //QoS 1 or 2 + //if the target subscription is not clean session and is not connected => store it m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID())); } } 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 6dbf1e5a..b30eb2ad 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 @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; +import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import java.io.Serializable; @@ -16,19 +17,22 @@ public class PublishEvent extends MessagingEvent implements Serializable { String m_clientID; //Optional attribute, available only fo QoS 1 and 2 int m_msgID; + + transient IoSession m_session; public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, - String clientID) { + String clientID, IoSession session) { m_topic = topic; m_qos = qos; m_message = message; m_retain = retain; m_clientID = clientID; + m_session = session; } public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, - String clientID, int msgID) { - this(topic, qos, message, retain, clientID); + String clientID, int msgID, IoSession session) { + this(topic, qos, message, retain, clientID, session); m_msgID = msgID; } @@ -55,4 +59,8 @@ public class PublishEvent extends MessagingEvent implements Serializable { public int getMessageID() { return m_msgID; } + + public IoSession getSession() { + return m_session; + } } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index 811ce739..38083ba6 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -57,7 +57,7 @@ public class SimpleMessagingTest { // messaging.setNotifier(notifier); //Exercise - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI"); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); messaging.processPublish(pubEvt); //Verify diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java index 22c79781..2c143703 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -86,7 +86,7 @@ public class MQTTHandlerTest { //Verify assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), - any(AbstractMessage.QOSType.class), anyBoolean(), eq("123")); + any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class)); } @@ -99,7 +99,7 @@ public class MQTTHandlerTest { (byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName)); IMessaging mockedMessaging = mock(IMessaging.class); - m_session.setAttribute(MQTTHandler.ATTR_CLIENTID, "fakeID"); + m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID"); //Exercise From ba7f5284cf79200ec1056b1d65bbe8944aa6fab6 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 10:57:59 +0200 Subject: [PATCH 09/13] Added cleanSession flag to Subscription class --- .../moquette/messaging/spi/IMessaging.java | 2 +- .../messaging/spi/impl/SimpleMessaging.java | 4 +-- .../messaging/spi/impl/Subscription.java | 8 ++++- .../dna/mqtt/moquette/server/MQTTHandler.java | 6 ++-- .../spi/impl/SimpleMessagingTest.java | 10 +++--- .../spi/impl/SubscriptionsStoreTest.java | 36 +++++++++---------- .../mqtt/moquette/server/MQTTHandlerTest.java | 4 +-- 7 files changed, 39 insertions(+), 31 deletions(-) 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 733f0615..4f86fb44 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 @@ -28,7 +28,7 @@ public interface IMessaging { /** * Subscribe a client to a specified topic with a defined level */ - void subscribe(String clientId, String topic, QOSType qos); + void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession); /** * Remove all subscription to any topic the client (identified by 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 7e97b652..b0a4dde3 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 @@ -128,8 +128,8 @@ public class SimpleMessaging implements IMessaging, Runnable { } } - public void subscribe(String clientId, String topic, QOSType qos) { - Subscription newSubscription = new Subscription(clientId, topic, qos); + public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession) { + Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession); try { LOG.debug("subscribe invoked for topic: " + topic); m_inboundQueue.put(new SubscribeEvent(newSubscription)); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java index 83fab74e..59f431ac 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java @@ -15,11 +15,13 @@ public class Subscription implements Serializable { QOSType requestedQos; String clientId; String topic; + boolean cleanSession; - public Subscription(String clientId, String topic, QOSType requestedQos) { + public Subscription(String clientId, String topic, QOSType requestedQos, boolean cleanSession) { this.requestedQos = requestedQos; this.clientId = clientId; this.topic = topic; + this.cleanSession = cleanSession; } public String getClientId() { @@ -34,6 +36,10 @@ public class Subscription implements Serializable { return topic; } + public boolean isCleanSession() { + return this.cleanSession; + } + @Override public boolean equals(Object obj) { if (obj == null) { 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 727f8672..cd2f5126 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 @@ -147,8 +147,10 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { protected void handleSubscribe(IoSession session, SubscribeMessage msg) { LOG.debug("handleSubscribe, registering the subscriptions"); for (SubscribeMessage.Couple req : msg.subscriptions()) { - m_messaging.subscribe((String) session.getAttribute(Constants.ATTR_CLIENTID), - req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()]); + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); + m_messaging.subscribe(clientID, req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()], + cleanSession); } //ack the client diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index 38083ba6..404aebf7 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -28,30 +28,30 @@ public class SimpleMessagingTest { @Test public void testSubscribe() { //Exercise - SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE)); + SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false)); messaging.processSubscribe(evt); //Verify - Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false); assertTrue(messaging.getSubscriptions().contains(expectedSubscription)); } @Test public void testDoubleSubscribe() { - SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE)); + SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false)); messaging.processSubscribe(evt); //Exercise messaging.processSubscribe(evt); //Verify - Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE); + Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false); assertEquals(1, messaging.getSubscriptions().size()); } @Test public void testPublish() throws InterruptedException { - SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE)); + SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false)); messaging.processSubscribe(evt); // INotifier notifier = mock(INotifier.class); // messaging.setNotifier(notifier); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java index e978cee9..01a02960 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java @@ -105,11 +105,11 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimple() { - Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE); + Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE, false); store.add(slashSub); assertTrue(store.matches("finance").isEmpty()); - Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE); + Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE, false); store.add(slashFinanceSub); assertTrue(store.matches("finance").isEmpty()); @@ -119,19 +119,19 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti() { - Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false); store.add(anySub); assertTrue(store.matches("finance").contains(anySub)); - Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); store.add(financeAnySub); assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub))); } @Test public void testMatchingDeepMulti_one_layer() { - Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE); - Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); store.add(anySub); store.add(financeAnySub); @@ -143,7 +143,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchingDeepMulti_two_layer() { - Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE, false); store.add(financeAnySub); //Verify @@ -152,18 +152,18 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleSingle() { - Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false); store.add(anySub); assertTrue(store.matches("finance").contains(anySub)); - Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE); + Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE, false); store.add(financeOne); assertTrue(store.matches("finance/stock").contains(financeOne)); } @Test public void testMatchManySingle() { - Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE); + Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE, false); store.add(manySub); //verify @@ -173,9 +173,9 @@ public class SubscriptionsStoreTest { @Test public void testMatchSlashSingle() { - Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE); + Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE, false); store.add(slashPlusSub); - Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false); store.add(anySub); //Verify @@ -187,10 +187,10 @@ public class SubscriptionsStoreTest { @Test public void testMatchManyDeepSingle() { - Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE); + Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE, false); store.add(slashPlusSub); - Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_1", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE); + Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_1", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE, false); store.add(slashPlusDeepSub); //Verify @@ -199,7 +199,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_allTheTree() { - store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE)); + store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false)); assertFalse(store.matches("finance").isEmpty()); assertFalse(store.matches("finance/ibm").isEmpty()); } @@ -207,7 +207,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_zeroLevel() { //check MULTI in case of zero level match - store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE)); + store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false)); assertFalse(store.matches("finance").isEmpty()); } @@ -215,7 +215,7 @@ public class SubscriptionsStoreTest { @Test public void testRemoveClientSubscriptions_existingClientID() { String cliendID = "FAKE_CLID_1"; - store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE)); + store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false)); //Exercise store.removeForClient(cliendID); @@ -227,7 +227,7 @@ public class SubscriptionsStoreTest { @Test public void testRemoveClientSubscriptions_notexistingClientID() { String cliendID = "FAKE_CLID_1"; - store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE)); + store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false)); //Exercise store.removeForClient("FAKE_CLID_2"); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java index 2c143703..dd227bf3 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -100,7 +100,7 @@ public class MQTTHandlerTest { IMessaging mockedMessaging = mock(IMessaging.class); m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID"); - + m_session.setAttribute(Constants.CLEAN_SESSION, false); //Exercise m_handler.setMessaging(mockedMessaging); @@ -108,7 +108,7 @@ public class MQTTHandlerTest { //Verify ArgumentCaptor argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class); - verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture()); + verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false)); assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue()); //TODO verify the message ACK and corresponding QoS!! } From 5644b8e048e955fbae08acedcb831569589da0f3 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 11:17:24 +0200 Subject: [PATCH 10/13] Minor change corrected typos in comments --- .../mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java index 72e5c25b..85be8916 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java @@ -192,7 +192,7 @@ public class SubscriptionsStore { } private TreeNode subscriptions = new TreeNode(null); private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); - //pesistent Map of clientID, list of Subscriptions + //persistent Map of clientID, list of Subscriptions private SortedIndex> m_persistent; /** @@ -207,7 +207,7 @@ public class SubscriptionsStore { m_persistent = (SortedIndex>) multiFactory.openOrCreate("subscriptions", indexFactory); - //reaload any subscriptions persisted + //reload any subscriptions persisted LOG.debug("Reloading all stored subscriptions..."); for (Map.Entry> entry : m_persistent) { for (Subscription subscription : entry.getValue()) { From 4ef628d0e66e86bc8ea862b9af14243957ca9427 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 11:41:57 +0200 Subject: [PATCH 11/13] Added active flag to Subscription --- .../mqtt/moquette/messaging/spi/impl/Subscription.java | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java index 59f431ac..b065985c 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java @@ -16,6 +16,7 @@ public class Subscription implements Serializable { String clientId; String topic; boolean cleanSession; + boolean active = true; public Subscription(String clientId, String topic, QOSType requestedQos, boolean cleanSession) { this.requestedQos = requestedQos; @@ -40,6 +41,14 @@ public class Subscription implements Serializable { return this.cleanSession; } + public boolean isActive() { + return active; + } + + public void setActive(boolean active) { + this.active = active; + } + @Override public boolean equals(Object obj) { if (obj == null) { From 599afa76e623000f993319c74cdabcb988ff4152 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 11:44:07 +0200 Subject: [PATCH 12/13] Added the logic to de-activate subscriptions on client close --- .../messaging/spi/impl/SubscriptionsStore.java | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java index 85be8916..de0ca1ce 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java @@ -134,6 +134,19 @@ public class SubscriptionsStore { child.removeClientSubscriptions(clientID); } } + + void disconnect(String clientID) { + for (Subscription s : m_subscriptions) { + if (s.clientId.equals(clientID)) { + s.setActive(false); + } + } + + //go deep + for (TreeNode child : m_children) { + child.disconnect(clientID); + } + } } protected static class Token { @@ -293,6 +306,10 @@ public class SubscriptionsStore { m_persistent.remove(clientID); } + public void disconnect(String clientID) { + subscriptions.disconnect(clientID); + } + public List matches(String topic) { List tokens = new ArrayList(); try { From e5fbf5c655d0be7ed6a6e0f1c9c581c6ce8d9ed9 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 12:33:06 +0200 Subject: [PATCH 13/13] 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();