diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java index 2f71d29d..dfbbe297 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java +++ b/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 m_subscribersList = new HashMap(); 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()); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java index 460ea3df..ac5edb6e 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java +++ b/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()); + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index cb74e75c..a4f354f5 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/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"); + } + } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index 1a02890e..8890ae09 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -4,7 +4,10 @@ import java.io.IOException; import org.apache.mina.core.service.IoConnector; import org.dna.mqtt.moquette.client.Client; import org.dna.mqtt.moquette.client.IPublishCallback; +import org.junit.After; import org.junit.Test; +import static org.junit.Assert.*; +import org.junit.Before; /** * @@ -13,26 +16,46 @@ import org.junit.Test; public class ServerIntegrationTest { IoConnector connector; + boolean received; + Server server; protected void startServer() throws IOException { - new Server().startServer(); + server = new Server(); + server.startServer(); } + @Before + public void setUp() throws IOException { + startServer(); + } + + @After + public void tearDown() { + server.stopServer(); + } + + @Test public void testSubscribe() throws IOException, InterruptedException { - startServer(); +// startServer(); Client client = new Client("localhost", Server.PORT); client.connect(); - client.publish("/topic", "Test my payload".getBytes()); + client.subscribe("/topic", new IPublishCallback() { public void published(String topic, byte[] message) { - throw new UnsupportedOperationException("Not supported yet."); + received = true; } }); + client.publish("/topic", "Test my payload".getBytes()); + + client.close(); + + + assertTrue(received); } }