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()); } 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..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 @@ -1,6 +1,8 @@ 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; /** @@ -16,12 +18,17 @@ 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, IoSession session); + + /** + * Used to publish messages with QoS > 0 + * */ + 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 */ - 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) @@ -38,4 +45,8 @@ public interface IMessaging { void unsubscribe(String topic, String clientID); 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/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 75d4c254..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; @@ -12,15 +13,9 @@ 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.Constants; import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; import org.fusesource.hawtdb.api.BTreeIndexFactory; @@ -66,6 +61,10 @@ 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; + //maps clientID to the list of pending messages stored + private SortedIndex> m_persistentMessageStore; public SimpleMessaging() { String storeFile = Server.STORAGE_FILE_PATH; @@ -86,6 +85,9 @@ 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(); } private void initRetainedMessageStore() { @@ -96,24 +98,47 @@ 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 in flight. + */ + 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(); } - public void publish(String topic, byte[] message, QOSType qos, boolean retain) { - //TODO for this moment no qos > 0 trivial implementation - QOSType defQos = QOSType.MOST_ONE; - + public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session) { try { - m_inboundQueue.put(new PublishEvent(topic, defQos, message, retain)); + m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, session)); } catch (InterruptedException ex) { LOG.error(null, ex); } } - public void subscribe(String clientId, String topic, QOSType qos) { - Subscription newSubscription = new Subscription(clientId, topic, qos); + 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, session)); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } + } + + 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)); @@ -139,8 +164,21 @@ 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); + } + + 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 */ @@ -198,8 +236,8 @@ public class SimpleMessaging implements IMessaging, Runnable { public BlockingQueue getNotifyEventQueue() { return m_outboundQueue; } - - + + protected void eventLoop() { LOG.debug("Started event loop"); boolean interrupted = false; @@ -217,7 +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(); @@ -225,21 +268,50 @@ 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 = String.format("%s%d", evt.getClientID(), evt.getMessageID()); + m_inflightStore.put(publishKey, evt); + cleanEvt = new CleanInFlightEvent(publishKey); + } for (final Subscription sub : subscriptions.matches(topic)) { - try { + if (qos == QOSType.MOST_ONE) { + //QoS 0 m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); - } catch (InterruptedException ex) { - LOG.error(null, ex); + } 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())); } } + + if (cleanEvt != null) { + m_outboundQueue.put(cleanEvt); + m_outboundQueue.put(new PubAckEvent(evt.getMessageID(), evt.getClientID())); + } + if (retain) { if (message.length == 0) { //clean the message from topic @@ -250,8 +322,8 @@ public class SimpleMessaging implements IMessaging, Runnable { } } } - - + + protected void processSubscribe(SubscribeEvent evt)/* throws InterruptedException*/ { LOG.debug("processSubscribe invoked"); Subscription newSubscription = evt.getSubscription(); @@ -284,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() { @@ -294,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/Subscription.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java index 83fab74e..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 @@ -15,11 +15,14 @@ public class Subscription implements Serializable { QOSType requestedQos; String clientId; String topic; + boolean cleanSession; + boolean active = true; - 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 +37,18 @@ public class Subscription implements Serializable { return topic; } + public boolean isCleanSession() { + 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) { 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..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 { @@ -192,7 +205,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 +220,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()) { @@ -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 { 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/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 355a56a3..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,22 +1,39 @@ 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; + /** * * @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 + int m_msgID; + + transient IoSession m_session; - public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain) { + public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, + 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, IoSession session) { + this(topic, qos, message, retain, clientID, session); + m_msgID = msgID; } public String getTopic() { @@ -34,4 +51,16 @@ public class PublishEvent extends MessagingEvent { public boolean isRetain() { return m_retain; } + + public String getClientID() { + return m_clientID; + } + + public int getMessageID() { + return m_msgID; + } + + public IoSession getSession() { + return m_session; + } } 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/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; 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..0a2ea7fd --- /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 { + 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 367cbef0..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 @@ -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; @@ -30,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 @@ -93,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()); @@ -110,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)); @@ -120,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()); + willQos, msg.isWillRetain(), msg.getClientID(), session); } //handle user authentication @@ -137,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(); @@ -154,8 +148,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(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 @@ -173,23 +169,30 @@ 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(); 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) { - m_messaging.publish(message.getTopicName(), message.getPayload(), - message.getQos(), message.isRetainFlag()); + 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, session); + } else { + m_messaging.publish(message.getTopicName(), message.getPayload(), + 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 { @@ -197,7 +200,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); @@ -221,12 +224,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(); @@ -244,7 +254,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); @@ -253,4 +263,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 676184b9..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 @@ -1,10 +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.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; @@ -18,10 +18,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() { @@ -33,10 +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 91b2a64d..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()); @@ -94,7 +80,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); 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..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,36 +28,36 @@ 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); //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", null); messaging.processPublish(pubEvt); //Verify 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/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)); } 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..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 @@ -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"), any(IoSession.class)); } @@ -99,8 +99,8 @@ 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"); + 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!! } 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); + } }