Added handling of publish messages to the server

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

@@ -15,7 +15,7 @@ public interface IMessaging {
      * notification also after all the current subscriber to the given topic has
      * been notified.
      */
-    void publish(String topic, Object message, byte qos, boolean retain) ;
+    void publish(String topic, byte[] message, QOSType qos, boolean retain);
     
     /**
      * Subscribe a client to a specified topic with a defined level

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

@@ -0,0 +1,13 @@
+package org.dna.mqtt.moquette.messaging.spi;
+
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
+
+/**
+ *
+ * @author andrea
+ */
+public interface INotifier {
+
+    public void notify(String clientId, String topic, QOSType qOSType, byte[] payload);
+    
+}

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

@@ -3,6 +3,7 @@ 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.messaging.spi.INotifier;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
 
 /**
@@ -10,19 +11,36 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
  * @author andrea
  */
 public class SimpleMessaging implements IMessaging {
-    
+    //TODO is poorly performace on concurrent load
     private List<Subscription> subscriptions = new ArrayList<Subscription>();
+    
+    private INotifier m_notifier;
+    
+    public void setNotifier(INotifier notifier) {
+        m_notifier = notifier;
+    }
 
-    public void publish(String topic, Object message, byte qos, boolean retain) {
-        throw new UnsupportedOperationException("Not supported yet.");
+    public void publish(String topic, byte[] message, QOSType qos, boolean retain) {
+        //TODO for this moment no retain management and no qos > 0
+        //trivial implementation
+        QOSType defQos = QOSType.MOST_ONE;
+        synchronized(subscriptions) {
+            for (Subscription sub : subscriptions) {
+                if (sub.match(topic)) {
+                    m_notifier.notify(sub.clientId, topic, defQos, message);
+                }
+            }
+        }
     }
 
     public void subscribe(String clientId, String topic, QOSType qos) {
         Subscription newSubscription = new Subscription(clientId, topic, qos);
-        if (subscriptions.contains(newSubscription)) {
-            return;
+        synchronized(subscriptions) {
+            if (subscriptions.contains(newSubscription)) {
+                return;
+            }
+            subscriptions.add(newSubscription);
         }
-        subscriptions.add(newSubscription);
     }
     
     protected List<Subscription> getSubscriptions() {

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

@@ -62,4 +62,11 @@ public class Subscription {
         hash = 37 * hash + (this.topic != null ? this.topic.hashCode() : 0);
         return hash;
     }
+
+    /**
+     * Trivial match method
+     */
+    boolean match(String topic) {
+        return this.topic.equals(topic);
+    }
 }

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

@@ -1,5 +1,7 @@
 package org.dna.mqtt.moquette.server;
 
+import org.dna.mqtt.moquette.messaging.spi.INotifier;
+import org.dna.mqtt.moquette.proto.messages.PublishMessage;
 import org.dna.mqtt.moquette.messaging.spi.IMessaging;
 import java.util.HashMap;
 import java.util.Map;
@@ -18,7 +20,7 @@ import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
  *
  * @author andrea
  */
-public class MQTTHandler extends IoHandlerAdapter {
+public class MQTTHandler extends IoHandlerAdapter implements INotifier {
     protected static final String ATTR_CLIENTID = "ClientID";
 
     private static final Logger LOG = Logger.getLogger(MQTTHandler.class.getName());
@@ -37,6 +39,8 @@ public class MQTTHandler extends IoHandlerAdapter {
                 handleConnect(session, (ConnectMessage) msg);
             case SUBSCRIBE:
                 handleSubscribe(session, (SubscribeMessage) msg);
+            case PUBLISH:
+                handlePublish(session, (PublishMessage) msg);
         }
     }
 
@@ -74,7 +78,9 @@ public class MQTTHandler extends IoHandlerAdapter {
 
         //Handle will flag
         if (msg.isWillFlag()) {
-            m_messaging.publish(msg.getWillTopic(), msg.getWillMessage(), msg.getWillQos(), msg.isWillRetain());
+            QOSType willQos = QOSType.values()[msg.getWillQos()];
+            m_messaging.publish(msg.getWillTopic(), msg.getWillMessage().getBytes(), 
+                    willQos, msg.isWillRetain());
         }
 
         //handle user authentication
@@ -114,7 +120,12 @@ public class MQTTHandler extends IoHandlerAdapter {
         }
         session.write(ackMessage);
     }
-
+    
+    protected void handlePublish(IoSession session, PublishMessage message) {
+        m_messaging.publish(message.getTopicName(), message.getPayload(), 
+                message.getQos(), message.isRetainFlag());
+    }
+    
     @Override
     public void sessionIdle(IoSession session, IdleStatus status) {
         session.close(false);
@@ -127,4 +138,12 @@ public class MQTTHandler extends IoHandlerAdapter {
     public void setAuthenticator(IAuthenticator authenticator) {
         m_authenticator = authenticator;
     }
+
+    public void notify(String clientId, String topic, QOSType qOSType, byte[] payload) {
+        PublishMessage pubMessage = new PublishMessage();
+        pubMessage.setTopicName(topic);
+        pubMessage.setQos(qOSType);
+        pubMessage.setPayload(payload);
+        m_clientIDs.get(clientId).getSession().write(pubMessage);
+    }
 }

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

@@ -1,9 +1,12 @@
 package org.dna.mqtt.moquette.messaging.spi.impl;
 
+import org.mockito.ArgumentCaptor;
 import java.util.List;
+import org.dna.mqtt.moquette.messaging.spi.INotifier;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
 import org.junit.Test;
 import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
 
 /**
  *
@@ -68,4 +71,23 @@ public class SimpleMessagingTest {
         }
         return count;
     }
+    
+    
+    @Test
+    public void testPublish() {
+        String clientID = "FAKE_123";
+        String topic = "/news";
+        SimpleMessaging messaging = new SimpleMessaging();
+        messaging.subscribe(clientID, topic, QOSType.MOST_ONE);
+        INotifier notifier = mock(INotifier.class);
+        messaging.setNotifier(notifier);
+        
+        //Exercise
+        messaging.publish(topic, "Hello".getBytes(), QOSType.MOST_ONE, false);
+        
+        //Verify
+        ArgumentCaptor<byte[]> argument = ArgumentCaptor.forClass(byte[].class);
+        verify(notifier).notify(eq(clientID), eq(topic), any(QOSType.class), argument.capture());
+        assertEquals("Hello", new String(argument.getValue()));
+    }
 }

parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java

@@ -85,7 +85,8 @@ public class MQTTHandlerTest {
         
         //Verify
         assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
-        verify(mockedMessaging).publish(eq("topic"), eq("Topic message"), anyByte(), anyBoolean());
+        verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), 
+                any(AbstractMessage.QOSType.class), anyBoolean());
     }