From 82e998e939142a17739dcf5e4e71447ab1f256b3 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 13:06:52 +0200 Subject: [PATCH 01/18] Created ProtocolProcessor and moved into it the processConnect from SimpleMessaging --- .../messaging/spi/impl/ProtocolProcessor.java | 264 ++++++++++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 224 +-------------- .../spi/impl/ProtocolProcessorTest.java | 108 +++++++ .../spi/impl/SimpleMessagingTest.java | 42 --- 4 files changed, 384 insertions(+), 254 deletions(-) create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java create mode 100644 broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java 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..84bb899d --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -0,0 +1,264 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +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.IMessaging; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; +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.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.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.PublishMessage; +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; + private IMessaging m_messaging; + + ProtocolProcessor() { + + } + + void init(Map clientIDs, SubscriptionsStore subscriptions, + IStorageService storageService, IMessaging messaging) { + m_clientIDs = clientIDs; + this.subscriptions = subscriptions; + m_storageService = storageService; + m_messaging = messaging; + } + + 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()); + } else { + //force the republish of stored QoS1 and QoS2 + m_messaging.republishStored(msg.getClientID()); + } + + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); + session.write(okResp); + } + + 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 + * */ + 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 + 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 ==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 + notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); + } + } + } + } + + 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() != AbstractMessage.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 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); + } + } + +} 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..031f987f 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 @@ -50,13 +50,13 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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() { } @@ -145,7 +145,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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; @@ -164,7 +164,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); if (cleanSession) { //cleanup topic subscriptions - processRemoveAllSubscriptions(clientID); + m_processor.processRemoveAllSubscriptions(clientID); } //close the TCP connection @@ -204,88 +204,13 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_storageService.initStore(); subscriptions.init(m_storageService); + m_processor.init(m_clientIDs, subscriptions, m_storageService, this); } - 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); - } +// protected void processConnect(IoSession session, ConnectMessage msg) { +// m_processor.processConnect(session, msg); +// } /** * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish @@ -300,7 +225,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { final byte[] message = evt.getMessage(); boolean retain = evt.isRetain(); - publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + m_processor.publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); m_storageService.removeQoS2Message(publishKey); @@ -312,66 +237,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } 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)); - } - } - } + m_processor.processPublish(evt); } private void processPubRec(String clientID, int messageID) { @@ -402,7 +268,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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)); + m_processor.notify(new NotifyEvent(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true)); } } @@ -445,14 +311,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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); @@ -480,69 +338,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { } for (PublishEvent pubEvt : publishedEvents) { - notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), + m_processor.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(); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java new file mode 100644 index 00000000..b0e1c14b --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -0,0 +1,108 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import java.util.HashMap; +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.IMessaging; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; +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.ConnAckMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.dna.mqtt.moquette.server.ConnectionDescriptor; +import static org.junit.Assert.assertEquals; +import org.junit.Before; +import org.junit.Test; +import static org.mockito.Mockito.*; + +/** + * + * @author andrea + */ +public class ProtocolProcessorTest { + IoSession m_session; + byte m_returnCode; + ConnectMessage connMsg; + ProtocolProcessor m_processor; + + @Before + public void setUp() throws InterruptedException { + connMsg = new ConnectMessage(); + connMsg.setProcotolVersion((byte) 0x03); + + m_session = new DummySession(); + + m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + + @Override + public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, + WriteRequest writeRequest) throws Exception { + try { + AbstractMessage receivedMessage = (AbstractMessage) writeRequest.getMessage(); + if (receivedMessage instanceof ConnAckMessage) { + ConnAckMessage buf = (ConnAckMessage) receivedMessage; + m_returnCode = buf.getReturnCode(); + } + } catch (Exception ex) { + throw new AssertionError("Wrong return code"); + } + } + }); + + //sleep to let the messaging batch processor to process the initEvent + Thread.sleep(300); + + IStorageService storageService = new HawtDBStorageService(); + storageService.initStore(); + + SubscriptionsStore subscriptions = new SubscriptionsStore(); + subscriptions.init(storageService); + m_processor = new ProtocolProcessor(); + IMessaging mockedMessaging = mock(IMessaging.class); + m_processor.init(new HashMap(), subscriptions, storageService, mockedMessaging); + } + + @Test + public void testHandleConnect_BadProtocol() { + connMsg.setProcotolVersion((byte) 0x02); + + //Exercise + m_processor.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 + m_processor.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"); + + + //Exercise + //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));*/ + } +} 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/SimpleMessagingTest.java index a7a9320d..29878297 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/SimpleMessagingTest.java @@ -184,48 +184,6 @@ public class SimpleMessagingTest { 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"; From 7d0cb705849d35d54118c24b5b4a34e8da96ea11 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 13:23:06 +0200 Subject: [PATCH 02/18] Moved processPubRel and processPubComp into ProtocolProcessor --- .../messaging/spi/impl/ProtocolProcessor.java | 33 ++++++++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 38 +------------------ 2 files changed, 34 insertions(+), 37 deletions(-) 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 index 84bb899d..8e02addc 100644 --- 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 @@ -10,6 +10,7 @@ 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; @@ -260,5 +261,37 @@ class ProtocolProcessor { 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. + * */ + 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 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); + } } 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 031f987f..90ce7bce 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 @@ -181,7 +181,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } 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(); @@ -208,34 +208,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { } -// protected void processConnect(IoSession session, ConnectMessage msg) { -// m_processor.processConnect(session, msg); -// } - - /** - * 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(); - - m_processor.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) { m_processor.processPublish(evt); } @@ -342,12 +314,4 @@ public class SimpleMessaging implements IMessaging, EventHandler { pubEvt.getMessage(), false, pubEvt.getMessageID())); } } - - 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); - } } From e74f6d565554b4a68172ac41aba44f9fe24ec925 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 13:24:53 +0200 Subject: [PATCH 03/18] Moved processPubRec into ProtocolProcessor --- .../messaging/spi/impl/ProtocolProcessor.java | 12 +++++++++++- .../moquette/messaging/spi/impl/SimpleMessaging.java | 11 +---------- 2 files changed, 12 insertions(+), 11 deletions(-) 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 index 8e02addc..56a65079 100644 --- 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 @@ -16,6 +16,7 @@ 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.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; @@ -266,7 +267,7 @@ class ProtocolProcessor { * 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) { + void processPubRel(String clientID, int messageID) { String publishKey = String.format("%s%d", clientID, messageID); PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); @@ -293,5 +294,14 @@ class ProtocolProcessor { 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); + } } 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 90ce7bce..a5c11aa9 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 @@ -185,7 +185,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } 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(); @@ -212,15 +212,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_processor.processPublish(evt); } - 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); From 353b70d6c717bde619917092b7f4aef7c4429ae4 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 13:28:55 +0200 Subject: [PATCH 04/18] Moved processPubComp into ProtocolProcessor --- .../messaging/spi/impl/ProtocolProcessor.java | 6 ++++++ .../messaging/spi/impl/SimpleMessaging.java | 13 +------------ 2 files changed, 7 insertions(+), 12 deletions(-) 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 index 56a65079..08a65c6a 100644 --- 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 @@ -303,5 +303,11 @@ class ProtocolProcessor { 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); + } } 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 a5c11aa9..d9262f4a 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 @@ -81,11 +81,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); @@ -189,7 +184,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } 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); } @@ -212,12 +207,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_processor.processPublish(evt); } - 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); From b3293a031f6ffda86a3e6870a7ccb5d88a23bcba Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 14:53:57 +0200 Subject: [PATCH 05/18] Moved processDisconnect into ProtocolProcessor --- .../messaging/spi/impl/ProtocolProcessor.java | 10 ++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 14 ++------------ 2 files changed, 12 insertions(+), 12 deletions(-) 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 index 08a65c6a..fafad0af 100644 --- 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 @@ -309,5 +309,15 @@ class ProtocolProcessor { String publishKey = String.format("%s%d", clientID, messageID); m_storageService.cleanInFlight(publishKey); } + + 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); + } } 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 d9262f4a..1922019d 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 @@ -133,7 +133,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else if (evt instanceof DisconnectEvent) { DisconnectEvent disEvt = (DisconnectEvent) evt; String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); - processDisconnect(disEvt.getSession(), clientID); + m_processor.processDisconnect(disEvt.getSession(), clientID); } else if (evt instanceof RepublishEvent) { processRepublish((RepublishEvent) evt); } else if (evt instanceof ProtocolEvent) { @@ -164,7 +164,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { //close the TCP connection //session.close(true); - processDisconnect(session, clientID); + m_processor.processDisconnect(session, clientID); } else if (message instanceof UnsubscribeMessage) { UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); @@ -263,16 +263,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { session.write(ackMessage); } - 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(); From 90e2277490becf312d805f44ad5838ae2e9b6f91 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 15:00:51 +0200 Subject: [PATCH 06/18] Moved processUnsubscribe into ProtocolProcessor --- .../messaging/spi/impl/ProtocolProcessor.java | 20 +++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 36 +++++++++---------- 2 files changed, 38 insertions(+), 18 deletions(-) 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 index fafad0af..4aef4ab2 100644 --- 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 @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import java.util.List; import java.util.Map; import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; @@ -18,6 +19,7 @@ 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.UnsubAckMessage; import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; import org.dna.mqtt.moquette.server.IAuthenticator; @@ -319,5 +321,23 @@ class ProtocolProcessor { // 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); + } } 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 1922019d..dd8a4cbb 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 @@ -168,7 +168,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } 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); @@ -245,23 +245,23 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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); - } +// /** +// * 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); +// } private void processStop() { LOG.debug("processStop invoked"); From dd4ca33ed00731ac0c6570eee101315da004750f Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 16:14:50 +0200 Subject: [PATCH 07/18] Moved processPublish from SimpleMessaging --- .../messaging/spi/impl/SimpleMessaging.java | 8 +- .../spi/impl/ProtocolProcessorTest.java | 109 ++++++++++++++- .../spi/impl/SimpleMessagingTest.java | 126 +++++++++--------- 3 files changed, 167 insertions(+), 76 deletions(-) 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 dd8a4cbb..4f57fe93 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 @@ -127,7 +127,7 @@ 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) { @@ -153,7 +153,7 @@ 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); @@ -203,10 +203,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { } - protected void processPublish(PublishEvent evt) { - m_processor.processPublish(evt); - } - private void subscribeSingleTopic(Subscription newSubscription, final String topic) { subscriptions.add(newSubscription); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index b0e1c14b..1f5cb5eb 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -1,6 +1,9 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +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; @@ -8,12 +11,19 @@ import org.apache.mina.core.session.IoSession; import org.apache.mina.core.write.WriteRequest; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessagingTest.FAKE_CLIENT_ID; +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.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.dna.mqtt.moquette.server.ConnectionDescriptor; 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.mockito.Mockito.*; @@ -23,11 +33,18 @@ import static org.mockito.Mockito.*; * @author andrea */ public class ProtocolProcessorTest { + + final static String FAKE_TOPIC = "/news"; + IoSession m_session; byte m_returnCode; ConnectMessage connMsg; ProtocolProcessor m_processor; + IStorageService m_storageService; + + AbstractMessage m_receivedMessage; + @Before public void setUp() throws InterruptedException { connMsg = new ConnectMessage(); @@ -41,9 +58,9 @@ public class ProtocolProcessorTest { public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception { try { - AbstractMessage receivedMessage = (AbstractMessage) writeRequest.getMessage(); - if (receivedMessage instanceof ConnAckMessage) { - ConnAckMessage buf = (ConnAckMessage) receivedMessage; + m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); + if (m_receivedMessage instanceof ConnAckMessage) { + ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; m_returnCode = buf.getReturnCode(); } } catch (Exception ex) { @@ -55,14 +72,14 @@ public class ProtocolProcessorTest { //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); - IStorageService storageService = new HawtDBStorageService(); - storageService.initStore(); + m_storageService = new HawtDBStorageService(); + m_storageService.initStore(); SubscriptionsStore subscriptions = new SubscriptionsStore(); - subscriptions.init(storageService); + subscriptions.init(m_storageService); m_processor = new ProtocolProcessor(); IMessaging mockedMessaging = mock(IMessaging.class); - m_processor.init(new HashMap(), subscriptions, storageService, mockedMessaging); + m_processor.init(new HashMap(), subscriptions, m_storageService, mockedMessaging); } @Test @@ -105,4 +122,82 @@ public class ProtocolProcessorTest { /*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 + IMessaging mockedMessaging = mock(IMessaging.class); + 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, mockedMessaging); + + + //Exercise + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); + m_processor.processPublish(pubEvt); + + //Verify + assertNotNull(m_receivedMessage); + //TODO check received message attributes + } + + + // @Test +// public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { +// m_session.getFilterChain().remove("MessageCatcher"); +// m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { +// +// @Override +// public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, +// WriteRequest writeRequest) throws Exception { +// try { +// System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); +// if (writeRequest.getMessage() instanceof PublishMessage) { +// m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); +// } +// +// if (m_receivedMessage instanceof ConnAckMessage) { +// ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; +// m_returnCode = buf.getReturnCode(); +// } +// } catch (Exception ex) { +// throw new AssertionError("Wrong return code"); +// } +// } +// }); +// +// //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); +// +// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); +// messaging.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); +// +// //Verify +// assertNotNull(m_receivedMessage); +// assertTrue(m_receivedMessage instanceof PublishMessage); +// PublishMessage pubMessage = (PublishMessage) m_receivedMessage; +// assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); +// } } 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/SimpleMessagingTest.java index 29878297..c9ac64ad 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/SimpleMessagingTest.java @@ -120,69 +120,69 @@ public class SimpleMessagingTest { assertEquals(1, messaging.getSubscriptions().size()); } - @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); - - //Exercise - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); - messaging.processPublish(pubEvt); - - //Verify - assertNotNull(m_receivedMessage); - //TODO check received message attributes - } - - @Test - public void testPublishOfRetainedMessage() throws Exception { - m_session.getFilterChain().remove("MessageCatcher"); - m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { - - @Override - public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, - WriteRequest writeRequest) throws Exception { - try { - System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); - if (writeRequest.getMessage() instanceof PublishMessage) { - m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); - } - - if (m_receivedMessage instanceof ConnAckMessage) { - ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; - m_returnCode = buf.getReturnCode(); - } - } catch (Exception ex) { - throw new AssertionError("Wrong return code"); - } - } - }); - - //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); - - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); - messaging.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); - - //Verify - assertNotNull(m_receivedMessage); - assertTrue(m_receivedMessage instanceof PublishMessage); - PublishMessage pubMessage = (PublishMessage) m_receivedMessage; - assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); - } +// @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); +// +// //Exercise +// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); +// messaging.processPublish(pubEvt); +// +// //Verify +// assertNotNull(m_receivedMessage); +// //TODO check received message attributes +// } +// +// @Test +// public void testPublishOfRetainedMessage() throws Exception { +// m_session.getFilterChain().remove("MessageCatcher"); +// m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { +// +// @Override +// public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, +// WriteRequest writeRequest) throws Exception { +// try { +// System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); +// if (writeRequest.getMessage() instanceof PublishMessage) { +// m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); +// } +// +// if (m_receivedMessage instanceof ConnAckMessage) { +// ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; +// m_returnCode = buf.getReturnCode(); +// } +// } catch (Exception ex) { +// throw new AssertionError("Wrong return code"); +// } +// } +// }); +// +// //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); +// +// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); +// messaging.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); +// +// //Verify +// assertNotNull(m_receivedMessage); +// assertTrue(m_receivedMessage instanceof PublishMessage); +// PublishMessage pubMessage = (PublishMessage) m_receivedMessage; +// assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); +// } @Ignore public void testHandleSubscribe() { From fb7c137a11effc0e754afaecc2755b4d2b049a63 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 17:06:15 +0200 Subject: [PATCH 08/18] Moved processSubscribe into ProtocolProcessor --- .../messaging/spi/impl/ProtocolProcessor.java | 43 ++++++ .../messaging/spi/impl/SimpleMessaging.java | 66 +------- .../spi/impl/ProtocolProcessorTest.java | 143 ++++++++++++------ .../spi/impl/SimpleMessagingTest.java | 135 ----------------- 4 files changed, 142 insertions(+), 245 deletions(-) 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 index 4aef4ab2..8c82a789 100644 --- 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 @@ -1,9 +1,11 @@ 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.NotifyEvent; @@ -19,6 +21,8 @@ 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; @@ -339,5 +343,44 @@ class ProtocolProcessor { 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("Inserting NotifyEvent into outbound for topic " + topic); + notify(new NotifyEvent(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 4f57fe93..9dca827f 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,8 +38,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { private RingBuffer m_ringBuffer; -// private INotifier m_notifier; - private IStorageService m_storageService; Map m_clientIDs = new HashMap(); @@ -172,7 +164,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } 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(); @@ -203,62 +195,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { } - 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); - m_processor.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); -// } - private void processStop() { LOG.debug("processStop invoked"); m_storageService.close(); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 1f5cb5eb..091fbd05 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -11,14 +11,15 @@ import org.apache.mina.core.session.IoSession; import org.apache.mina.core.write.WriteRequest; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; -import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessagingTest.FAKE_CLIENT_ID; 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 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 static org.junit.Assert.assertEquals; @@ -33,7 +34,7 @@ import static org.mockito.Mockito.*; * @author andrea */ public class ProtocolProcessorTest { - + final static String FAKE_CLIENT_ID = "FAKE_123"; final static String FAKE_TOPIC = "/news"; IoSession m_session; @@ -42,9 +43,10 @@ public class ProtocolProcessorTest { ProtocolProcessor m_processor; IStorageService m_storageService; - + SubscriptionsStore subscriptions; AbstractMessage m_receivedMessage; + @Before public void setUp() throws InterruptedException { connMsg = new ConnectMessage(); @@ -75,7 +77,7 @@ public class ProtocolProcessorTest { m_storageService = new HawtDBStorageService(); m_storageService.initStore(); - SubscriptionsStore subscriptions = new SubscriptionsStore(); + subscriptions = new SubscriptionsStore(); subscriptions.init(m_storageService); m_processor = new ProtocolProcessor(); IMessaging mockedMessaging = mock(IMessaging.class); @@ -156,48 +158,99 @@ public class ProtocolProcessorTest { assertNotNull(m_receivedMessage); //TODO check received message attributes } + + @Test + 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() { -// -// @Override -// public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, -// WriteRequest writeRequest) throws Exception { -// try { -// System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); -// if (writeRequest.getMessage() instanceof PublishMessage) { -// m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); -// } -// -// if (m_receivedMessage instanceof ConnAckMessage) { -// ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; -// m_returnCode = buf.getReturnCode(); -// } -// } catch (Exception ex) { -// throw new AssertionError("Wrong return code"); -// } -// } -// }); -// + @Test + public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { + m_session.getFilterChain().remove("MessageCatcher"); + m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + + @Override + public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, + WriteRequest writeRequest) throws Exception { + try { + System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); + if (writeRequest.getMessage() instanceof PublishMessage) { + m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); + } + + if (m_receivedMessage instanceof ConnAckMessage) { + ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; + m_returnCode = buf.getReturnCode(); + } + } catch (Exception ex) { + throw new AssertionError("Wrong return code"); + } + } + }); + // //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); -// -// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); -// messaging.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); -// -// //Verify -// assertNotNull(m_receivedMessage); -// assertTrue(m_receivedMessage instanceof PublishMessage); -// PublishMessage pubMessage = (PublishMessage) m_receivedMessage; -// assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); -// } + 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(m_storageService); + + //simulate a connect that register a clientID to an IoSession + IMessaging mockedMessaging = mock(IMessaging.class); + 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, mockedMessaging); + 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(), "#")); + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Verify + assertNotNull(m_receivedMessage); + assertTrue(m_receivedMessage instanceof PublishMessage); + PublishMessage pubMessage = (PublishMessage) m_receivedMessage; + assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); + } } 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/SimpleMessagingTest.java index c9ac64ad..d27fa915 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/SimpleMessagingTest.java @@ -1,32 +1,18 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import java.io.File; -import java.util.List; -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.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.SubscribeMessage; -import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Server; import org.junit.After; import org.junit.Before; -import org.junit.Test; - -import static org.junit.Assert.assertEquals; import org.junit.BeforeClass; -import org.junit.Ignore; /** * @@ -89,125 +75,4 @@ public class SimpleMessagingTest { 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)); - } - - @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); - - //Exercise - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); - - //Verify - Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false); - assertEquals(1, messaging.getSubscriptions().size()); - } - -// @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); -// -// //Exercise -// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); -// messaging.processPublish(pubEvt); -// -// //Verify -// assertNotNull(m_receivedMessage); -// //TODO check received message attributes -// } -// -// @Test -// public void testPublishOfRetainedMessage() throws Exception { -// m_session.getFilterChain().remove("MessageCatcher"); -// m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { -// -// @Override -// public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, -// WriteRequest writeRequest) throws Exception { -// try { -// System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); -// if (writeRequest.getMessage() instanceof PublishMessage) { -// m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); -// } -// -// if (m_receivedMessage instanceof ConnAckMessage) { -// ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; -// m_returnCode = buf.getReturnCode(); -// } -// } catch (Exception ex) { -// throw new AssertionError("Wrong return code"); -// } -// } -// }); -// -// //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); -// -// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); -// messaging.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); -// -// //Verify -// assertNotNull(m_receivedMessage); -// assertTrue(m_receivedMessage instanceof PublishMessage); -// PublishMessage pubMessage = (PublishMessage) m_receivedMessage; -// assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); -// } - - @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!! - } } From 22269ccfb1e9c946b41166afb891afaca1f1e055 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 17:06:48 +0200 Subject: [PATCH 09/18] Removed empty test case --- .../spi/impl/SimpleMessagingTest.java | 78 ------------------- 1 file changed, 78 deletions(-) delete mode 100644 broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java 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/SimpleMessagingTest.java deleted file mode 100644 index d27fa915..00000000 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ /dev/null @@ -1,78 +0,0 @@ -package org.dna.mqtt.moquette.messaging.spi.impl; - -import java.io.File; -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.proto.messages.AbstractMessage; - -import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; -import org.dna.mqtt.moquette.server.Server; -import org.junit.After; -import org.junit.Before; -import org.junit.BeforeClass; - -/** - * - * @author andrea - */ -public class SimpleMessagingTest { - - 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; - ConnectMessage connMsg; - - 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); - - m_session = new DummySession(); - - m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { - - @Override - public void filterWrite(NextFilter nextFilter, IoSession session, - WriteRequest writeRequest) throws Exception { - try { - m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); - if (m_receivedMessage instanceof ConnAckMessage) { - ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; - m_returnCode = buf.getReturnCode(); - } - } catch (Exception ex) { - throw new AssertionError("Wrong return code"); - } - } - }); - - //sleep to let the messaging batch processor to process the initEvent - Thread.sleep(300); - } - - @After - public void tearDown() { - m_receivedMessage = null; - messaging.stop(); - } - -} From d2554788eaaa83bbad7f0626f1080d278a8b1e77 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 17:40:56 +0200 Subject: [PATCH 10/18] Added memory implementation of DummyStorageService for the subscriptions part --- .../spi/impl/DummyStorageService.java | 26 ++++++++++++++++--- .../spi/impl/ProtocolProcessorTest.java | 2 +- 2 files changed, 24 insertions(+), 4 deletions(-) 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 index e4849ce8..a23208dd 100644 --- 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 @@ -1,5 +1,6 @@ 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; @@ -8,11 +9,18 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; /** */ public class DummyStorageService implements IStorageService { + + private Map> m_persistentSubscriptions = new HashMap>(); + public void initStore() { //To change body of implemented methods use File | Settings | File Templates. } @@ -46,15 +54,27 @@ public class DummyStorageService implements IStorageService { } public void addNewSubscription(Subscription newSubscription, String clientID) { - //To change body of implemented methods use File | Settings | File Templates. + 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) { - //To change body of implemented methods use File | Settings | File Templates. + m_persistentSubscriptions.remove(clientID); } public List retrieveAllSubscriptions() { - return Collections.EMPTY_LIST; + List allSubscriptions = new ArrayList(); + for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + allSubscriptions.addAll(entry.getValue()); + } + return allSubscriptions; } public void close() { diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 091fbd05..e4a35d71 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -231,7 +231,7 @@ public class ProtocolProcessorTest { } } }; - subs.init(m_storageService); + subs.init(new DummyStorageService()); //simulate a connect that register a clientID to an IoSession IMessaging mockedMessaging = mock(IMessaging.class); From 63333b11ad660445600c86ab3a74e5328516964f Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 18:08:06 +0200 Subject: [PATCH 11/18] Completed the memory implementation of DummyStorageService --- .../spi/impl/DummyStorageService.java | 55 +++++++++++++++---- .../spi/impl/ProtocolProcessorTest.java | 4 +- 2 files changed, 47 insertions(+), 12 deletions(-) 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 index a23208dd..cee54bbd 100644 --- 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 @@ -14,43 +14,77 @@ 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 DummyStorageService 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(DummyStorageService.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) { - //To change body of implemented methods use File | Settings | File Templates. + 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) { - return null; //To change body of implemented methods use File | Settings | File Templates. + 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) { - //To change body of implemented methods use File | Settings | File Templates. + 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 null; //To change body of implemented methods use File | Settings | File Templates. + return m_persistentMessageStore.get(clientID); } public void cleanPersistedPublishes(String clientID) { - //To change body of implemented methods use File | Settings | File Templates. + m_persistentMessageStore.remove(clientID); } public void cleanInFlight(String msgID) { - //To change body of implemented methods use File | Settings | File Templates. + m_inflightStore.remove(msgID); } public void addInFlight(PublishEvent evt, String publishKey) { - //To change body of implemented methods use File | Settings | File Templates. + m_inflightStore.put(publishKey, evt); } public void addNewSubscription(Subscription newSubscription, String clientID) { @@ -82,14 +116,15 @@ public class DummyStorageService implements IStorageService { } public void persistQoS2Message(String publishKey, PublishEvent evt) { - //To change body of implemented methods use File | Settings | File Templates. + LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt)); + m_qos2Store.put(publishKey, evt); } public void removeQoS2Message(String publishKey) { - //To change body of implemented methods use File | Settings | File Templates. + m_qos2Store.remove(publishKey); } public PublishEvent retrieveQoS2Message(String publishKey) { - return null; //To change body of implemented methods use File | Settings | File Templates. + return m_qos2Store.get(publishKey); } } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index e4a35d71..88e40325 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -74,8 +74,8 @@ public class ProtocolProcessorTest { //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); - m_storageService = new HawtDBStorageService(); - m_storageService.initStore(); + m_storageService = new DummyStorageService(); + //m_storageService.initStore(); subscriptions = new SubscriptionsStore(); subscriptions.init(m_storageService); From 635d37112ef4109b8f5e2af5e3ecab21d38d45a8 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 18:09:24 +0200 Subject: [PATCH 12/18] Renamed DummyStorageService to MemoryStorageService --- .../{DummyStorageService.java => MemoryStorageService.java} | 4 ++-- .../moquette/messaging/spi/impl/ProtocolProcessorTest.java | 4 ++-- .../spi/impl/subscriptions/SubscriptionsStoreTest.java | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) rename broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/{DummyStorageService.java => MemoryStorageService.java} (96%) 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/MemoryStorageService.java similarity index 96% rename from broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java rename to broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index cee54bbd..bba6b3d1 100644 --- 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/MemoryStorageService.java @@ -19,7 +19,7 @@ import org.slf4j.LoggerFactory; /** */ -public class DummyStorageService implements IStorageService { +public class MemoryStorageService implements IStorageService { private Map> m_persistentSubscriptions = new HashMap>(); private Map m_retainedStore = new HashMap(); @@ -27,7 +27,7 @@ public class DummyStorageService implements IStorageService { private Map m_inflightStore = new HashMap(); private Map m_qos2Store = new HashMap(); - private static final Logger LOG = LoggerFactory.getLogger(DummyStorageService.class); + private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class); public void initStore() { //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/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 88e40325..b3d5e5bf 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -74,7 +74,7 @@ public class ProtocolProcessorTest { //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); - m_storageService = new DummyStorageService(); + m_storageService = new MemoryStorageService(); //m_storageService.initStore(); subscriptions = new SubscriptionsStore(); @@ -231,7 +231,7 @@ public class ProtocolProcessorTest { } } }; - subs.init(new DummyStorageService()); + subs.init(new MemoryStorageService()); //simulate a connect that register a clientID to an IoSession IMessaging mockedMessaging = mock(IMessaging.class); 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 From e579180abf3f41a2401b4d5c52c02382c5870f95 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 18:10:12 +0200 Subject: [PATCH 13/18] Moved MemoryStorageService from test to src --- .../mqtt/moquette/messaging/spi/impl/MemoryStorageService.java | 1 + 1 file changed, 1 insertion(+) rename broker/src/{test => main}/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java (98%) diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java similarity index 98% rename from broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index bba6b3d1..29300c68 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java @@ -14,6 +14,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; From 2b7858de7c8c0cb58a8f497ebc0b99d03e818e2f Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 18:10:41 +0200 Subject: [PATCH 14/18] Minor change, cleaned imports --- .../mqtt/moquette/messaging/spi/impl/MemoryStorageService.java | 2 -- 1 file changed, 2 deletions(-) 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 index 29300c68..e29157a1 100644 --- 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 @@ -8,13 +8,11 @@ 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.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; -import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; From ff0139d809be10f0d7a30827276ee26c4249933f Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 18:55:11 +0200 Subject: [PATCH 15/18] Changed some methods visibility --- .../messaging/spi/impl/ProtocolProcessor.java | 10 +++++--- .../messaging/spi/impl/SimpleMessaging.java | 23 ++++--------------- 2 files changed, 12 insertions(+), 21 deletions(-) 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 index 8c82a789..8e075ab9 100644 --- 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 @@ -140,7 +140,7 @@ class ProtocolProcessor { session.write(okResp); } - void processRemoveAllSubscriptions(String clientID) { + private void processRemoveAllSubscriptions(String clientID) { LOG.debug("processRemoveAllSubscriptions invoked"); subscriptions.removeForClient(clientID); @@ -185,7 +185,7 @@ class ProtocolProcessor { /** * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 * */ - void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) { + 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 @@ -316,7 +316,11 @@ class ProtocolProcessor { m_storageService.cleanInFlight(publishKey); } - void processDisconnect(IoSession session, String clientID) throws InterruptedException { + 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); 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 9dca827f..07fe461f 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 @@ -88,9 +88,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { } //method used by hte Notifier to re-put an event on the inbound queue - private void refill(MessagingEvent evt) { - disruptorPublish(evt); - } +// private void refill(MessagingEvent evt) { +// disruptorPublish(evt); +// } public void republishStored(String clientID) { //create the event to push @@ -102,15 +102,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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()); } @@ -125,7 +116,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else if (evt instanceof DisconnectEvent) { DisconnectEvent disEvt = (DisconnectEvent) evt; String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); - m_processor.processDisconnect(disEvt.getSession(), clientID); + m_processor.processDisconnect(disEvt.getSession(), clientID, false); } else if (evt instanceof RepublishEvent) { processRepublish((RepublishEvent) evt); } else if (evt instanceof ProtocolEvent) { @@ -149,14 +140,10 @@ public class SimpleMessaging implements IMessaging, EventHandler { } 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 - m_processor.processRemoveAllSubscriptions(clientID); - } //close the TCP connection //session.close(true); - m_processor.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); From 2c11cade06323ca7b8ee1be6202e32290cb9ec5a Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 19:05:09 +0200 Subject: [PATCH 16/18] Removed a ping pong beteen the ProtocolProcessor and SimpleMessaging, avoid call method, send Event, receive Event e callback a method into ProcotolProcessor --- .../moquette/messaging/spi/IMessaging.java | 6 +--- .../messaging/spi/impl/ProtocolProcessor.java | 25 +++++++++++--- .../messaging/spi/impl/SimpleMessaging.java | 34 +++++++++---------- 3 files changed, 39 insertions(+), 26 deletions(-) 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..eeccae6b 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,7 @@ public interface IMessaging { void disconnect(IoSession session); - void republishStored(String clientID); +// void republishStored(String clientID); void handleProtocolMessage(IoSession session, AbstractMessage msg); } 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 index 8e075ab9..45a02907 100644 --- 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 @@ -11,6 +11,7 @@ import org.dna.mqtt.moquette.messaging.spi.IStorageService; 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.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.RepublishEvent; 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; @@ -130,14 +131,30 @@ class ProtocolProcessor { //remove all prev subscriptions //cleanup topic subscriptions processRemoveAllSubscriptions(msg.getClientID()); - } else { - //force the republish of stored QoS1 and QoS2 - m_messaging.republishStored(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) { + notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), + pubEvt.getMessage(), false, pubEvt.getMessageID())); + } } private void processRemoveAllSubscriptions(String clientID) { @@ -211,7 +228,7 @@ class ProtocolProcessor { } } - void notify(NotifyEvent evt) { + private void notify(NotifyEvent evt) { LOG.debug("notify invoked with event " + evt); String clientId = evt.getClientId(); PublishMessage pubMessage = new PublishMessage(); 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 07fe461f..1d106873 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 @@ -92,11 +92,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { // disruptorPublish(evt); // } - public void republishStored(String clientID) { + /*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)); @@ -117,9 +117,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { DisconnectEvent disEvt = (DisconnectEvent) evt; String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); m_processor.processDisconnect(disEvt.getSession(), clientID, false); - } else if (evt instanceof RepublishEvent) { + } /*else if (evt instanceof RepublishEvent) { processRepublish((RepublishEvent) evt); - } else if (evt instanceof ProtocolEvent) { + } */else if (evt instanceof ProtocolEvent) { IoSession session = ((ProtocolEvent) evt).getSession(); AbstractMessage message = ((ProtocolEvent) evt).getMessage(); if (message instanceof ConnectMessage) { @@ -190,17 +190,17 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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) { - m_processor.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), - pubEvt.getMessage(), false, pubEvt.getMessageID())); - } - } +// 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) { +// m_processor.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), +// pubEvt.getMessage(), false, pubEvt.getMessageID())); +// } +// } } From 0694f28f320c0730f734b88a1e194c9551b2be7e Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 19:21:11 +0200 Subject: [PATCH 17/18] Simplified APIs --- .../moquette/messaging/spi/IMessaging.java | 2 - .../moquette/messaging/spi/INotifier.java | 21 ------ .../messaging/spi/impl/ProtocolProcessor.java | 37 +++++----- .../messaging/spi/impl/SimpleMessaging.java | 31 +-------- .../spi/impl/events/NotifyEvent.java | 69 ------------------- .../spi/impl/ProtocolProcessorTest.java | 9 +-- 6 files changed, 23 insertions(+), 146 deletions(-) delete mode 100644 broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java delete mode 100644 broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java 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 eeccae6b..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 @@ -14,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/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 45a02907..9e5f63e3 100644 --- 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 @@ -8,10 +8,8 @@ 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.NotifyEvent; 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.events.RepublishEvent; 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; @@ -47,18 +45,16 @@ class ProtocolProcessor { private SubscriptionsStore subscriptions; private IStorageService m_storageService; private IAuthenticator m_authenticator; - private IMessaging m_messaging; ProtocolProcessor() { } void init(Map clientIDs, SubscriptionsStore subscriptions, - IStorageService storageService, IMessaging messaging) { + IStorageService storageService) { m_clientIDs = clientIDs; this.subscriptions = subscriptions; m_storageService = storageService; - m_messaging = messaging; } void processConnect(IoSession session, ConnectMessage msg) { @@ -152,8 +148,8 @@ class ProtocolProcessor { } for (PublishEvent pubEvt : publishedEvents) { - notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), - pubEvt.getMessage(), false, pubEvt.getMessageID())); + sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), + pubEvt.getMessage(), false, pubEvt.getMessageID()); } } @@ -206,7 +202,7 @@ class ProtocolProcessor { for (final Subscription sub : subscriptions.matches(topic)) { if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS 0 - notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); + 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 @@ -222,22 +218,25 @@ class ProtocolProcessor { m_storageService.addInFlight(newPublishEvt, publishKey); } //publish - notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); + sendPublish(sub.getClientId(), topic, qos, message, false); } } } } - private void notify(NotifyEvent evt) { - LOG.debug("notify invoked with event " + evt); - String clientId = evt.getClientId(); + 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(evt.isRetained()); - pubMessage.setTopicName(evt.getTopic()); - pubMessage.setQos(evt.getQos()); - pubMessage.setPayload(evt.getMessage()); + pubMessage.setRetainFlag(retained); + pubMessage.setTopicName(topic); + pubMessage.setQos(qos); + pubMessage.setPayload(message); if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { - pubMessage.setMessageID(evt.getMessageID()); + pubMessage.setMessageID(messageID); } try { @@ -399,8 +398,8 @@ class ProtocolProcessor { for (HawtDBStorageService.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)); + 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 1d106873..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 @@ -87,17 +87,6 @@ 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)); } @@ -117,9 +106,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { DisconnectEvent disEvt = (DisconnectEvent) evt; String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); m_processor.processDisconnect(disEvt.getSession(), clientID, false); - } /*else if (evt instanceof RepublishEvent) { - processRepublish((RepublishEvent) evt); - } */else if (evt instanceof ProtocolEvent) { + } else if (evt instanceof ProtocolEvent) { IoSession session = ((ProtocolEvent) evt).getSession(); AbstractMessage message = ((ProtocolEvent) evt).getMessage(); if (message instanceof ConnectMessage) { @@ -178,7 +165,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_storageService.initStore(); subscriptions.init(m_storageService); - m_processor.init(m_clientIDs, subscriptions, m_storageService, this); + m_processor.init(m_clientIDs, subscriptions, m_storageService); } @@ -189,18 +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) { -// m_processor.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), -// pubEvt.getMessage(), false, pubEvt.getMessageID())); -// } -// } } 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/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index b3d5e5bf..b4a3a327 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -80,8 +80,7 @@ public class ProtocolProcessorTest { subscriptions = new SubscriptionsStore(); subscriptions.init(m_storageService); m_processor = new ProtocolProcessor(); - IMessaging mockedMessaging = mock(IMessaging.class); - m_processor.init(new HashMap(), subscriptions, m_storageService, mockedMessaging); + m_processor.init(new HashMap(), subscriptions, m_storageService); } @Test @@ -143,11 +142,10 @@ public class ProtocolProcessorTest { }; //simulate a connect that register a clientID to an IoSession - IMessaging mockedMessaging = mock(IMessaging.class); 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, mockedMessaging); + m_processor.init(connectionsMap, subs, m_storageService); //Exercise @@ -234,11 +232,10 @@ public class ProtocolProcessorTest { subs.init(new MemoryStorageService()); //simulate a connect that register a clientID to an IoSession - IMessaging mockedMessaging = mock(IMessaging.class); 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, mockedMessaging); + 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); From 78daf364185d807332cd01f56b20107d4dadf1e3 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 19:21:55 +0200 Subject: [PATCH 18/18] Minor fix, cleaned imports --- .../mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index b4a3a327..393bbbc0 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -9,7 +9,6 @@ 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.IMessaging; 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; @@ -27,7 +26,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import org.junit.Before; import org.junit.Test; -import static org.mockito.Mockito.*; /** *