Added simple implementation for handle subscribe message

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) {
+    void publish(String topic, Object message, byte qos, boolean retain) ;
-        throw new UnsupportedOperationException("Not yet implemented");
-    }
     
+    /**
+     * Subscribe a client to a specified topic with a defined level
+     */
+    void subscribe(String clientId, String topic, QOSType qos);
 }

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<Subscription> subscriptions = new ArrayList<Subscription>();
+
+    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<Subscription> getSubscriptions() {
+        return subscriptions;
+    }
+}

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;
+    }
+}

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;
+    }
+}

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<String, IoSession> m_clientIDs = new HashMap<String, IoSession>();
+    Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
     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));
 
@@ -89,6 +98,23 @@ public class MQTTHandler extends IoHandlerAdapter {
         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) {
         session.close(false);

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)));
+    }
+}

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<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);
+        verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture());
+        assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue());
+    }
 }