First step in moving protocol message processing inside the event handler

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

@@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.messaging.spi;
 import org.apache.mina.core.session.IoSession;
 import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
 import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
 import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
 
@@ -49,5 +50,5 @@ public interface IMessaging {
 
     void republishStored(String clientID);
 
-    void connect(IoSession session, ConnectMessage msg);
+    void handleProtocolMessage(IoSession session, AbstractMessage msg);
 }

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

@@ -28,6 +28,10 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
+ *
+ * Singleton class that orchestrate the execution of the protocol.
+ *
+ * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion.
  *
  * @author andrea
  */
@@ -144,8 +148,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
         disruptorPublish(new RepublishEvent(clientID));
     }
 
-    public void connect(IoSession session, ConnectMessage msg) {
-        disruptorPublish(new ConnectEvent(session, msg));
+    public void handleProtocolMessage(IoSession session, AbstractMessage msg) {
+        disruptorPublish(new ProtocolEvent(session, msg));
     }
 
     /**
@@ -214,8 +218,15 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
             m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId());
         } else if (evt instanceof RepublishEvent) {
             processRepublish((RepublishEvent) evt);
-        } else if (evt instanceof ConnectEvent) {
-            processConnect((ConnectEvent) evt);
+        } else if (evt instanceof ProtocolEvent) {
+            IoSession session = ((ProtocolEvent) evt).getSession();
+            AbstractMessage message = ((ProtocolEvent) evt).getMessage();
+            if (message instanceof ConnectMessage) {
+                processConnect(session, (ConnectMessage) message);
+            } else {
+                throw new RuntimeException("Illegal message received " + message);
+            }
+
         } else if (evt instanceof InitEvent) {
             processInit();
         }
@@ -229,10 +240,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
     }
 
 
-    protected void processConnect(ConnectEvent evt) {
-        ConnectMessage msg = evt.getMessage();
-        IoSession session = evt.getSession();
-
+    protected void processConnect(IoSession session, ConnectMessage msg) {
         if (msg.getProcotolVersion() != 0x03) {
             ConnAckMessage badProto = new ConnAckMessage();
             badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);

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

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

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

@@ -61,7 +61,7 @@ public class MQTTHandler extends IoHandlerAdapter {
     protected void handleConnect(IoSession session, ConnectMessage msg) {
         LOG.info("handleConnect invoked");
 
-        m_messaging.connect(session, msg);
+        m_messaging.handleProtocolMessage(session, msg);
     }
 
     protected void handleSubscribe(IoSession session, SubscribeMessage msg) {

parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java

@@ -146,8 +146,7 @@ public class SimpleMessagingTest {
         connMsg.setProcotolVersion((byte) 0x02);
 
         //Exercise
-        ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
-        messaging.processConnect(pubEvt);
+        messaging.processConnect(m_session, connMsg);
 
         //Verify
         assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
@@ -158,8 +157,7 @@ public class SimpleMessagingTest {
         connMsg.setClientID("extremely_long_clientID_greater_than_23");
 
         //Exercise
-        ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
-        messaging.processConnect(pubEvt);
+        messaging.processConnect(m_session, connMsg);
 
         //Verify
         assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
@@ -175,8 +173,7 @@ public class SimpleMessagingTest {
 
         //Exercise
         //m_handler.setMessaging(mockedMessaging);
-        ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
-        messaging.processConnect(pubEvt);
+        messaging.processConnect(m_session, connMsg);
 
         //Verify
         assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);