add reconnect api

Author: koush

AndroidAsync/src/com/koushikdutta/async/future/Future.java

@@ -8,4 +8,5 @@ public interface Future<T> extends Cancellable, java.util.concurrent.Future<T> {
      * @return
      */
     public Future<T> setCallback(FutureCallback<T> callback);
+    public <C extends TransformFuture<?, T>> C then(C callback);
 }

AndroidAsync/src/com/koushikdutta/async/future/SimpleFuture.java

@@ -151,6 +151,13 @@ public SimpleFuture<T> setCallback(FutureCallback<T> callback) {
         return this;
     }
 
+    @Override
+    public <C extends TransformFuture<?, T>> C then(C callback) {
+        callback.setParent(this);
+        setCallback(callback);
+        return callback;
+    }
+
     @Override
     public SimpleFuture<T> setParent(Cancellable parent) {
         super.setParent(parent);

AndroidAsync/src/com/koushikdutta/async/http/socketio/SocketIOClient.java

@@ -174,6 +174,10 @@ public void of(String endpoint, ConnectCallback connectCallback) {
         connection.connect(new SocketIOClient(connection, endpoint, connectCallback));
     }
 
+    public void reconnect() {
+        connection.reconnect(null);
+    }
+
     public WebSocket getWebSocket() {
         return connection.webSocket;
     }

AndroidAsync/src/com/koushikdutta/async/http/socketio/SocketIOConnection.java

@@ -7,6 +7,8 @@
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.future.Cancellable;
 import com.koushikdutta.async.future.DependentCancellable;
+import com.koushikdutta.async.future.FutureCallback;
+import com.koushikdutta.async.future.TransformFuture;
 import com.koushikdutta.async.http.AsyncHttpClient;
 import com.koushikdutta.async.http.AsyncHttpResponse;
 import com.koushikdutta.async.http.WebSocket;
@@ -85,66 +87,63 @@ public void disconnect(SocketIOClient client) {
         webSocket = null;
     }
 
+    Cancellable connecting;
     void reconnect(final DependentCancellable child) {
         if (isConnected()) {
             return;
         }
 
+        // if a connection is in progress, just wait.
+        if (connecting != null && !connecting.isDone() && !connecting.isCancelled()) {
+            if (child != null)
+                child.setParent(connecting);
+            return;
+        }
+
         request.logi("Reconnecting socket.io");
 
         // dont invoke onto main handler, as it is unnecessary until a session is ready or failed
         request.setHandler(null);
-        // initiate a session
-        Cancellable cancel = httpClient.executeString(request, new AsyncHttpClient.StringCallback() {
+
+        Cancellable connecting = httpClient.executeString(request)
+        .then(new TransformFuture<WebSocket, String>() {
             @Override
-            public void onCompleted(final Exception e, AsyncHttpResponse response, String result) {
-                request.logi("socket.io session received");
+            protected void transform(String result) throws Exception {
+                String[] parts = result.split(":");
+                String session = parts[0];
+                if (!"".equals(parts[1]))
+                    heartbeat = Integer.parseInt(parts[1]) / 2 * 1000;
+                else
+                    heartbeat = 0;
+
+                String transportsLine = parts[3];
+                String[] transports = transportsLine.split(",");
+                HashSet<String> set = new HashSet<String>(Arrays.asList(transports));
+                if (!set.contains("websocket"))
+                    throw new Exception("websocket not supported");
+
+                final String sessionUrl = request.getUri().toString() + "websocket/" + session + "/";
+
+                httpClient.websocket(sessionUrl, null, null)
+                .setCallback(getCompletionCallback());
+            }
+        })
+        .setCallback(new FutureCallback<WebSocket>() {
+            @Override
+            public void onCompleted(Exception e, WebSocket result) {
                 if (e != null) {
                     reportDisconnect(e);
                     return;
                 }
 
-                try {
-                    String[] parts = result.split(":");
-                    String session = parts[0];
-                    if (!"".equals(parts[1]))
-                        heartbeat = Integer.parseInt(parts[1]) / 2 * 1000;
-                    else
-                        heartbeat = 0;
-
-                    String transportsLine = parts[3];
-                    String[] transports = transportsLine.split(",");
-                    HashSet<String> set = new HashSet<String>(Arrays.asList(transports));
-                    if (!set.contains("websocket"))
-                        throw new Exception("websocket not supported");
-
-                    final String sessionUrl = request.getUri().toString() + "websocket/" + session + "/";
-
-                    Cancellable cancel = httpClient.websocket(sessionUrl, null, new AsyncHttpClient.WebSocketConnectCallback() {
-                        @Override
-                        public void onCompleted(Exception ex, WebSocket webSocket) {
-                            if (ex != null) {
-                                reportDisconnect(ex);
-                                return;
-                            }
-
-                            reconnectDelay = 1000L;
-                            SocketIOConnection.this.webSocket = webSocket;
-                            attach();
-                        }
-                    });
-
-                    if (child != null)
-                        child.setParent(cancel);
-                }
-                catch (Exception ex) {
-                    reportDisconnect(ex);
-                }
+                reconnectDelay = 1000L;
+                SocketIOConnection.this.webSocket = result;
+                attach();
             }
         });
 
         if (child != null)
-            child.setParent(cancel);
+            child.setParent(connecting);
     }
 
     void setupHeartbeat() {