From 0694f28f320c0730f734b88a1e194c9551b2be7e Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 19:21:11 +0200 Subject: [PATCH] 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);