diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 379ebe53..8fbd48a1 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/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); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 5388f3ae..1d457c3b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/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 topics, int messageID) { + void processUnsubscribe(ServerChannel session, String clientID, List 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()) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index d8141dee..0dd55221 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/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 { } - 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 { 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); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java index bf8ecd27..f90ac9c7 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java +++ b/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; } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java index 7bc6036d..46a638ea 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java +++ b/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; } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index 316c3a65..7e2c2b5b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/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; } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java index 326bc196..f2a4634f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java +++ b/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; } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 507d4e06..36cfdfe5 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/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 m_channelMapper = new HashMap(); @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 } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ServerChannel.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerChannel.java new file mode 100644 index 00000000..b741de5f --- /dev/null +++ b/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); +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java new file mode 100644 index 00000000..be591d51 --- /dev/null +++ b/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); + } + + +} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 393bbbc0..9021579a 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/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 = new MinaChannel(m_minaSession); - m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + 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_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + m_minaSession.getFilterChain().remove("MessageCatcher"); + m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { @Override public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java index ded1696e..cea7268e 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/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();