From 3be2ccca8f18aea895492d169d6c14c9a1b21e9b Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 10:22:53 +0200 Subject: [PATCH] Added IoSession attribute to PublishEvent to avoid looking up --- .../mqtt/moquette/messaging/spi/IMessaging.java | 4 ++-- .../messaging/spi/impl/SimpleMessaging.java | 12 ++++++++---- .../messaging/spi/impl/events/PublishEvent.java | 14 +++++++++++--- .../messaging/spi/impl/SimpleMessagingTest.java | 2 +- .../dna/mqtt/moquette/server/MQTTHandlerTest.java | 4 ++-- 5 files changed, 24 insertions(+), 12 deletions(-) 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 cacf9da1..733f0615 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 @@ -18,12 +18,12 @@ public interface IMessaging { * notification also after all the current subscriber to the given topic has * been notified. */ - void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID); + void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session); /** * Used to publish messages with QoS > 0 * */ - void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID); + void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session); /** * Subscribe a client to a specified topic with a defined level 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 05c27b8c..7e97b652 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,6 +14,7 @@ import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.server.Constants; import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; import org.fusesource.hawtdb.api.BTreeIndexFactory; @@ -108,20 +109,20 @@ public class SimpleMessaging implements IMessaging, Runnable { } - public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID) { + public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session) { //TODO for this moment no qos > 0 trivial implementation // QOSType defQos = QOSType.MOST_ONE; try { - m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID)); + m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, session)); } catch (InterruptedException ex) { LOG.error(null, ex); } } - public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID) { + public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session) { try { - m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID)); + m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID, session)); } catch (InterruptedException ex) { LOG.error(null, ex); } @@ -267,8 +268,11 @@ public class SimpleMessaging implements IMessaging, Runnable { for (final Subscription sub : subscriptions.matches(topic)) { if (qos == QOSType.MOST_ONE) { + //QoS 0 m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); } else { + //QoS 1 or 2 + //if the target subscription is not clean session and is not connected => store it m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID())); } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index 6dbf1e5a..b30eb2ad 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; +import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import java.io.Serializable; @@ -16,19 +17,22 @@ public class PublishEvent extends MessagingEvent implements Serializable { String m_clientID; //Optional attribute, available only fo QoS 1 and 2 int m_msgID; + + transient IoSession m_session; public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, - String clientID) { + String clientID, IoSession session) { m_topic = topic; m_qos = qos; m_message = message; m_retain = retain; m_clientID = clientID; + m_session = session; } public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, - String clientID, int msgID) { - this(topic, qos, message, retain, clientID); + String clientID, int msgID, IoSession session) { + this(topic, qos, message, retain, clientID, session); m_msgID = msgID; } @@ -55,4 +59,8 @@ public class PublishEvent extends MessagingEvent implements Serializable { public int getMessageID() { return m_msgID; } + + public IoSession getSession() { + return m_session; + } } 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 811ce739..38083ba6 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 @@ -57,7 +57,7 @@ public class SimpleMessagingTest { // messaging.setNotifier(notifier); //Exercise - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI"); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); messaging.processPublish(pubEvt); //Verify diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java index 22c79781..2c143703 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -86,7 +86,7 @@ public class MQTTHandlerTest { //Verify assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), - any(AbstractMessage.QOSType.class), anyBoolean(), eq("123")); + any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class)); } @@ -99,7 +99,7 @@ public class MQTTHandlerTest { (byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName)); IMessaging mockedMessaging = mock(IMessaging.class); - m_session.setAttribute(MQTTHandler.ATTR_CLIENTID, "fakeID"); + m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID"); //Exercise