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 0022d3e4..f412cbe7 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 @@ -22,6 +22,8 @@ import org.dna.mqtt.moquette.PublishException; 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.PublishEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.DisconnectMessage; @@ -42,7 +44,7 @@ public final class Client { private static final Logger LOG = Logger.getLogger(Client.class.getName()); // private static final String HOSTNAME = /*"localhost"*/ "127.0.0.1"; // private static final int PORT = Server.PORT; - private static final long CONNECT_TIMEOUT = 30 * 1000L; // 30 seconds + private static final long CONNECT_TIMEOUT = 3 * 1000L; // 30 seconds final static int DEFAULT_RETRIES = 3; private int m_connectRetries = DEFAULT_RETRIES; @@ -67,6 +69,7 @@ public final class Client { DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); + encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder()); m_connector = new NioSocketConnector(); @@ -160,6 +163,9 @@ public final class Client { msg.setTopicName(topic); msg.setPayload(payload); + //Untill the server could handle all the Qos levels + msg.setQos(AbstractMessage.QOSType.MOST_ONE); + WriteFuture wf = m_session.write(msg); try { wf.await(); 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 2647b2ee..66289cf3 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 @@ -12,6 +12,7 @@ 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.PublishDecoder; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; /** * Launch a configured version of the server. @@ -32,6 +33,7 @@ public class Server { protected void startServer() throws IOException { DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); decoder.addMessageDecoder(new ConnectDecoder()); + decoder.addMessageDecoder(new PublishDecoder()); decoder.addMessageDecoder(new DisconnectDecoder()); DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); 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 95e9583e..6bf26994 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();