From 0e1a3cef6c5b001a6cea3cae22fb3f5101563aa7 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Tue, 4 Sep 2012 16:20:53 +0200 Subject: [PATCH] Moved subscription logic inside the event loop --- .../moquette/messaging/spi/IMessaging.java | 5 -- .../messaging/spi/impl/SimpleMessaging.java | 60 +++++++++---------- .../spi/impl/events/ConnectEvent.java | 25 -------- .../dna/mqtt/moquette/server/MQTTHandler.java | 18 +----- .../spi/impl/SimpleMessagingTest.java | 27 ++++++--- 5 files changed, 46 insertions(+), 89 deletions(-) delete mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index ec3da3ec..3cad0278 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -14,11 +14,6 @@ import org.dna.mqtt.moquette.proto.messages.ConnectMessage; */ public interface IMessaging { - /** - * Subscribe a client to a specified topic with a defined level - */ - void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID); - void stop(); void disconnect(IoSession session); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 0c003f3a..626572a9 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -14,7 +14,6 @@ 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.apache.mina.filter.util.SessionAttributeInitializingFilter; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; @@ -115,13 +114,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { } - public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID) { - Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession); - LOG.debug("subscribe invoked for topic: " + topic); - disruptorPublish(new SubscribeEvent(newSubscription, messageID)); - } - - public void disconnect(IoSession session) { disruptorPublish(new DisconnectEvent(session)); } @@ -178,9 +170,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { return subscriptions; } -/* public void removeSubscriptions(String clientID) { - disruptorPublish(new RemoveAllSubscriptionsEvent(clientID)); - }*/ public void stop() { disruptorPublish(new StopEvent()); @@ -191,8 +180,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { MessagingEvent evt = t.getEvent(); if (evt instanceof PublishEvent) { processPublish((PublishEvent) evt); - } else if (evt instanceof SubscribeEvent) { - processSubscribe((SubscribeEvent) evt); } else if (evt instanceof StopEvent) { processStop(); } else if (evt instanceof DisconnectEvent) { @@ -237,7 +224,10 @@ public class SimpleMessaging implements IMessaging, EventHandler { UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID()); - + } else if (message instanceof SubscribeMessage) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); + processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); } else { throw new RuntimeException("Illegal message received " + message); } @@ -373,11 +363,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } - protected void processSubscribe(SubscribeEvent evt) { - LOG.debug("processSubscribe invoked"); - Subscription newSubscription = evt.getSubscription(); - final String topic = newSubscription.getTopic(); - + private void subscribeSingleTopic(Subscription newSubscription, final String topic) { subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription @@ -390,22 +376,30 @@ 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.clientId, topic, storedMsg.getQos(), - storedMsg.getPayload(), true)); + notify(new NotifyEvent(newSubscription.clientId, topic, 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(evt.getMessageID()); -// -// //TODO by now it handles only QoS 0 messages -// /*for (int i = 0; i < evt.subscriptions().size(); i++) { -// ackMessage.addType(QOSType.MOST_ONE); -// } */ -// LOG.info("replying with SubAct to MSG ID {0}", evt.getMessageID()); -// //session.write(ackMessage); -// IoSession session = m_clientIDs.get(evt.getSubscription().clientId).getSession(); -// session.write(ackMessage).awaitUninterruptibly(); + //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 {0}", msg.getMessageID()); + session.write(ackMessage); + } /** diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java deleted file mode 100644 index 7e2f6bea..00000000 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java +++ /dev/null @@ -1,25 +0,0 @@ -package org.dna.mqtt.moquette.messaging.spi.impl.events; - -import org.apache.mina.core.session.IoSession; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; - -/** - */ -public class ConnectEvent extends MessagingEvent { - - IoSession m_session; - ConnectMessage message; - - public ConnectEvent(IoSession session, ConnectMessage message) { - this.m_session = session; - this.message = message; - } - - public IoSession getSession() { - return m_session; - } - - public ConnectMessage getMessage() { - return message; - } -} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 904b47c1..535a8f14 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -66,23 +66,7 @@ public class MQTTHandler extends IoHandlerAdapter { protected void handleSubscribe(IoSession session, SubscribeMessage msg) { LOG.debug("handleSubscribe, registering the subscriptions"); - for (SubscribeMessage.Couple req : msg.subscriptions()) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); - m_messaging.subscribe(clientID, req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()], - cleanSession, msg.getMessageID()); - } - - //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 {0}", msg.getMessageID()); - session.write(ackMessage); + m_messaging.handleProtocolMessage(session, msg); } private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) { diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index 4dda658d..ee1ee312 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -76,8 +76,9 @@ public class SimpleMessagingTest { @Test public void testSubscribe() { //Exercise - SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0); - messaging.processSubscribe(evt); + 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); @@ -86,11 +87,14 @@ public class SimpleMessagingTest { @Test public void testDoubleSubscribe() { - SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0); - messaging.processSubscribe(evt); +// 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 - messaging.processSubscribe(evt); + messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); //Verify Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false); @@ -103,8 +107,11 @@ public class SimpleMessagingTest { 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); - messaging.processSubscribe(evt); +// 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); @@ -198,8 +205,10 @@ public class SimpleMessagingTest { //Exercise //m_handler.setMessaging(mockedMessaging); - SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0); - messaging.processSubscribe(pubEvt); +// 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);