diff --git a/parser/src/main/java/org/dna/mqtt/moquette/PublishException.java b/parser/src/main/java/org/dna/mqtt/moquette/PublishException.java new file mode 100644 index 00000000..7fbde985 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/PublishException.java @@ -0,0 +1,16 @@ +package org.dna.mqtt.moquette; + +/** + * + * @author andrea + */ +public class PublishException extends ConnectionException { + + public PublishException(String msg) { + super(msg); + } + + public PublishException(Throwable e) { + super(e); + } +} 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 34a63e6f..1e3e9cc1 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 @@ -7,6 +7,7 @@ import java.util.logging.Level; import java.util.logging.Logger; import org.apache.mina.core.RuntimeIoException; import org.apache.mina.core.future.ConnectFuture; +import org.apache.mina.core.future.WriteFuture; import org.apache.mina.core.service.IoConnector; import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; @@ -17,12 +18,14 @@ 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; +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.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.DisconnectMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.server.Server; /** @@ -79,7 +82,7 @@ public final class Client { public void connect() throws MQTTException { int retries = 0; - //TODO perhapsh by proto deinifinition should be an implicit retry policy + //TODO perhaps by proto deinifinition should be an implicit retry policy for (; retries < m_connectRetries; retries++) { try { ConnectFuture future = m_connector.connect(new InetSocketAddress(m_hostname, m_port)); @@ -148,6 +151,29 @@ public final class Client { } } + /** + * Publish to the connected server the payload message to the given topic. + * It's admitted to publish a 0 -length payload. + */ + public void publish(String topic, byte[] payload) throws PublishException{ + PublishMessage msg = new PublishMessage(); + msg.setTopicName(topic); + msg.setPayload(payload); + + WriteFuture wf = m_session.write(msg); + try { + wf.await(); + } catch (InterruptedException ex) { + LOG.log(Level.SEVERE, null, ex); + throw new PublishException(ex); + } + + Throwable ex = wf.getException(); + if (ex != null) { + throw new PublishException(ex); + } + } + /** * TODO extract this SPI method in a SPI 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 98d066ff..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(); - //TODO client.publish("/topic", what) + client.publish("/topic", "Test my payload".getBytes()); //TODO client.subscribe("/topic", callback) client.close();