From 82e998e939142a17739dcf5e4e71447ab1f256b3 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 13:06:52 +0200 Subject: [PATCH] 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";