From bfc4193b7667a230b5836824119451a2c2e82223 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Tue, 4 Sep 2012 14:30:17 +0200 Subject: [PATCH] First step in moving protocol message processing inside the event handler --- .../moquette/messaging/spi/IMessaging.java | 3 ++- .../messaging/spi/impl/SimpleMessaging.java | 24 ++++++++++++------ .../spi/impl/events/ProtocolEvent.java | 25 +++++++++++++++++++ .../dna/mqtt/moquette/server/MQTTHandler.java | 2 +- .../spi/impl/SimpleMessagingTest.java | 9 +++---- 5 files changed, 47 insertions(+), 16 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.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 ce1958af..27eb963b 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 @@ -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); } 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 f3b61c64..5a820f08 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 @@ -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/*, 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/*, 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/*, } - 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); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java new file mode 100644 index 00000000..7bc6036d --- /dev/null +++ b/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; + } +} 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 5b15cf77..8a771b8f 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 @@ -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) { 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 c7d834f5..4dda658d 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 @@ -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);