From be9d70855f49ee06e05dfb259179724ee00d913d Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Tue, 24 Jul 2012 22:44:21 +0200 Subject: [PATCH] Fixed unit test for connect-close --- .../src/main/java/org/dna/mqtt/moquette/client/Client.java | 3 ++- .../java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java | 4 ++-- .../src/main/java/org/dna/mqtt/moquette/server/Server.java | 5 +++-- .../org/dna/mqtt/moquette/server/ServerIntegrationTest.java | 2 +- 4 files changed, 8 insertions(+), 6 deletions(-) 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 1e3e9cc1..0022d3e4 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 @@ -115,7 +115,7 @@ public final class Client { //suspend until the server respond with CONN_ACK boolean unlocked = false; try { - unlocked = m_connectBarrier.await(1000, TimeUnit.MILLISECONDS); //TODO parametrize + unlocked = m_connectBarrier.await(CONNECT_TIMEOUT, TimeUnit.MILLISECONDS); //TODO parametrize } catch (InterruptedException ex) { throw new ConnectionException(ex); } @@ -179,6 +179,7 @@ public final class Client { * TODO extract this SPI method in a SPI */ protected void connectionAckCallback(byte returnCode) { + LOG.info("connectionAckCallback invoked"); m_returnCode = returnCode; m_connectBarrier.countDown(); } 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 bcc0bc76..b84648d9 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 @@ -4,8 +4,8 @@ import java.util.logging.Logger; import org.apache.mina.core.service.IoHandlerAdapter; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; /** * @@ -28,7 +28,7 @@ public class ClientMQTTHandler extends IoHandlerAdapter { switch (msg.getMessageType()) { case CONNACK: handleConnectAck(session, (ConnAckMessage) msg); - break; + break; // case SUBSCRIBE: // handleSubscribe(session, (SubscribeMessage) msg); // break; 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 370e34d7..2647b2ee 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 @@ -8,10 +8,10 @@ import org.apache.mina.core.session.IdleStatus; 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.NioSocketAcceptor; 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.messages.ConnAckMessage; /** * Launch a configured version of the server. @@ -32,6 +32,7 @@ public class Server { protected void startServer() throws IOException { DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); decoder.addMessageDecoder(new ConnectDecoder()); + decoder.addMessageDecoder(new DisconnectDecoder()); DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); // encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); @@ -39,7 +40,7 @@ public class Server { IoAcceptor acceptor = new NioSocketAcceptor(); - acceptor.getFilterChain().addLast( "logger", new LoggingFilter() ); +// acceptor.getFilterChain().addLast( "logger", new LoggingFilter("SERVER LOG") ); acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder)); acceptor.setHandler( new MQTTHandler() ); 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 6bf26994..95e9583e 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 @@ -23,7 +23,7 @@ public class ServerIntegrationTest { Client client = new Client("localhost", Server.PORT); client.connect(); - client.publish("/topic", "Test my payload".getBytes()); +// client.publish("/topic", "Test my payload".getBytes()); //TODO client.subscribe("/topic", callback) client.close();