diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 3cad0278..379ebe53 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -1,11 +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.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; /** * Interface to the underling messaging system used to publish, subscribe. @@ -18,7 +14,5 @@ public interface IMessaging { void disconnect(IoSession session); - void republishStored(String clientID); - void handleProtocolMessage(IoSession session, AbstractMessage msg); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java deleted file mode 100644 index 5196ca2e..00000000 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java +++ /dev/null @@ -1,21 +0,0 @@ -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; - -/** - * - * @author andrea - */ -public interface INotifier { - - 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/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java new file mode 100644 index 00000000..e29157a1 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java @@ -0,0 +1,129 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import java.util.ArrayList; +import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; + +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + */ +public class MemoryStorageService implements IStorageService { + + private Map> m_persistentSubscriptions = new HashMap>(); + private Map m_retainedStore = new HashMap(); + private Map> m_persistentMessageStore = new HashMap>(); + private Map m_inflightStore = new HashMap(); + private Map m_qos2Store = new HashMap(); + + private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class); + + public void initStore() { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { + if (message.length == 0) { + //clean the message from topic + m_retainedStore.remove(topic); + } else { + //store the message to the topic + m_retainedStore.put(topic, new HawtDBStorageService.StoredMessage(message, qos, topic)); + } + } + + public Collection searchMatching(IMatchingCondition condition) { + LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size()); + + List results = new ArrayList(); + + for (Map.Entry entry : m_retainedStore.entrySet()) { + HawtDBStorageService.StoredMessage storedMsg = entry.getValue(); + if (condition.match(entry.getKey())) { + results.add(storedMsg); + } + } + + return results; + } + + public void storePublishForFuture(PublishEvent evt) { + LOG.debug("storePublishForFuture store evt " + evt); + 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); + } + + public List retrivePersistedPublishes(String clientID) { + return m_persistentMessageStore.get(clientID); + } + + public void cleanPersistedPublishes(String clientID) { + m_persistentMessageStore.remove(clientID); + } + + public void cleanInFlight(String msgID) { + m_inflightStore.remove(msgID); + } + + public void addInFlight(PublishEvent evt, String publishKey) { + m_inflightStore.put(publishKey, evt); + } + + public void addNewSubscription(Subscription newSubscription, String clientID) { + if (!m_persistentSubscriptions.containsKey(clientID)) { + m_persistentSubscriptions.put(clientID, new HashSet()); + } + + Set subs = m_persistentSubscriptions.get(clientID); + if (!subs.contains(newSubscription)) { + subs.add(newSubscription); + m_persistentSubscriptions.put(clientID, subs); + } + } + + public void removeAllSubscriptions(String clientID) { + m_persistentSubscriptions.remove(clientID); + } + + public List retrieveAllSubscriptions() { + List allSubscriptions = new ArrayList(); + for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + allSubscriptions.addAll(entry.getValue()); + } + return allSubscriptions; + } + + public void close() { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void persistQoS2Message(String publishKey, PublishEvent evt) { + LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt)); + m_qos2Store.put(publishKey, evt); + } + + public void removeQoS2Message(String publishKey) { + m_qos2Store.remove(publishKey); + } + + public PublishEvent retrieveQoS2Message(String publishKey) { + return m_qos2Store.get(publishKey); + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java new file mode 100644 index 00000000..9e5f63e3 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -0,0 +1,406 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import java.util.Collection; +import java.util.List; +import java.util.Map; +import org.apache.mina.core.session.IdleStatus; +import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; +import org.dna.mqtt.moquette.messaging.spi.IMessaging; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; +import org.dna.mqtt.moquette.proto.PubCompMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.dna.mqtt.moquette.proto.messages.PubAckMessage; +import org.dna.mqtt.moquette.proto.messages.PubRecMessage; +import org.dna.mqtt.moquette.proto.messages.PubRelMessage; +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.server.ConnectionDescriptor; +import org.dna.mqtt.moquette.server.Constants; +import org.dna.mqtt.moquette.server.IAuthenticator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Class responsible to handle the logic of MQTT protocol it's the director of + * the protocol execution. + * + * Used by the front facing class SimpleMessaging. + * + * @author andrea + */ +class ProtocolProcessor { + + private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); + + private Map m_clientIDs; + private SubscriptionsStore subscriptions; + private IStorageService m_storageService; + private IAuthenticator m_authenticator; + + ProtocolProcessor() { + + } + + void init(Map clientIDs, SubscriptionsStore subscriptions, + IStorageService storageService) { + m_clientIDs = clientIDs; + this.subscriptions = subscriptions; + m_storageService = storageService; + } + + void processConnect(IoSession session, ConnectMessage msg) { + if (msg.getProcotolVersion() != 0x03) { + ConnAckMessage badProto = new ConnAckMessage(); + badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); + session.write(badProto); + session.close(false); + return; + } + + if (msg.getClientID() == null || msg.getClientID().length() > 23) { + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); + session.write(okResp); + return; + } + + //if an old client with the same ID already exists close its session. + 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(Constants.CLEAN_SESSION); + if (cleanSession) { + //cleanup topic subscriptions + processRemoveAllSubscriptions(msg.getClientID()); + } + + m_clientIDs.get(msg.getClientID()).getSession().close(false); + } + + ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession()); + m_clientIDs.put(msg.getClientID(), connDescr); + + int keepAlive = msg.getKeepAlive(); + LOG.debug(String.format("Connect with keepAlive %d s", keepAlive)); + session.setAttribute("keepAlive", keepAlive); + session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); + //used to track the client in the subscription and publishing phases. + session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID()); + + session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f)); + + //Handle will flag + if (msg.isWillFlag()) { + AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()]; + PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(), + msg.isWillRetain(), msg.getClientID(), session); + processPublish(pubEvt); + } + + //handle user authentication + if (msg.isUserFlag()) { + String pwd = null; + if (msg.isPasswordFlag()) { + pwd = msg.getPassword(); + } + if (!m_authenticator.checkValid(msg.getUsername(), pwd)) { + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); + session.write(okResp); + return; + } + } + + subscriptions.activate(msg.getClientID()); + + //handle clean session flag + if (msg.isCleanSession()) { + //remove all prev subscriptions + //cleanup topic subscriptions + processRemoveAllSubscriptions(msg.getClientID()); + } + + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); + session.write(okResp); + + if (!msg.isCleanSession()) { + //force the republish of stored QoS1 and QoS2 + republishStored(msg.getClientID()); + } + } + + private void republishStored(String clientID) { + LOG.debug("republishStored invoked"); + List publishedEvents = m_storageService.retrivePersistedPublishes(clientID); + if (publishedEvents == null) { + LOG.debug("republishStored, no stored publish events"); + return; + } + + for (PublishEvent pubEvt : publishedEvents) { + sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), + pubEvt.getMessage(), false, pubEvt.getMessageID()); + } + } + + private void processRemoveAllSubscriptions(String clientID) { + LOG.debug("processRemoveAllSubscriptions invoked"); + subscriptions.removeForClient(clientID); + + //remove also the messages stored of type QoS1/2 + m_storageService.cleanPersistedPublishes(clientID); + } + + protected void processPublish(PublishEvent evt) { + LOG.debug("processPublish invoked with " + evt); + final String topic = evt.getTopic(); + final AbstractMessage.QOSType qos = evt.getQos(); + final byte[] message = evt.getMessage(); + boolean retain = evt.isRetain(); + + String publishKey = null; + if (qos == AbstractMessage.QOSType.LEAST_ONE) { + //store the temporary message + publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); + m_storageService.addInFlight(evt, publishKey); + } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { + publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); + //store the message in temp store + m_storageService.persistQoS2Message(publishKey, evt); + sendPubRec(evt.getClientID(), evt.getMessageID()); + } + + publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + + if (qos == AbstractMessage.QOSType.LEAST_ONE) { + if (publishKey == null) { + throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + evt); + } + m_storageService.cleanInFlight(publishKey); + sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); + } + + if (retain) { + m_storageService.storeRetained(topic, message, qos); + } + } + + /** + * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 + * */ + private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) { + for (final Subscription sub : subscriptions.matches(topic)) { + if (qos == AbstractMessage.QOSType.MOST_ONE) { + //QoS 0 + sendPublish(sub.getClientId(), topic, qos, message, false); + } else { + //QoS 1 or 2 + //if the target subscription is not clean session and is not connected => store it + if (!sub.isCleanSession() && !sub.isActive()) { + //clone the event with matching clientID + PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); + m_storageService.storePublishForFuture(newPublishEvt); + } else { + //if QoS 2 then store it in temp memory + if (qos ==AbstractMessage.QOSType.EXACTLY_ONCE) { + String publishKey = String.format("%s%d", sub.getClientId(), messageID); + PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); + m_storageService.addInFlight(newPublishEvt, publishKey); + } + //publish + sendPublish(sub.getClientId(), topic, qos, message, false); + } + } + } + } + + private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained) { + sendPublish(clientId, topic, qos, message, retained, 0); + } + + private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained, int messageID) { + LOG.debug("notify invoked with event "); + PublishMessage pubMessage = new PublishMessage(); + pubMessage.setRetainFlag(retained); + pubMessage.setTopicName(topic); + pubMessage.setQos(qos); + pubMessage.setPayload(message); + if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { + pubMessage.setMessageID(messageID); + } + + try { + if (m_clientIDs == null) { + throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); + } + LOG.debug("clientIDs are " + m_clientIDs); + if (m_clientIDs.get(clientId) == null) { + throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs)); + } + LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); + m_clientIDs.get(clientId).getSession().write(pubMessage); + }catch(Throwable t) { + LOG.error(null, t); + } + } + + private void sendPubRec(String clientID, int messageID) { + LOG.debug(String.format("sendPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + PubRecMessage pubRecMessage = new PubRecMessage(); + pubRecMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubRecMessage); + } + + private void sendPubAck(PubAckEvent evt) { + LOG.debug("sendPubAck invoked"); + + String clientId = evt.getClientID(); + + PubAckMessage pubAckMessage = new PubAckMessage(); + pubAckMessage.setMessageID(evt.getMessageId()); + + try { + if (m_clientIDs == null) { + throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); + } + LOG.debug("clientIDs are " + m_clientIDs); + if (m_clientIDs.get(clientId) == null) { + throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs)); + } + 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); + } + } + + /** + * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish + * to all interested subscribers. + * */ + void processPubRel(String clientID, int messageID) { + String publishKey = String.format("%s%d", clientID, messageID); + PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); + + final String topic = evt.getTopic(); + final AbstractMessage.QOSType qos = evt.getQos(); + final byte[] message = evt.getMessage(); + boolean retain = evt.isRetain(); + + publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + + m_storageService.removeQoS2Message(publishKey); + + if (retain) { + m_storageService.storeRetained(topic, message, qos); + } + + sendPubComp(clientID, messageID); + } + + private void sendPubComp(String clientID, int messageID) { + LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); + PubCompMessage pubCompMessage = new PubCompMessage(); + pubCompMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubCompMessage); + } + + void processPubRec(String clientID, int messageID) { + //once received a PUBREC reply with a PUBREL(messageID) + LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + PubRelMessage pubRelMessage = new PubRelMessage(); + pubRelMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubRelMessage); + } + + void processPubComp(String clientID, int messageID) { + //once received the PUBCOMP then remove the message from the temp memory + String publishKey = String.format("%s%d", clientID, messageID); + m_storageService.cleanInFlight(publishKey); + } + + void processDisconnect(IoSession session, String clientID, boolean cleanSession) throws InterruptedException { + if (cleanSession) { + //cleanup topic subscriptions + processRemoveAllSubscriptions(clientID); + } +// m_notifier.disconnect(evt.getSession()); + m_clientIDs.remove(clientID); + session.close(true); + + //de-activate the subscriptions for this ClientID +// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); + subscriptions.deactivate(clientID); + } + + /** + * Remove the clientID from topic subscription, if not previously subscribed, + * doesn't reply any error + */ + void processUnsubscribe(IoSession session, String clientID, List topics, int messageID) { + LOG.debug("processSubscribe invoked"); + + for (String topic : topics) { + subscriptions.removeSubscription(topic, clientID); + } + //ack the client + UnsubAckMessage ackMessage = new UnsubAckMessage(); + ackMessage.setMessageID(messageID); + + LOG.info("replying with UnsubAck to MSG ID {0}", messageID); + session.write(ackMessage); + } + + + void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) { + LOG.debug("processSubscribe invoked"); + + for (SubscribeMessage.Couple req : msg.subscriptions()) { + AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; + Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); + subscribeSingleTopic(newSubscription, req.getTopic()); + } + + //ack the client + SubAckMessage ackMessage = new SubAckMessage(); + ackMessage.setMessageID(msg.getMessageID()); + + //TODO by now it handles only QoS 0 messages + for (int i = 0; i < msg.subscriptions().size(); i++) { + ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); + } + LOG.info("replying with SubAct to MSG ID " + msg.getMessageID()); + session.write(ackMessage); + } + + private void subscribeSingleTopic(Subscription newSubscription, final String topic) { + subscriptions.add(newSubscription); + + //scans retained messages to be published to the new subscription + Collection messages = m_storageService.searchMatching(new IMatchingCondition() { + public boolean match(String key) { + return SubscriptionsStore.matchTopics(key, topic); + } + }); + + for (HawtDBStorageService.StoredMessage storedMsg : messages) { + //fire the as retained the message + LOG.debug("send publish message for topic " + topic); + sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); + } + } + +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 8c57f078..e00a772f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -4,27 +4,21 @@ import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; -import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; -import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; -import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; -import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.proto.PubCompMessage; import org.dna.mqtt.moquette.proto.messages.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; -import org.dna.mqtt.moquette.server.IAuthenticator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -44,19 +38,17 @@ public class SimpleMessaging implements IMessaging, EventHandler { private RingBuffer m_ringBuffer; -// private INotifier m_notifier; - private IStorageService m_storageService; Map m_clientIDs = new HashMap(); - private IAuthenticator m_authenticator; - private ExecutorService m_executor; BatchEventProcessor m_eventProcessor; private static SimpleMessaging INSTANCE; + private ProtocolProcessor m_processor = new ProtocolProcessor(); + private SimpleMessaging() { } @@ -81,11 +73,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { } -/* public void setNotifier(INotifier notifier) { - m_notifier= notifier; - } */ - - private void disruptorPublish(MessagingEvent msgEvent) { long sequence = m_ringBuffer.next(); ValueEvent event = m_ringBuffer.get(sequence); @@ -100,30 +87,10 @@ public class SimpleMessaging implements IMessaging, EventHandler { disruptorPublish(new DisconnectEvent(session)); } - //method used by hte Notifier to re-put an event on the inbound queue - private void refill(MessagingEvent evt) { - disruptorPublish(evt); - } - - public void republishStored(String clientID) { - //create the event to push - LOG.debug("republishStored invoked to publish soterd messages for clientID " + clientID); - disruptorPublish(new RepublishEvent(clientID)); - } - public void handleProtocolMessage(IoSession session, AbstractMessage msg) { disruptorPublish(new ProtocolEvent(session, msg)); } - - /** - * NOT SAFE Method, to be removed because used only in tests - */ - protected SubscriptionsStore getSubscriptions() { - return subscriptions; - } - - public void stop() { disruptorPublish(new StopEvent()); } @@ -132,20 +99,18 @@ public class SimpleMessaging implements IMessaging, EventHandler { MessagingEvent evt = t.getEvent(); LOG.debug("onEvent processing messaging event " + evt); if (evt instanceof PublishEvent) { - processPublish((PublishEvent) evt); + m_processor.processPublish((PublishEvent) evt); } else if (evt instanceof StopEvent) { processStop(); } else if (evt instanceof DisconnectEvent) { DisconnectEvent disEvt = (DisconnectEvent) evt; String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); - processDisconnect(disEvt.getSession(), clientID); - } else if (evt instanceof RepublishEvent) { - processRepublish((RepublishEvent) evt); + m_processor.processDisconnect(disEvt.getSession(), clientID, false); } else if (evt instanceof ProtocolEvent) { IoSession session = ((ProtocolEvent) evt).getSession(); AbstractMessage message = ((ProtocolEvent) evt).getMessage(); if (message instanceof ConnectMessage) { - processConnect(session, (ConnectMessage) message); + m_processor.processConnect(session, (ConnectMessage) message); } else if (message instanceof PublishMessage) { PublishMessage pubMsg = (PublishMessage) message; PublishEvent pubEvt; @@ -158,38 +123,34 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else { pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session); } - processPublish(pubEvt); + m_processor.processPublish(pubEvt); } else if (message instanceof DisconnectMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); - if (cleanSession) { - //cleanup topic subscriptions - processRemoveAllSubscriptions(clientID); - } //close the TCP connection //session.close(true); - processDisconnect(session, clientID); + m_processor.processDisconnect(session, clientID, cleanSession); } else if (message instanceof UnsubscribeMessage) { UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID()); + m_processor.processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID()); } else if (message instanceof SubscribeMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); - processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); + m_processor.processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); } else if (message instanceof PubRelMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = ((PubRelMessage) message).getMessageID(); - processPubRel(clientID, messageID); + m_processor.processPubRel(clientID, messageID); } else if (message instanceof PubRecMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = ((PubRecMessage) message).getMessageID(); - processPubRec(clientID, messageID); + m_processor.processPubRec(clientID, messageID); } else if (message instanceof PubCompMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = ((PubCompMessage) message).getMessageID(); - processPubComp(clientID, messageID); + m_processor.processPubComp(clientID, messageID); } else { throw new RuntimeException("Illegal message received " + message); } @@ -204,265 +165,10 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_storageService.initStore(); subscriptions.init(m_storageService); + m_processor.init(m_clientIDs, subscriptions, m_storageService); } - protected void processConnect(IoSession session, ConnectMessage msg) { - if (msg.getProcotolVersion() != 0x03) { - ConnAckMessage badProto = new ConnAckMessage(); - badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); - session.write(badProto); - session.close(false); - return; - } - - if (msg.getClientID() == null || msg.getClientID().length() > 23) { - ConnAckMessage okResp = new ConnAckMessage(); - okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); - session.write(okResp); - return; - } - - //if an old client with the same ID already exists close its session. - 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(Constants.CLEAN_SESSION); - if (cleanSession) { - //cleanup topic subscriptions - processRemoveAllSubscriptions(msg.getClientID()); - } - - m_clientIDs.get(msg.getClientID()).getSession().close(false); - } - - ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession()); - m_clientIDs.put(msg.getClientID(), connDescr); - - int keepAlive = msg.getKeepAlive(); - LOG.debug(String.format("Connect with keepAlive %d s", keepAlive)); - session.setAttribute("keepAlive", keepAlive); - session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); - //used to track the client in the subscription and publishing phases. - session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID()); - - session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f)); - - //Handle will flag - if (msg.isWillFlag()) { - QOSType willQos = QOSType.values()[msg.getWillQos()]; - PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(), - msg.isWillRetain(), msg.getClientID(), session); - processPublish(pubEvt); - } - - //handle user authentication - if (msg.isUserFlag()) { - String pwd = null; - if (msg.isPasswordFlag()) { - pwd = msg.getPassword(); - } - if (!m_authenticator.checkValid(msg.getUsername(), pwd)) { - ConnAckMessage okResp = new ConnAckMessage(); - okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); - session.write(okResp); - return; - } - } - - subscriptions.activate(msg.getClientID()); - - //handle clean session flag - if (msg.isCleanSession()) { - //remove all prev subscriptions - //cleanup topic subscriptions - processRemoveAllSubscriptions(msg.getClientID()); - } else { - //force the republish of stored QoS1 and QoS2 - republishStored(msg.getClientID()); - } - - ConnAckMessage okResp = new ConnAckMessage(); - okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); - session.write(okResp); - } - - /** - * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish - * to all interested subscribers. - * */ - protected void processPubRel(String clientID, int messageID) { - String publishKey = String.format("%s%d", clientID, messageID); - PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); - - final String topic = evt.getTopic(); - final QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); - boolean retain = evt.isRetain(); - - publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); - - m_storageService.removeQoS2Message(publishKey); - - if (retain) { - m_storageService.storeRetained(topic, message, qos); - } - - sendPubComp(clientID, messageID); - } - - protected void processPublish(PublishEvent evt) { - LOG.debug("processPublish invoked with " + evt); - final String topic = evt.getTopic(); - final QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); - boolean retain = evt.isRetain(); - - String publishKey = null; - if (qos == QOSType.LEAST_ONE) { - //store the temporary message - publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); - m_storageService.addInFlight(evt, publishKey); - } else if (qos == QOSType.EXACTLY_ONCE) { - publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); - //store the message in temp store - m_storageService.persistQoS2Message(publishKey, evt); - sendPubRec(evt.getClientID(), evt.getMessageID()); - } - - publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); - - if (qos == QOSType.LEAST_ONE) { - if (publishKey == null) { - throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + evt); - } - m_storageService.cleanInFlight(publishKey); - sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); - } - - if (retain) { - m_storageService.storeRetained(topic, message, qos); - } - } - - /** - * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 - * */ - private void publish2Subscribers(String topic, QOSType qos, byte[] message, boolean retain, Integer messageID) { - for (final Subscription sub : subscriptions.matches(topic)) { - if (qos == QOSType.MOST_ONE) { - //QoS 0 - notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); - } else { - //QoS 1 or 2 - //if the target subscription is not clean session and is not connected => store it - if (!sub.isCleanSession() && !sub.isActive()) { - //clone the event with matching clientID - PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); - m_storageService.storePublishForFuture(newPublishEvt); - } else { - //if QoS 2 then store it in temp memory - if (qos ==QOSType.EXACTLY_ONCE) { - String publishKey = String.format("%s%d", sub.getClientId(), messageID); - PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); - m_storageService.addInFlight(newPublishEvt, publishKey); - } - //publish - notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); - } - } - } - } - - private void processPubRec(String clientID, int messageID) { - //once received a PUBREC reply with a PUBREL(messageID) - LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); - PubRelMessage pubRelMessage = new PubRelMessage(); - pubRelMessage.setMessageID(messageID); - - m_clientIDs.get(clientID).getSession().write(pubRelMessage); - } - - private void processPubComp(String clientID, int messageID) { - //once received the PUBCOMP then remove the message from the temp memory - String publishKey = String.format("%s%d", clientID, messageID); - m_storageService.cleanInFlight(publishKey); - } - - private void subscribeSingleTopic(Subscription newSubscription, final String topic) { - subscriptions.add(newSubscription); - - //scans retained messages to be published to the new subscription - Collection messages = m_storageService.searchMatching(new IMatchingCondition() { - public boolean match(String key) { - return SubscriptionsStore.matchTopics(key, topic); - } - }); - - for (StoredMessage storedMsg : messages) { - //fire the as retained the message - LOG.debug("Inserting NotifyEvent into outbound for topic " + topic); - notify(new NotifyEvent(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true)); - } - } - - protected void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) { - LOG.debug("processSubscribe invoked"); - - for (SubscribeMessage.Couple req : msg.subscriptions()) { - QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; - Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); - subscribeSingleTopic(newSubscription, req.getTopic()); - } - - //ack the client - SubAckMessage ackMessage = new SubAckMessage(); - ackMessage.setMessageID(msg.getMessageID()); - - //TODO by now it handles only QoS 0 messages - for (int i = 0; i < msg.subscriptions().size(); i++) { - ackMessage.addType(QOSType.MOST_ONE); - } - LOG.info("replying with SubAct to MSG ID " + msg.getMessageID()); - session.write(ackMessage); - } - - /** - * Remove the clientID from topic subscription, if not previously subscribed, - * doesn't reply any error - */ - protected void processUnsubscribe(IoSession session, String clientID, List topics, int messageID) { - LOG.debug("processSubscribe invoked"); - - for (String topic : topics) { - subscriptions.removeSubscription(topic, clientID); - } - //ack the client - UnsubAckMessage ackMessage = new UnsubAckMessage(); - ackMessage.setMessageID(messageID); - - LOG.info("replying with UnsubAck to MSG ID {0}", messageID); - session.write(ackMessage); - } - - protected void processRemoveAllSubscriptions(String clientID) { - LOG.debug("processRemoveAllSubscriptions invoked"); - subscriptions.removeForClient(clientID); - - //remove also the messages stored of type QoS1/2 - m_storageService.cleanPersistedPublishes(clientID); - } - - private void processDisconnect(IoSession session, String clientID) throws InterruptedException { -// m_notifier.disconnect(evt.getSession()); - m_clientIDs.remove(clientID); - session.close(true); - - //de-activate the subscriptions for this ClientID -// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); - subscriptions.deactivate(clientID); - } - private void processStop() { LOG.debug("processStop invoked"); m_storageService.close(); @@ -470,84 +176,4 @@ public class SimpleMessaging implements IMessaging, EventHandler { // m_eventProcessor.halt(); m_executor.shutdown(); } - - private void processRepublish(RepublishEvent evt) throws InterruptedException { - LOG.debug("processRepublish invoked"); - List publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID()); - if (publishedEvents == null) { - LOG.debug("processRepublish, no stored publish events"); - return; - } - - for (PublishEvent pubEvt : publishedEvents) { - notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), - pubEvt.getMessage(), false, pubEvt.getMessageID())); - } - } - - private void notify(NotifyEvent evt) { - LOG.debug("notify invoked with event " + evt); - String clientId = evt.getClientId(); - PublishMessage pubMessage = new PublishMessage(); - 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()); - } - - try { - if (m_clientIDs == null) { - throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); - } - LOG.debug("clientIDs are " + m_clientIDs); - if (m_clientIDs.get(clientId) == null) { - throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs)); - } - LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); - m_clientIDs.get(clientId).getSession().write(pubMessage); - }catch(Throwable t) { - LOG.error(null, t); - } - } - - private void sendPubAck(PubAckEvent evt) { - LOG.debug("sendPubAck invoked"); - - String clientId = evt.getClientID(); - - PubAckMessage pubAckMessage = new PubAckMessage(); - pubAckMessage.setMessageID(evt.getMessageId()); - - try { - if (m_clientIDs == null) { - throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); - } - LOG.debug("clientIDs are " + m_clientIDs); - if (m_clientIDs.get(clientId) == null) { - throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs)); - } - 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); - } - } - - private void sendPubRec(String clientID, int messageID) { - LOG.debug(String.format("sendPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); - PubRecMessage pubRecMessage = new PubRecMessage(); - pubRecMessage.setMessageID(messageID); - - m_clientIDs.get(clientID).getSession().write(pubRecMessage); - } - - private void sendPubComp(String clientID, int messageID) { - LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); - PubCompMessage pubCompMessage = new PubCompMessage(); - pubCompMessage.setMessageID(messageID); - - m_clientIDs.get(clientID).getSession().write(pubCompMessage); - } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java deleted file mode 100644 index 06acfebc..00000000 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java +++ /dev/null @@ -1,69 +0,0 @@ -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 - */ -public class NotifyEvent extends MessagingEvent { - - private String m_clientId; - private String m_topic; - 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) { - m_clientId = clientId; - m_topic = topic; - m_qos = qos; - m_message = message; - 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; - } - - public String getTopic() { - return m_topic; - } - - public QOSType getQos() { - return m_qos; - } - - public byte[] getMessage() { - return m_message; - } - - public boolean isRetained() { - return m_retaned; - } - - public int getMessageID() { - return m_msgID; - } - - @Override - public String toString() { - return "NotifyEvent{" + - "m_retaned=" + m_retaned + - ", m_msgID=" + m_msgID + - ", m_qos=" + m_qos + - ", m_topic='" + m_topic + '\'' + - ", m_clientId='" + m_clientId + '\'' + - '}'; - } -} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java deleted file mode 100644 index e4849ce8..00000000 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java +++ /dev/null @@ -1,75 +0,0 @@ -package org.dna.mqtt.moquette.messaging.spi.impl; - -import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; -import org.dna.mqtt.moquette.messaging.spi.IStorageService; -import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; - -import java.util.Collection; -import java.util.Collections; -import java.util.List; - -/** - */ -public class DummyStorageService implements IStorageService { - public void initStore() { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public Collection searchMatching(IMatchingCondition condition) { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - public void storePublishForFuture(PublishEvent evt) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public List retrivePersistedPublishes(String clientID) { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - public void cleanPersistedPublishes(String clientID) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void cleanInFlight(String msgID) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void addInFlight(PublishEvent evt, String publishKey) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void addNewSubscription(Subscription newSubscription, String clientID) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void removeAllSubscriptions(String clientID) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public List retrieveAllSubscriptions() { - return Collections.EMPTY_LIST; - } - - public void close() { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void persistQoS2Message(String publishKey, PublishEvent evt) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void removeQoS2Message(String publishKey) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public PublishEvent retrieveQoS2Message(String publishKey) { - return null; //To change body of implemented methods use File | Settings | File Templates. - } -} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java similarity index 55% rename from broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java rename to broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index a7a9320d..393bbbc0 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -1,62 +1,52 @@ package org.dna.mqtt.moquette.messaging.spi.impl; -import java.io.File; +import java.util.Arrays; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.apache.mina.core.filterchain.IoFilter; import org.apache.mina.core.filterchain.IoFilterAdapter; import org.apache.mina.core.session.DummySession; import org.apache.mina.core.session.IoSession; import org.apache.mina.core.write.WriteRequest; -import org.dna.mqtt.moquette.messaging.spi.impl.events.*; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import static org.junit.Assert.*; - import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; 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.server.ConnectionDescriptor; -import org.dna.mqtt.moquette.server.Server; -import org.junit.After; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import org.junit.Before; import org.junit.Test; -import static org.junit.Assert.assertEquals; -import org.junit.BeforeClass; -import org.junit.Ignore; - /** * * @author andrea */ -public class SimpleMessagingTest { - +public class ProtocolProcessorTest { final static String FAKE_CLIENT_ID = "FAKE_123"; final static String FAKE_TOPIC = "/news"; - SimpleMessaging messaging; - - byte m_returnCode; + IoSession m_session; -// MQTTHandler m_handler; + byte m_returnCode; ConnectMessage connMsg; - + ProtocolProcessor m_processor; + + IStorageService m_storageService; + SubscriptionsStore subscriptions; AbstractMessage m_receivedMessage; - @BeforeClass - public static void beforeClass() { - File dbFile = new File(Server.STORAGE_FILE_PATH); - if (dbFile.exists()) { - dbFile.delete(); - } - } - + @Before public void setUp() throws InterruptedException { - messaging = SimpleMessaging.getInstance(); - messaging.init(); connMsg = new ConnectMessage(); connMsg.setProcotolVersion((byte) 0x03); @@ -65,7 +55,7 @@ public class SimpleMessagingTest { m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { @Override - public void filterWrite(NextFilter nextFilter, IoSession session, + public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception { try { m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); @@ -81,60 +71,84 @@ public class SimpleMessagingTest { //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); - } + + m_storageService = new MemoryStorageService(); + //m_storageService.initStore(); - @After - public void tearDown() { - m_receivedMessage = null; - messaging.stop(); - } - - @Test - public void testSubscribe() { - //Exercise - SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); - - //Verify - Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false); - assertTrue(messaging.getSubscriptions().contains(expectedSubscription)); + subscriptions = new SubscriptionsStore(); + subscriptions.init(m_storageService); + m_processor = new ProtocolProcessor(); + m_processor.init(new HashMap(), subscriptions, m_storageService); } @Test - public void testDoubleSubscribe() { -// SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0); - SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - - messaging.getSubscriptions().clearAllSubscriptions(); - assertEquals(0, messaging.getSubscriptions().size()); - - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); - + public void testHandleConnect_BadProtocol() { + connMsg.setProcotolVersion((byte) 0x02); + //Exercise - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + m_processor.processConnect(m_session, connMsg); //Verify - Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false); - assertEquals(1, messaging.getSubscriptions().size()); + assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode); + } + + @Test + public void testConnect_badClientID() { + connMsg.setClientID("extremely_long_clientID_greater_than_23"); + + //Exercise + m_processor.processConnect(m_session, connMsg); + + //Verify + assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode); } @Test - public void testPublish() throws InterruptedException { - //simulate a connect that register a clientID to an IoSession - ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true); - messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr); - -// SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0); - SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + public void testWill() { + connMsg.setClientID("123"); + connMsg.setWillFlag(true); + connMsg.setWillTopic("topic"); + connMsg.setWillMessage("Topic message"); + //Exercise - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); - messaging.processPublish(pubEvt); + //m_handler.setMessaging(mockedMessaging); + m_processor.processConnect(m_session, connMsg); + + //Verify + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); + //TODO verify the call + /*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), + any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/ + } + + @Test + public void testPublish() throws InterruptedException { + final Subscription subscription = new Subscription(FAKE_CLIENT_ID, + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + + //subscriptions.matches(topic) redefine the method to return true + SubscriptionsStore subs = new SubscriptionsStore() { + @Override + public List matches(String topic) { + if (topic.equals(FAKE_TOPIC)) { + return Arrays.asList(subscription); + } else { + throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); + } + } + }; + + //simulate a connect that register a clientID to an IoSession + Map connectionsMap = new HashMap(); + ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, subscription.isCleanSession()); + connectionsMap.put(FAKE_CLIENT_ID, connDescr); + m_processor.init(connectionsMap, subs, m_storageService); + + + //Exercise + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); + m_processor.processPublish(pubEvt); //Verify assertNotNull(m_receivedMessage); @@ -142,7 +156,40 @@ public class SimpleMessagingTest { } @Test - public void testPublishOfRetainedMessage() throws Exception { + public void testSubscribe() { + //Exercise + SubscribeMessage msg = new SubscribeMessage(); + msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Verify + assertTrue(m_receivedMessage instanceof SubAckMessage); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); + assertTrue(subscriptions.contains(expectedSubscription)); + } + + @Test + public void testDoubleSubscribe() { + SubscribeMessage msg = new SubscribeMessage(); + msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); + + subscriptions.clearAllSubscriptions(); + assertEquals(0, subscriptions.size()); + + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Exercise + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Verify + assertEquals(1, subscriptions.size()); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); + assertTrue(subscriptions.contains(expectedSubscription)); + } + + + @Test + public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { m_session.getFilterChain().remove("MessageCatcher"); m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { @@ -165,17 +212,35 @@ public class SimpleMessagingTest { } }); - //simulate a connect that register a clientID to an IoSession - ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true); - messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr); +// //simulate a connect that register a clientID to an IoSession + final Subscription subscription = new Subscription(FAKE_CLIENT_ID, + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + + //subscriptions.matches(topic) redefine the method to return true + SubscriptionsStore subs = new SubscriptionsStore() { + @Override + public List matches(String topic) { + if (topic.equals(FAKE_TOPIC)) { + return Arrays.asList(subscription); + } else { + throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); + } + } + }; + subs.init(new MemoryStorageService()); - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); - messaging.processPublish(pubEvt); + //simulate a connect that register a clientID to an IoSession + Map connectionsMap = new HashMap(); + ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, subscription.isCleanSession()); + connectionsMap.put(FAKE_CLIENT_ID, connDescr); + m_processor.init(connectionsMap, subs, m_storageService); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); + m_processor.processPublish(pubEvt); //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#")); - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); //Verify assertNotNull(m_receivedMessage); @@ -183,73 +248,4 @@ public class SimpleMessagingTest { PublishMessage pubMessage = (PublishMessage) m_receivedMessage; assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); } - - @Test - public void testHandleConnect_BadProtocol() { - connMsg.setProcotolVersion((byte) 0x02); - - //Exercise - messaging.processConnect(m_session, connMsg); - - //Verify - assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode); - } - - @Test - public void testConnect_badClientID() { - connMsg.setClientID("extremely_long_clientID_greater_than_23"); - - //Exercise - messaging.processConnect(m_session, connMsg); - - //Verify - assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode); - } - - @Test - public void testWill() { - connMsg.setClientID("123"); - connMsg.setWillFlag(true); - connMsg.setWillTopic("topic"); - connMsg.setWillMessage("Topic message"); - //IMessaging mockedMessaging = mock(IMessaging.class); - - //Exercise - //m_handler.setMessaging(mockedMessaging); - messaging.processConnect(m_session, connMsg); - - //Verify - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); - //TODO verify the call - /*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), - any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/ - } - - - @Ignore - public void testHandleSubscribe() { - /*String topicName = "/news"; - SubscribeMessage subscribeMsg = new SubscribeMessage(); - subscribeMsg.setMessageID(0x555); - subscribeMsg.addSubscription(new SubscribeMessage.Couple( - (byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName)); - //IMessaging mockedMessaging = mock(IMessaging.class); - - m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID"); - m_session.setAttribute(Constants.CLEAN_SESSION, false); - Subscription subscription = new Subscription("fakeID", "/news", QOSType.EXACTLY_ONCE, true);*/ - - //Exercise - //m_handler.setMessaging(mockedMessaging); -// SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0); - SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.EXACTLY_ONCE.ordinal(), "/news")); - messaging.processSubscribe(m_session, msg, "fakeID", false); - - //Verify - /*ArgumentCaptor argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class); - 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/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java index 6550d349..5cb4bab7 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -6,7 +6,7 @@ import java.text.ParseException; import java.util.Arrays; import java.util.List; -import org.dna.mqtt.moquette.messaging.spi.impl.DummyStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.MemoryStorageService; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.fusesource.hawtdb.api.MultiIndexFactory; import org.fusesource.hawtdb.api.PageFile; @@ -40,7 +40,7 @@ public class SubscriptionsStoreTest { pageFactory.open(); PageFile pageFile = pageFactory.getPageFile(); MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile); - store.init(new DummyStorageService()); + store.init(new MemoryStorageService()); } @After