diff --git a/parser/src/main/java/org/dna/mqtt/moquette/ConnectionException.java b/parser/src/main/java/org/dna/mqtt/moquette/ConnectionException.java new file mode 100644 index 00000000..3ceecde9 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/ConnectionException.java @@ -0,0 +1,16 @@ +package org.dna.mqtt.moquette; + +/** + * + * @author andrea + */ +public class ConnectionException extends MQTTException { + + public ConnectionException(String msg) { + super(msg); + } + + public ConnectionException(Throwable e) { + super(e); + } +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/MQTTException.java b/parser/src/main/java/org/dna/mqtt/moquette/MQTTException.java new file mode 100644 index 00000000..23e41ba4 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/MQTTException.java @@ -0,0 +1,24 @@ +package org.dna.mqtt.moquette; + +/** + * + * @author andrea + */ +public class MQTTException extends RuntimeException { + + public MQTTException() { + super(); + } + + public MQTTException(String msg) { + super(msg); + } + + public MQTTException(Throwable cause) { + super(cause); + } + + public MQTTException(String msg, Throwable cause) { + super(msg, cause); + } +} 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 7a0b5c54..336c9bf5 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,75 +1,162 @@ package org.dna.mqtt.moquette.client; import java.net.InetSocketAddress; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +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.service.IoConnector; +import org.apache.mina.core.session.IdleStatus; 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; import org.dna.mqtt.moquette.proto.ConnAckDecoder; -import org.dna.mqtt.moquette.proto.ConnectDecoder; import org.dna.mqtt.moquette.proto.ConnectEncoder; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.server.Server; /** - * Simple client to connect to a server + * The API to connect to a MQTT server. * + * Pass to the constructor the host and port to connect, invoke the connect + * message, invoke publish to send notifications messages, invoke subscribe + * with a callback to be notified when something happen on the desired topic. * * @author andrea */ -public class Client { - +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 String HOSTNAME = /*"localhost"*/ "127.0.0.1"; +// private static final int PORT = Server.PORT; private static final long CONNECT_TIMEOUT = 30 * 1000L; // 30 seconds + final static int DEFAULT_RETRIES = 3; + private int m_connectRetries = DEFAULT_RETRIES; + + private String m_hostname; + private int m_port; + + //internal mangement used for conversation with the server + private IoConnector m_connector; + private IoSession m_session; + private CountDownLatch m_connectBarrier; + private byte m_returnCode; - public static void main(String[] args) throws Throwable { + public Client(String host, int port) { + m_hostname = host; + m_port = port; + init(); + } + + protected void init() { DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); decoder.addMessageDecoder(new ConnAckDecoder()); - + DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); - - - NioSocketConnector connector = new NioSocketConnector(); - // Configure the service. - connector.setConnectTimeoutMillis(CONNECT_TIMEOUT); - connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder)); - connector.getFilterChain().addLast("logger", new LoggingFilter()); - connector.setHandler(new ClientMQTTHandler()); + m_connector = new NioSocketConnector(); - IoSession session; - for (;;) { + m_connector.getFilterChain().addLast("logger", new LoggingFilter()); + m_connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder)); + + m_connector.setHandler(new ClientMQTTHandler(this)); + m_connector.getSessionConfig().setReadBufferSize(2048); + m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Server.DEFAULT_CONNECT_TIMEOUT); + } + + public void connect() throws MQTTException { + int retries = 0; + + //TODO perhapsh by proto deinifinition should be an implicit retry policy + for (; retries < m_connectRetries; retries++) { try { - ConnectFuture future = connector.connect(new InetSocketAddress(HOSTNAME, PORT)); - LOG.info("Client waiting to connect to server"); + ConnectFuture future = m_connector.connect(new InetSocketAddress(m_hostname, m_port)); + LOG.fine("Client waiting to connect to server"); future.awaitUninterruptibly(); - session = future.getSession(); + m_session = future.getSession(); break; } catch (RuntimeIoException e) { - System.err.println("Failed to connect."); - e.printStackTrace(); - Thread.sleep(5000); + LOG.log(Level.FINE, "Failed to connect, retry " + retries + " of ("+m_connectRetries+")", e); + } + try { + Thread.sleep(1000); + } catch (InterruptedException ex) { + LOG.log(Level.SEVERE, "Error while sleeping for retry", ex); } } + if (retries == m_connectRetries) { + throw new MQTTException("Can't connect to the server after " + retries + "retries"); + } + + m_connectBarrier = new CountDownLatch(1); + //send a message over the session ConnectMessage connMsg = new ConnectMessage(); connMsg.setKeepAlive(3); - session.write(connMsg); - LOG.info("Client wrote message to server"); - - // wait until the summation is done - session.getCloseFuture().awaitUninterruptibly(); - connector.dispose(); + connMsg.setClientID("FAKE_MSG_ID1"); //TODO create a generator for the ID + m_session.write(connMsg); + + //suspend until the server respond with CONN_ACK + boolean unlocked = false; + try { + unlocked = m_connectBarrier.await(1000, TimeUnit.MILLISECONDS); //TODO parametrize + } catch (InterruptedException ex) { + throw new ConnectionException(ex); + } + + //if not arrive into certain limit, raise an error + if (!unlocked) { + throw new ConnectionException("Connection timeout elapsed unless server responded with a CONN_ACK"); + } + + //also raise an error when CONN_ACK is received with some error code inside + if (m_returnCode != ConnAckMessage.CONNECTION_ACCEPTED) { + String errMsg; + switch (m_returnCode) { + case ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION: + errMsg = "Unacceptable protocol version"; + break; + case ConnAckMessage.IDENTIFIER_REJECTED: + errMsg = "Identifier rejected"; + break; + case ConnAckMessage.SERVER_UNAVAILABLE: + errMsg = "Server unavailable"; + break; + case ConnAckMessage.BAD_USERNAME_OR_PASSWORD: + errMsg = "Bad username or password"; + break; + case ConnAckMessage.NOT_AUTHORIZED: + errMsg = "Not authorized"; + break; + default: + errMsg = "Not idetified erro code " + m_returnCode; + } + throw new ConnectionException(errMsg); + } } + protected void connectionAckCallback(byte returnCode) { + m_returnCode = returnCode; + m_connectBarrier.countDown(); + } + + public void close() { + //TODO send the CLOSE message + + + // wait until the summation is done + m_session.getCloseFuture().awaitUninterruptibly(); + m_connector.dispose(); + } + } 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 12f557a5..bcc0bc76 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,6 +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.*; /** * @@ -12,21 +14,31 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage; public class ClientMQTTHandler extends IoHandlerAdapter { private static final Logger LOG = Logger.getLogger(ClientMQTTHandler.class.getName()); + + Client m_callback; + ClientMQTTHandler(Client callback) { + m_callback = callback; + } + @Override public void messageReceived(IoSession session, Object message) throws Exception { AbstractMessage msg = (AbstractMessage) message; LOG.fine("Received a message of type " + msg.getMessageType()); -// switch (msg.getMessageType()) { -// case CONNECT: -// handleConnect(session, (ConnectMessage) msg); -// break; + switch (msg.getMessageType()) { + case CONNACK: + handleConnectAck(session, (ConnAckMessage) msg); + break; // case SUBSCRIBE: // handleSubscribe(session, (SubscribeMessage) msg); // break; // case PUBLISH: // handlePublish(session, (PublishMessage) msg); // break; -// } + } + } + + private void handleConnectAck(IoSession session, ConnAckMessage connAckMessage) { + m_callback.connectionAckCallback(connAckMessage.getReturnCode()); } } 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 new file mode 100644 index 00000000..98d066ff --- /dev/null +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -0,0 +1,32 @@ +package org.dna.mqtt.moquette.server; + +import java.io.IOException; +import org.apache.mina.core.service.IoConnector; +import org.dna.mqtt.moquette.client.Client; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class ServerIntegrationTest { + + IoConnector connector; + + protected void startServer() throws IOException { + new Server().startServer(); + } + + @Test + public void testSubscribe() throws IOException, InterruptedException { + startServer(); + Client client = new Client("localhost", Server.PORT); + client.connect(); + + //TODO client.publish("/topic", what) + //TODO client.subscribe("/topic", callback) + + client.close(); + } + +}