Implementaed handlig of publish notification on the client

parser/src/main/java/org/dna/mqtt/moquette/client/Client.java

@@ -1,6 +1,8 @@
 package org.dna.mqtt.moquette.client;
 
 import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
@@ -14,7 +16,6 @@ import org.apache.mina.core.session.IoSession;
 import org.apache.mina.filter.codec.ProtocolCodecFilter;
 import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
 import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
-import org.apache.mina.filter.logging.LoggingFilter;
 import org.apache.mina.transport.socket.nio.NioSocketConnector;
 import org.dna.mqtt.moquette.ConnectionException;
 import org.dna.mqtt.moquette.MQTTException;
@@ -23,6 +24,7 @@ import org.dna.mqtt.moquette.SubscribeException;
 import org.dna.mqtt.moquette.proto.ConnAckDecoder;
 import org.dna.mqtt.moquette.proto.ConnectEncoder;
 import org.dna.mqtt.moquette.proto.DisconnectEncoder;
+import org.dna.mqtt.moquette.proto.PublishDecoder;
 import org.dna.mqtt.moquette.proto.PublishEncoder;
 import org.dna.mqtt.moquette.proto.SubAckDecoder;
 import org.dna.mqtt.moquette.proto.SubscribeEncoder;
@@ -62,6 +64,10 @@ public final class Client {
     private CountDownLatch m_connectBarrier;
     private CountDownLatch m_subscribeBarrier;
     private byte m_returnCode;
+    
+    //TODO synchronize the access
+    //Refact the da model should be a list of callback for each topic
+    private Map<String, IPublishCallback> m_subscribersList = new HashMap<String, IPublishCallback>();
 
     public Client(String host, int port) {
         m_hostname = host;
@@ -73,6 +79,7 @@ public final class Client {
         DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
         decoder.addMessageDecoder(new ConnAckDecoder());
         decoder.addMessageDecoder(new SubAckDecoder());
+        decoder.addMessageDecoder(new PublishDecoder());
 
         DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
         encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
@@ -209,6 +216,7 @@ public final class Client {
         }
         
         //TODO register the publishCallback in some registry to be notified
+        m_subscribersList.put(topic, publishCallback);
         
         //wait for the SubAck
         m_subscribeBarrier = new CountDownLatch(1);
@@ -247,6 +255,11 @@ public final class Client {
     }
     
     
+    protected void publishCallback(String topic, byte[] payload) {
+        m_subscribersList.get(topic).published(topic, payload);
+    }
+    
+    
     public void close() {
         //send the CLOSE message
         m_session.write(new DisconnectMessage());

parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java

@@ -6,6 +6,7 @@ import org.apache.mina.core.session.IoSession;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
 import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
+import org.dna.mqtt.moquette.proto.messages.PublishMessage;
 import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
 
 /**
@@ -36,9 +37,9 @@ public class ClientMQTTHandler extends IoHandlerAdapter {
 //            case SUBSCRIBE:
 //                handleSubscribe(session, (SubscribeMessage) msg);
 //        break;
-//            case PUBLISH:
-//                handlePublish(session, (PublishMessage) msg);
-//        break;
+            case PUBLISH:
+                handlePublish(session, (PublishMessage) msg);
+                break;
         }
     }
 
@@ -49,4 +50,8 @@ public class ClientMQTTHandler extends IoHandlerAdapter {
     private void handleSubscribeAck(IoSession session, SubAckMessage subAckMessage) {
         m_callback.subscribeAckCallback();
     }
+    
+    private void handlePublish(IoSession session, PublishMessage pubMessage) {
+        m_callback.publishCallback(pubMessage.getTopicName(), pubMessage.getPayload());
+    }
 }

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

@@ -15,9 +15,11 @@ import org.dna.mqtt.moquette.proto.ConnAckEncoder;
 import org.dna.mqtt.moquette.proto.ConnectDecoder;
 import org.dna.mqtt.moquette.proto.DisconnectDecoder;
 import org.dna.mqtt.moquette.proto.PublishDecoder;
+import org.dna.mqtt.moquette.proto.PublishEncoder;
 import org.dna.mqtt.moquette.proto.SubAckEncoder;
 import org.dna.mqtt.moquette.proto.SubscribeDecoder;
 import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
+import org.dna.mqtt.moquette.proto.messages.PublishMessage;
 import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
 /**
  * Launch a  configured version of the server.
@@ -29,6 +31,7 @@ public class Server {
     
     public static final int PORT = 9191;
     public static final int DEFAULT_CONNECT_TIMEOUT = 10;
+    private IoAcceptor m_acceptor;
     
     public static void main(String[] args) throws IOException {
         new Server().startServer();
@@ -46,11 +49,12 @@ public class Server {
 //        encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
         encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder());
         encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder());
+        encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
         
-        IoAcceptor acceptor = new NioSocketAcceptor();
+        m_acceptor = new NioSocketAcceptor();
 
-        acceptor.getFilterChain().addLast( "logger", new LoggingFilter("SERVER LOG") );
-        acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
+        m_acceptor.getFilterChain().addLast( "logger", new LoggingFilter("SERVER LOG") );
+        m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
 
         MQTTHandler handler = new MQTTHandler();
         SimpleMessaging messaging = new SimpleMessaging();
@@ -59,12 +63,17 @@ public class Server {
         messaging.setNotifier(handler);
         
         
-        acceptor.setHandler(handler);
-        ((NioSocketAcceptor)acceptor).getSessionConfig().setReuseAddress(true);
-        acceptor.getSessionConfig().setReadBufferSize( 2048 );
-        acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT );
-        acceptor.bind( new InetSocketAddress(PORT) );
+        m_acceptor.setHandler(handler);
+        ((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true);
+        m_acceptor.getSessionConfig().setReadBufferSize( 2048 );
+        m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT );
+        m_acceptor.bind( new InetSocketAddress(PORT) );
         LOG.info("Server binded");
     }
     
+    protected void stopServer() {
+        m_acceptor.unbind();
+        LOG.info("Server unbinded");
+    }
+    
 }