Added IoSession attribute to PublishEvent to avoid looking up

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

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()));
             }
         }

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;
+    }
 }

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

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