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 5a6d2ef1..58ea792e 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 @@ -1,11 +1,13 @@ package org.dna.mqtt.moquette.messaging.spi; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; + /** * Interface to the underling messaging system used to publish, subscribe. * * @author andrea */ -public class IMessaging { +public interface IMessaging { /** * Publish a message on the messaging system. The qos is the level of quality @@ -13,8 +15,10 @@ public class IMessaging { * notification also after all the current subscriber to the given topic has * been notified. */ - public void publish(String topic, Object message, byte qos, boolean retain) { - throw new UnsupportedOperationException("Not yet implemented"); - } + void publish(String topic, Object message, byte qos, boolean retain) ; + /** + * Subscribe a client to a specified topic with a defined level + */ + void subscribe(String clientId, String topic, QOSType qos); } 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 new file mode 100644 index 00000000..42ed999f --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -0,0 +1,27 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import java.util.ArrayList; +import java.util.List; +import org.dna.mqtt.moquette.messaging.spi.IMessaging; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; + +/** + * + * @author andrea + */ +public class SimpleMessaging implements IMessaging { + + private List subscriptions = new ArrayList(); + + public void publish(String topic, Object message, byte qos, boolean retain) { + throw new UnsupportedOperationException("Not supported yet."); + } + + public void subscribe(String clientId, String topic, QOSType qos) { + subscriptions.add(new Subscription(clientId, topic, qos)); + } + + protected List getSubscriptions() { + return subscriptions; + } +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java new file mode 100644 index 00000000..4c484348 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java @@ -0,0 +1,65 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; + +/** + * Maintain the information about which Topic a certain ClientID is subscribed + * and at which QoS + * + * + * @author andrea + */ +public class Subscription { + + QOSType requestedQos; + String clientId; + String topic; + + public Subscription(String clientId, String topic, QOSType requestedQos) { + this.requestedQos = requestedQos; + this.clientId = clientId; + this.topic = topic; + } + + public String getClientId() { + return clientId; + } + + public QOSType getRequestedQos() { + return requestedQos; + } + + public String getTopic() { + return topic; + } + + @Override + public boolean equals(Object obj) { + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + final Subscription other = (Subscription) obj; + if (this.requestedQos != other.requestedQos) { + return false; + } + if ((this.clientId == null) ? (other.clientId != null) : !this.clientId.equals(other.clientId)) { + return false; + } + if ((this.topic == null) ? (other.topic != null) : !this.topic.equals(other.topic)) { + return false; + } + return true; + } + + @Override + public int hashCode() { + int hash = 3; + hash = 37 * hash + (this.requestedQos != null ? this.requestedQos.hashCode() : 0); + hash = 37 * hash + (this.clientId != null ? this.clientId.hashCode() : 0); + hash = 37 * hash + (this.topic != null ? this.topic.hashCode() : 0); + return hash; + } +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java b/parser/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java new file mode 100644 index 00000000..326bc196 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java @@ -0,0 +1,35 @@ +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. + * + * + * @author andrea + */ +public class ConnectionDescriptor { + + private String m_clientID; + private IoSession m_session; + private boolean m_cleanSession; + + public ConnectionDescriptor(String clientID, IoSession session, boolean cleanSession) { + this.m_clientID = clientID; + this.m_session = session; + this.m_cleanSession = cleanSession; + } + + public boolean isCleanSession() { + return m_cleanSession; + } + + public String getClientID() { + return m_clientID; + } + + public IoSession getSession() { + return m_session; + } +} 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 8ca5d5a0..296fd75d 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 @@ -10,6 +10,8 @@ import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; /** @@ -17,11 +19,12 @@ import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; * @author andrea */ public class MQTTHandler extends IoHandlerAdapter { + protected static final String ATTR_CLIENTID = "ClientID"; private static final Logger LOG = Logger.getLogger(MQTTHandler.class.getName()); /** Maps CLIENT_ID to the IoSession that represents the connection*/ - Map m_clientIDs = new HashMap(); + Map m_clientIDs = new HashMap(); private IMessaging m_messaging; private IAuthenticator m_authenticator; @@ -32,6 +35,8 @@ public class MQTTHandler extends IoHandlerAdapter { switch (msg.getMessageType()) { case CONNECT: handleConnect(session, (ConnectMessage) msg); + case SUBSCRIBE: + handleSubscribe(session, (SubscribeMessage) msg); } } @@ -53,13 +58,17 @@ public class MQTTHandler extends IoHandlerAdapter { //if an old client with the same ID already exists close its session. if (m_clientIDs.containsKey(msg.getClientID())) { - m_clientIDs.get(msg.getClientID()).close(false); + m_clientIDs.get(msg.getClientID()).getSession().close(false); } - m_clientIDs.put(msg.getClientID(), session); + + ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession()); + m_clientIDs.put(msg.getClientID(), connDescr); int keepAlive = msg.getKeepAlive(); session.setAttribute("keepAlive", keepAlive); session.setAttribute("cleanSession", msg.isCleanSession()); + //used to track the client in the subscription and publishing phases. + session.setAttribute(ATTR_CLIENTID, msg.getClientID()); session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f)); @@ -88,6 +97,23 @@ public class MQTTHandler extends IoHandlerAdapter { okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); session.write(okResp); } + + protected void handleSubscribe(IoSession session, SubscribeMessage msg) { + for(SubscribeMessage.Couple req : msg.subscriptions()) { + m_messaging.subscribe((String) session.getAttribute(ATTR_CLIENTID), + req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()]); + } + + //ack the client + SubAckMessage ackMessage = new SubAckMessage(); + ackMessage.setMessageID(msg.getMessageID()); + + //TODO by now it handles only QoS 0 messages + for (int i=0; i < msg.subscriptions().size(); i++) { + ackMessage.addType(QOSType.MOST_ONE); + } + session.write(ackMessage); + } @Override public void sessionIdle(IoSession session, IdleStatus status) { 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 new file mode 100644 index 00000000..f48612ab --- /dev/null +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -0,0 +1,48 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class SimpleMessagingTest { + +// public SimpleMessagingTest() { +// } +// +// @BeforeClass +// public static void setUpClass() throws Exception { +// } +// +// @AfterClass +// public static void tearDownClass() throws Exception { +// } +// +// @Before +// public void setUp() { +// } +// +// @After +// public void tearDown() { +// } + + @Test + public void testSubscribe() { + String clientID = "FAKE_123"; + String topic = "/news"; + SimpleMessaging messaging = new SimpleMessaging(); + + //Exercise + messaging.subscribe(clientID, topic, QOSType.MOST_ONE); + + //Verify + assertTrue(messaging.getSubscriptions().contains(new Subscription(clientID, topic, QOSType.MOST_ONE))); + } +} diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java index b08b1776..93b8ec68 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -1,5 +1,7 @@ package org.dna.mqtt.moquette.server; +import org.mockito.ArgumentCaptor; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.apache.mina.core.write.WriteRequest; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.apache.mina.core.filterchain.IoFilterAdapter; @@ -7,6 +9,7 @@ import org.apache.mina.core.session.DummySession; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.junit.Before; import org.junit.Test; import static org.junit.Assert.*; @@ -84,4 +87,27 @@ public class MQTTHandlerTest { assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); verify(mockedMessaging).publish(eq("topic"), eq("Topic message"), anyByte(), anyBoolean()); } + + + @Test + public void testHandleSubscribe() { + String topicName = "/news"; + SubscribeMessage subscribeMsg = new SubscribeMessage(); + subscribeMsg.setMessageID(0x555); + subscribeMsg.addSubscription(new SubscribeMessage.Couple( + (byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName)); + IMessaging mockedMessaging = mock(IMessaging.class); + + m_session.setAttribute(MQTTHandler.ATTR_CLIENTID, "fakeID"); + + + //Exercise + m_handler.setMessaging(mockedMessaging); + m_handler.handleSubscribe(m_session, subscribeMsg); + + //Verify + ArgumentCaptor argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class); + verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture()); + assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue()); + } }