Implemented a Client API blocking connect

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);
+    }
+}

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);
+    }
+}

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 String HOSTNAME = /*"localhost"*/ "127.0.0.1";
+//    private static final int PORT = Server.PORT;
-    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;
     
-    public static void main(String[] args) throws Throwable {
+    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 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());
 
+        m_connector = new NioSocketConnector();
 
-        NioSocketConnector connector = new NioSocketConnector();
+        m_connector.getFilterChain().addLast("logger", new LoggingFilter());
+        m_connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder));
 
-        // Configure the service.
+        m_connector.setHandler(new ClientMQTTHandler(this));
-        connector.setConnectTimeoutMillis(CONNECT_TIMEOUT);
+        m_connector.getSessionConfig().setReadBufferSize(2048);
-        connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder));
+        m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Server.DEFAULT_CONNECT_TIMEOUT);
-        connector.getFilterChain().addLast("logger", new LoggingFilter());
+    }
-        connector.setHandler(new  ClientMQTTHandler());
 
-        IoSession session;
+    public void connect() throws MQTTException {
-        for (;;) {
+        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));
+                ConnectFuture future = m_connector.connect(new InetSocketAddress(m_hostname, m_port));
-                LOG.info("Client waiting to connect to server");
+                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.");
+                LOG.log(Level.FINE, "Failed to connect, retry " + retries + " of ("+m_connectRetries+")", e);
-                e.printStackTrace();
+            }
-                Thread.sleep(5000);
+            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);
+        connMsg.setClientID("FAKE_MSG_ID1"); //TODO create a generator for the ID
-        LOG.info("Client wrote message to server");
+        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
-        session.getCloseFuture().awaitUninterruptibly();
+        m_session.getCloseFuture().awaitUninterruptibly();
-        connector.dispose();
+        m_connector.dispose();
     }
 
 }

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.*;
 
 /**
  *
@@ -13,20 +15,30 @@ 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()) {
+        switch (msg.getMessageType()) {
-//            case CONNECT:
+            case CONNACK:
-//                handleConnect(session, (ConnectMessage) msg);
+                handleConnectAck(session, (ConnAckMessage) msg);
-//        break;
+        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());
     }
 }

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();
+    }
+    
+}