Separated all mina related stuff in classes to avoid main protocol loop to depend on mina related stuff

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java

@@ -1,7 +1,7 @@
 package org.dna.mqtt.moquette.messaging.spi;
 
-import org.apache.mina.core.session.IoSession;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
+import org.dna.mqtt.moquette.server.ServerChannel;
 
 /**
  * Interface to the underling messaging system used to publish, subscribe.
@@ -12,7 +12,7 @@ public interface IMessaging {
 
     void stop();
 
-    void disconnect(IoSession session);
+    void disconnect(ServerChannel session);
 
-    void handleProtocolMessage(IoSession session, AbstractMessage msg);
+    void handleProtocolMessage(ServerChannel session, AbstractMessage msg);
 }

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java

@@ -3,8 +3,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl;
 import java.util.Collection;
 import java.util.List;
 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.IMatchingCondition;
 import org.dna.mqtt.moquette.messaging.spi.IStorageService;
 import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
@@ -25,6 +23,7 @@ import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
 import org.dna.mqtt.moquette.server.ConnectionDescriptor;
 import org.dna.mqtt.moquette.server.Constants;
 import org.dna.mqtt.moquette.server.IAuthenticator;
+import org.dna.mqtt.moquette.server.ServerChannel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -56,7 +55,7 @@ class ProtocolProcessor {
         m_storageService = storageService;
     }
     
-    void processConnect(IoSession session, ConnectMessage msg) {
+    void processConnect(ServerChannel session, ConnectMessage msg) {
         if (msg.getProcotolVersion() != 0x03) {
             ConnAckMessage badProto = new ConnAckMessage();
             badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
@@ -75,7 +74,7 @@ class ProtocolProcessor {
         //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();
+            ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).getSession();
             boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
             if (cleanSession) {
                 //cleanup topic subscriptions
@@ -95,7 +94,7 @@ class ProtocolProcessor {
         //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));
+        session.setIdleTime(Math.round(keepAlive * 1.5f));
 
         //Handle will flag
         if (msg.isWillFlag()) {
@@ -331,7 +330,7 @@ class ProtocolProcessor {
         m_storageService.cleanInFlight(publishKey);
     }
     
-    void processDisconnect(IoSession session, String clientID, boolean cleanSession) throws InterruptedException {
+    void processDisconnect(ServerChannel session, String clientID, boolean cleanSession) throws InterruptedException {
         if (cleanSession) {
             //cleanup topic subscriptions
             processRemoveAllSubscriptions(clientID);
@@ -349,7 +348,7 @@ class ProtocolProcessor {
      * Remove the clientID from topic subscription, if not previously subscribed,
      * doesn't reply any error
      */
-    void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
+    void processUnsubscribe(ServerChannel session, String clientID, List<String> topics, int messageID) {
         LOG.debug("processSubscribe invoked");
 
         for (String topic : topics) {
@@ -364,7 +363,7 @@ class ProtocolProcessor {
     }
     
     
-    void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) {
+    void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) {
         LOG.debug("processSubscribe invoked");
 
         for (SubscribeMessage.Couple req : msg.subscriptions()) {

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java

@@ -5,11 +5,9 @@ import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.SequenceBarrier;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-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.*;
@@ -19,6 +17,7 @@ import org.dna.mqtt.moquette.proto.messages.*;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
 import org.dna.mqtt.moquette.server.ConnectionDescriptor;
 import org.dna.mqtt.moquette.server.Constants;
+import org.dna.mqtt.moquette.server.ServerChannel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -83,11 +82,11 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     }
     
 
-    public void disconnect(IoSession session) {
+    public void disconnect(ServerChannel session) {
         disruptorPublish(new DisconnectEvent(session));
     }
 
-    public void handleProtocolMessage(IoSession session, AbstractMessage msg) {
+    public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
         disruptorPublish(new ProtocolEvent(session, msg));
     }
 
@@ -107,7 +106,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
             String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
             m_processor.processDisconnect(disEvt.getSession(), clientID, false);
         } else if (evt instanceof ProtocolEvent) {
-            IoSession session = ((ProtocolEvent) evt).getSession();
+            ServerChannel session = ((ProtocolEvent) evt).getSession();
             AbstractMessage message = ((ProtocolEvent) evt).getMessage();
             if (message instanceof ConnectMessage) {
                 m_processor.processConnect(session, (ConnectMessage) message);

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java

@@ -1,6 +1,6 @@
 package org.dna.mqtt.moquette.messaging.spi.impl.events;
 
-import org.apache.mina.core.session.IoSession;
+import org.dna.mqtt.moquette.server.ServerChannel;
 
 /**
  *
@@ -8,13 +8,13 @@ import org.apache.mina.core.session.IoSession;
  */
 public class DisconnectEvent extends MessagingEvent {
     
-    IoSession m_session;
+    ServerChannel m_session;
     
-    public DisconnectEvent(IoSession session) {
+    public DisconnectEvent(ServerChannel session) {
         m_session = session;
     }
 
-    public IoSession getSession() {
+    public ServerChannel getSession() {
         return m_session;
     }
     

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java

@@ -1,21 +1,21 @@
 package org.dna.mqtt.moquette.messaging.spi.impl.events;
 
-import org.apache.mina.core.session.IoSession;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
+import org.dna.mqtt.moquette.server.ServerChannel;
 
 /**
  * Event used to carry ProtocolMessages from front handler to event processor
  */
 public class ProtocolEvent extends MessagingEvent {
-    IoSession m_session;
+    ServerChannel m_session;
     AbstractMessage message;
 
-    public ProtocolEvent(IoSession session, AbstractMessage message) {
+    public ProtocolEvent(ServerChannel session, AbstractMessage message) {
         this.m_session = session;
         this.message = message;
     }
 
-    public IoSession getSession() {
+    public ServerChannel getSession() {
         return m_session;
     }
 

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java

@@ -1,9 +1,9 @@
 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;
+import org.dna.mqtt.moquette.server.ServerChannel;
 
 /**
  *
@@ -18,10 +18,10 @@ public class PublishEvent extends MessagingEvent implements Serializable {
     //Optional attribute, available only fo QoS 1 and 2
     int m_msgID;
 
-    transient IoSession m_session;
+    transient ServerChannel m_session;
     
     public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
-            String clientID, IoSession session) {
+            String clientID, ServerChannel session) {
         m_topic = topic;
         m_qos = qos;
         m_message = message;
@@ -31,7 +31,7 @@ public class PublishEvent extends MessagingEvent implements Serializable {
     }
 
     public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
-                        String clientID, int msgID, IoSession session) {
+                        String clientID, int msgID, ServerChannel session) {
         this(topic, qos, message, retain, clientID, session);
         m_msgID = msgID;
     }
@@ -60,7 +60,7 @@ public class PublishEvent extends MessagingEvent implements Serializable {
         return m_msgID;
     }
 
-    public IoSession getSession() {
+    public ServerChannel getSession() {
         return m_session;
     }
 

broker/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java

@@ -1,7 +1,5 @@
 package org.dna.mqtt.moquette.server;
 
-import org.apache.mina.core.session.IoSession;
-
 /**
  * Maintains the information of single connection, like ClientID, IoSession,
  * and other connection related flags.
@@ -12,10 +10,10 @@ import org.apache.mina.core.session.IoSession;
 public class ConnectionDescriptor {
     
     private String m_clientID;
-    private IoSession m_session;
+    private ServerChannel m_session;
     private boolean m_cleanSession;
     
-    public ConnectionDescriptor(String clientID, IoSession session, boolean cleanSession) {
+    public ConnectionDescriptor(String clientID, ServerChannel session, boolean cleanSession) {
         this.m_clientID = clientID;
         this.m_session = session;
         this.m_cleanSession = cleanSession;
@@ -29,7 +27,7 @@ public class ConnectionDescriptor {
         return m_clientID;
     }
 
-    public IoSession getSession() {
+    public ServerChannel getSession() {
         return m_session;
     }
 }

broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java

@@ -1,5 +1,7 @@
 package org.dna.mqtt.moquette.server;
 
+import java.util.HashMap;
+import java.util.Map;
 import org.apache.mina.core.service.IoHandlerAdapter;
 import org.apache.mina.core.session.IdleStatus;
 import org.apache.mina.core.session.IoSession;
@@ -8,6 +10,7 @@ import org.dna.mqtt.moquette.proto.Utils;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
 import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
+import org.dna.mqtt.moquette.server.mina.MinaChannel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -20,6 +23,7 @@ public class MQTTHandler extends IoHandlerAdapter {
 
     private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
     private IMessaging m_messaging;
+    private Map<IoSession, MinaChannel> m_channelMapper = new HashMap<IoSession, MinaChannel>();
 
     @Override
     public void messageReceived(IoSession session, Object message) throws Exception {
@@ -35,7 +39,11 @@ public class MQTTHandler extends IoHandlerAdapter {
                 case PUBCOMP:
                 case PUBREL:
                 case DISCONNECT:
-                    m_messaging.handleProtocolMessage(session, msg);
+                    if (!m_channelMapper.containsKey(session)) {
+                        m_channelMapper.put(session, new MinaChannel(session));
+                    }
+                    
+                    m_messaging.handleProtocolMessage(m_channelMapper.get(session), msg);
                     break;
                 case PINGREQ:
                     PingRespMessage pingResp = new PingRespMessage();
@@ -51,6 +59,7 @@ public class MQTTHandler extends IoHandlerAdapter {
     public void sessionIdle(IoSession session, IdleStatus status) {
         if (status == IdleStatus.READER_IDLE) {
             session.close(false);
+            m_channelMapper.remove(session);
             //TODO send a notification to messaging part to remove the bining clientID-ConnConfig
         }
     }

broker/src/main/java/org/dna/mqtt/moquette/server/ServerChannel.java

@@ -0,0 +1,18 @@
+package org.dna.mqtt.moquette.server;
+
+/**
+ *
+ * @author andrea
+ */
+public interface ServerChannel {
+    
+    Object getAttribute(Object key);
+    
+    void setAttribute(Object key, Object value);
+    
+    void setIdleTime(int idleTime);
+    
+    void close(boolean immediately);
+    
+    void write(Object value);
+}

broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java

@@ -0,0 +1,40 @@
+package org.dna.mqtt.moquette.server.mina;
+
+import org.apache.mina.core.session.IdleStatus;
+import org.apache.mina.core.session.IoSession;
+import org.dna.mqtt.moquette.server.ServerChannel;
+
+/**
+ *
+ * @author andrea
+ */
+public class MinaChannel implements ServerChannel {
+    
+    private IoSession m_session;
+    
+    public MinaChannel(IoSession session) {
+        m_session = session;
+    }
+
+    public Object getAttribute(Object key) {
+        return m_session.getAttribute(key);
+    }
+    
+    public void setAttribute(Object key, Object value) {
+        m_session.setAttribute(key, value);
+    }
+
+    public void close(boolean immediately) {
+        m_session.close(immediately);
+    }
+
+    public void write(Object value) {
+        m_session.write(value);
+    }
+
+    public void setIdleTime(int idleTime) {
+        m_session.getConfig().setIdleTime(IdleStatus.READER_IDLE, idleTime);
+    }
+
+    
+}

broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java

@@ -21,6 +21,8 @@ import org.dna.mqtt.moquette.proto.messages.PublishMessage;
 import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
 import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
 import org.dna.mqtt.moquette.server.ConnectionDescriptor;
+import org.dna.mqtt.moquette.server.ServerChannel;
+import org.dna.mqtt.moquette.server.mina.MinaChannel;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
@@ -35,7 +37,7 @@ public class ProtocolProcessorTest {
     final static String FAKE_CLIENT_ID = "FAKE_123";
     final static String FAKE_TOPIC = "/news";
     
-    IoSession m_session;
+    ServerChannel m_session;
     byte m_returnCode;
     ConnectMessage connMsg;
     ProtocolProcessor m_processor;
@@ -44,15 +46,19 @@ public class ProtocolProcessorTest {
     SubscriptionsStore subscriptions;
     AbstractMessage m_receivedMessage;
     
+    IoSession m_minaSession;
+    
     
     @Before
     public void setUp() throws InterruptedException {
         connMsg = new ConnectMessage();
         connMsg.setProcotolVersion((byte) 0x03);
 
-        m_session = new DummySession();
+        m_minaSession = new DummySession();
         
-        m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
+        m_session = new MinaChannel(m_minaSession);
+
+        m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
 
             @Override
             public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
@@ -190,8 +196,8 @@ public class ProtocolProcessorTest {
     
     @Test
     public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
-        m_session.getFilterChain().remove("MessageCatcher");
+        m_minaSession.getFilterChain().remove("MessageCatcher");
-        m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
+        m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
 
             @Override
             public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,

broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java

@@ -51,6 +51,7 @@ public class ServerIntegrationFuseTest {
 
     @Test
     public void testSubscribe() throws Exception {
+        System.out.println("*** testSubscribe ***");
         m_connection = m_mqtt.blockingConnection();
         m_connection.connect();
 
@@ -66,6 +67,7 @@ public class ServerIntegrationFuseTest {
     
     @Test
     public void testCleanSession_maintainClientSubscriptions() throws Exception {
+        System.out.println("*** testCleanSession_maintainClientSubscriptions ***");
         MQTT localMqtt = new MQTT(m_mqtt);
         localMqtt.setCleanSession(false);
         BlockingConnection connection = localMqtt.blockingConnection();
@@ -87,6 +89,7 @@ public class ServerIntegrationFuseTest {
     
     @Test
     public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() throws Exception {
+        System.out.println("*** testCleanSession_maintainClientSubscriptions_againstClientDestruction ***");
         MQTT localMqtt = new MQTT(m_mqtt);
         localMqtt.setCleanSession(false);
         BlockingConnection connection = localMqtt.blockingConnection();
@@ -114,6 +117,7 @@ public class ServerIntegrationFuseTest {
      */
     @Test
     public void testCleanSession_correctlyClientSubscriptions() throws Exception {
+        System.out.println("*** testCleanSession_correctlyClientSubscriptions ***");
         MQTT localMqtt = new MQTT(m_mqtt);
         localMqtt.setCleanSession(false);
         BlockingConnection connection = localMqtt.blockingConnection();
@@ -134,6 +138,7 @@ public class ServerIntegrationFuseTest {
     
     @Test
     public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws Exception {
+        System.out.println("*** testCleanSession_maintainClientSubscriptions_withServerRestart ***");
         MQTT localMqtt = new MQTT(m_mqtt);
         localMqtt.setCleanSession(false);
         BlockingConnection connection = localMqtt.blockingConnection();
@@ -160,6 +165,7 @@ public class ServerIntegrationFuseTest {
     
     @Test
     public void testRetain_maintainMessage_againstClientDestruction() throws Exception {
+        System.out.println("*** testRetain_maintainMessage_againstClientDestruction ***");
         m_connection = m_mqtt.blockingConnection();
         m_connection.connect(); 
         m_connection.publish("/topic", "Test my payload".getBytes(), QoS.AT_MOST_ONCE, true);
@@ -176,6 +182,7 @@ public class ServerIntegrationFuseTest {
     
     @Test
     public void testUnsubscribe_do_not_notify_anymore_same_session() throws Exception {
+        System.out.println("*** testUnsubscribe_do_not_notify_anymore_same_session ***");
         m_connection = m_mqtt.blockingConnection();
         m_connection.connect(); 
         m_connection.subscribe(new Topic[] {new Topic("/topic", QoS.AT_MOST_ONCE)});
@@ -195,6 +202,7 @@ public class ServerIntegrationFuseTest {
     
     @Test
     public void testUnsubscribe_do_not_notify_anymore_new_session() throws Exception {
+        System.out.println("*** testUnsubscribe_do_not_notify_anymore_new_session ***");
         m_connection = m_mqtt.blockingConnection();
         m_connection.connect(); 
         m_connection.subscribe(new Topic[] {new Topic("/topic", QoS.AT_MOST_ONCE)});
@@ -217,6 +225,7 @@ public class ServerIntegrationFuseTest {
     
     @Test
     public void testPublishWithQoS1() throws Exception {
+        System.out.println("*** testPublishWithQoS1 ***");
         m_connection = m_mqtt.blockingConnection();
         m_connection.connect();
 
@@ -236,6 +245,7 @@ public class ServerIntegrationFuseTest {
 
     @Test
     public void testPublishWithQoS1_notCleanSession() throws Exception {
+        System.out.println("*** testPublishWithQoS1_notCleanSession ***");
         MQTT mqtt = new MQTT(m_mqtt);
         mqtt.setCleanSession(false);
         m_connection = mqtt.blockingConnection();
@@ -275,6 +285,7 @@ public class ServerIntegrationFuseTest {
 
     @Test
     public void testPublishWithQoS2() throws Exception {
+        System.out.println("*** testPublishWithQoS2 ***");
         m_mqtt.setClientId("TestClient");
         m_connection = m_mqtt.blockingConnection();
         m_connection.connect();
@@ -295,6 +306,7 @@ public class ServerIntegrationFuseTest {
 
     @Test
     public void testPublishReceiveWithQoS2() throws Exception {
+        System.out.println("*** testPublishReceiveWithQoS2 ***");
         m_mqtt.setCleanSession(false);
         m_connection = m_mqtt.blockingConnection();
         m_connection.connect();