Merge pull request koush#268 from nebulist/upstream-pullrequest

koush · koush · commit 192e1a8f6f8d · 2014-10-24T17:13:36.000-07:00
Socket.io configurable reconnect delay, reconnect bugfix, optional randomized reconnect delay, &amp; named/minimized synchronousWorkers threads
diff --git a/AndroidAsync/src/com/koushikdutta/async/AsyncServer.java b/AndroidAsync/src/com/koushikdutta/async/AsyncServer.java
@@ -31,7 +31,12 @@
 import java.util.WeakHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class AsyncServer {
     public static final String LOGTAG = "NIO";
@@ -85,7 +90,7 @@ public static void post(Handler handler, Runnable runnable) {
         catch (Throwable ex) {
         }
     }
-    
+
     static AsyncServer mInstance = new AsyncServer();
     public static AsyncServer getDefault() {
         return mInstance;
@@ -114,7 +119,7 @@ private void handleSocket(final AsyncNetworkSocket handler) throws ClosedChannel
         ckey.attach(handler);
         handler.setup(this, ckey);
     }
-    
+
     public void removeAllCallbacks(Object scheduled) {
         synchronized (this) {
             mQueue.remove(scheduled);
@@ -134,7 +139,7 @@ public void run() {
             }
         });
     }
-    
+
     public Object postDelayed(Runnable runnable, long delay) {
         Scheduled s;
         synchronized (this) {
@@ -161,11 +166,11 @@ public Object postDelayed(Runnable runnable, long delay) {
         }
         return s;
     }
-    
+
     public Object post(Runnable runnable) {
         return postDelayed(runnable, 0);
     }
-    
+
     public Object post(final CompletedCallback callback, final Exception e) {
         return post(new Runnable() {
             @Override
@@ -174,7 +179,7 @@ public void run() {
             }
         });
     }
-    
+
     public void run(final Runnable runnable) {
         if (Thread.currentThread() == mAffinity) {
             post(runnable);
@@ -263,7 +268,7 @@ public void run() {
         catch (Exception e) {
         }
     }
-    
+
     protected void onDataReceived(int transmitted) {
     }
 
@@ -335,7 +340,7 @@ protected void cancelCleanup() {
         SocketChannel socket;
         ConnectCallback callback;
     }
-    
+
     private ConnectFuture connectResolvedInetSocketAddress(final InetSocketAddress address, final ConnectCallback callback) {
         final ConnectFuture cancel = new ConnectFuture();
         assert !address.isUnresolved();
@@ -396,7 +401,14 @@ public Cancellable connectSocket(final String host, final int port, final Connec
         return connectSocket(InetSocketAddress.createUnresolved(host, port), callback);
     }
 
-    private static ExecutorService synchronousWorkers = Executors.newFixedThreadPool(4);
+    private static ExecutorService newSynchronousWorkers() {
+        ThreadFactory tf = new NamedThreadFactory("AsyncServer-worker-");
+        ThreadPoolExecutor tpe = new ThreadPoolExecutor(1, 4, 10L,
+            TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), tf);
+        return tpe;
+    }
+
+    private static ExecutorService synchronousWorkers = newSynchronousWorkers();
     public Future<InetAddress[]> getAllByName(final String host) {
         final SimpleFuture<InetAddress[]> ret = new SimpleFuture<InetAddress[]>();
         synchronousWorkers.execute(new Runnable() {
@@ -487,7 +499,7 @@ public void run() {
         });
         return handler;
     }
-    
+
     public AsyncDatagramSocket connectDatagram(final SocketAddress remote) throws IOException {
         final DatagramChannel socket = DatagramChannel.open();
         final AsyncDatagramSocket handler = new AsyncDatagramSocket();
@@ -509,7 +521,7 @@ public void run() {
         });
         return handler;
     }
-    
+
     final static WeakHashMap<Thread, AsyncServer> mServers = new WeakHashMap<Thread, AsyncServer>();
 
     private boolean addMe() {
@@ -527,7 +539,7 @@ private boolean addMe() {
     public static AsyncServer getCurrentThreadServer() {
         return mServers.get(Thread.currentThread());
     }
-    
+
     Thread mAffinity;
     private void run(boolean newThread) {
         final SelectorWrapper selector;
@@ -596,10 +608,10 @@ public void run() {
             }
             return;
         }
-        
+
         run(this, selector, queue);
     }
-    
+
     private static void run(final AsyncServer server, final SelectorWrapper selector, final PriorityQueue<Scheduled> queue) {
 //        Log.i(LOGTAG, "****AsyncServer is starting.****");
         // at this point, this local queue and selector are owned
@@ -666,11 +678,11 @@ private static void shutdownEverything(SelectorWrapper selector) {
         catch (Exception e) {
         }
     }
-    
+
     private static final long QUEUE_EMPTY = Long.MAX_VALUE;
     private static long lockAndRunQueue(final AsyncServer server, final PriorityQueue<Scheduled> queue) {
         long wait = QUEUE_EMPTY;
-        
+
         // find the first item we can actually run
         while (true) {
             Scheduled run = null;
@@ -689,10 +701,10 @@ private static long lockAndRunQueue(final AsyncServer server, final PriorityQueu
                     }
                 }
             }
-            
+
             if (run == null)
                 break;
-            
+
             run.runnable.run();
         }
 
@@ -834,11 +846,11 @@ public void run() {
             }
         });
     }
-    
+
     public Thread getAffinity() {
         return mAffinity;
     }
-    
+
     public boolean isAffinityThread() {
         return mAffinity == Thread.currentThread();
     }
@@ -847,4 +859,27 @@ public boolean isAffinityThreadOrStopped() {
         Thread affinity = mAffinity;
         return affinity == null || affinity == Thread.currentThread();
     }
+
+    private static class NamedThreadFactory implements ThreadFactory {
+        private final ThreadGroup group;
+        private final AtomicInteger threadNumber = new AtomicInteger(1);
+        private final String namePrefix;
+
+        NamedThreadFactory(String namePrefix) {
+            SecurityManager s = System.getSecurityManager();
+            group = (s != null) ? s.getThreadGroup() :
+                Thread.currentThread().getThreadGroup();
+            this.namePrefix = namePrefix;
+        }
+
+        public Thread newThread(Runnable r) {
+            Thread t = new Thread(group, r,
+                namePrefix + threadNumber.getAndIncrement(), 0);
+            if (t.isDaemon()) t.setDaemon(false);
+            if (t.getPriority() != Thread.NORM_PRIORITY) {
+                t.setPriority(Thread.NORM_PRIORITY);
+            }
+            return t;
+        }
+    }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/socketio/SocketIOConnection.java b/AndroidAsync/src/com/koushikdutta/async/http/socketio/SocketIOConnection.java
@@ -30,13 +30,15 @@
 class SocketIOConnection {
     AsyncHttpClient httpClient;
     int heartbeat;
+    long reconnectDelay;
     ArrayList<SocketIOClient> clients = new ArrayList<SocketIOClient>();
     SocketIOTransport transport;
     SocketIORequest request;
 
     public SocketIOConnection(AsyncHttpClient httpClient, SocketIORequest request) {
         this.httpClient = httpClient;
         this.request = request;
+        this.reconnectDelay = this.request.config.reconnectDelay;
     }
 
     public boolean isConnected() {
@@ -104,7 +106,7 @@ void reconnect(final DependentCancellable child) {
 
         request.logi("Reconnecting socket.io");
 
-        Cancellable connecting = httpClient.executeString(request, null)
+        connecting = httpClient.executeString(request, null)
         .then(new TransformFuture<SocketIOTransport, String>() {
             @Override
             protected void transform(String result) throws Exception {
@@ -157,7 +159,7 @@ public void onCompleted(Exception e, SocketIOTransport result) {
                     return;
                 }
 
-                reconnectDelay = 1000L;
+                reconnectDelay = request.config.reconnectDelay;
                 SocketIOConnection.this.transport = result;
                 attach();
             }
@@ -215,11 +217,23 @@ private void delayReconnect() {
             public void run() {
                 reconnect(null);
             }
-        }, reconnectDelay);
-        reconnectDelay *= 2;
+        }, nextReconnectDelay(reconnectDelay));
+
+        reconnectDelay = reconnectDelay * 2;
+        if (request.config.reconnectDelayMax > 0L) {
+            reconnectDelay = Math.min(reconnectDelay, request.config.reconnectDelayMax);
+        }
+    }
+
+    private long nextReconnectDelay(long targetDelay) {
+        if (targetDelay < 2L || targetDelay > (Long.MAX_VALUE >> 1) ||
+            !request.config.randomizeReconnectDelay)
+        {
+            return targetDelay;
+        }
+        return (targetDelay >> 1) + (long) (targetDelay * Math.random());
     }
 
-    long reconnectDelay = 1000L;
     private void reportDisconnect(final Exception ex) {
         if (ex != null) {
             request.loge("socket.io disconnected", ex);
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/socketio/SocketIORequest.java b/AndroidAsync/src/com/koushikdutta/async/http/socketio/SocketIORequest.java
@@ -1,7 +1,6 @@
 package com.koushikdutta.async.http.socketio;
 
 import android.net.Uri;
-import android.text.TextUtils;
 
 import com.koushikdutta.async.http.AsyncHttpPost;
 
@@ -10,6 +9,11 @@ public SocketIORequest(String uri) {
         this(uri, "");
     }
 
+    Config config;
+    public Config getConfig() {
+        return config;
+    }
+
     String endpoint;
     public String getEndpoint() {
         return endpoint;
@@ -25,8 +29,45 @@ public SocketIORequest(String uri, String endpoint) {
     }
 
     public SocketIORequest(String uri, String endpoint, String query) {
+        this(uri, endpoint, query, null);
+    }
+
+    public SocketIORequest(String uri, String endpoint, String query, Config config) {
         super(Uri.parse(uri + (query == null ? "" : "?" + query)).buildUpon().encodedPath("/socket.io/1/").build().toString());
+        this.config = (config != null) ? config : new Config();
         this.endpoint = endpoint;
         this.query = query;
     }
+
+    public static class Config {
+        boolean randomizeReconnectDelay = false;
+        public void setRandomizeReconnectDelay(boolean randomizeReconnectDelay) {
+            this.randomizeReconnectDelay = randomizeReconnectDelay;
+        }
+        public boolean isRandomizeReconnectDelay() {
+            return randomizeReconnectDelay;
+        }
+
+        long reconnectDelay = 1000L;
+        public void setReconnectDelay(long reconnectDelay) {
+            if (reconnectDelay < 0L) {
+                throw new IllegalArgumentException("reconnectDelay must be >= 0");
+            }
+            this.reconnectDelay = reconnectDelay;
+        }
+        public long getReconnectDelay() {
+            return reconnectDelay;
+        }
+
+        long reconnectDelayMax = 0L;
+        public void setReconnectDelayMax(long reconnectDelayMax) {
+            if (reconnectDelay < 0L) {
+                throw new IllegalArgumentException("reconnectDelayMax must be >= 0");
+            }
+            this.reconnectDelayMax = reconnectDelayMax;
+        }
+        public long getReconnectDelayMax() {
+            return reconnectDelayMax;
+        }
+    }
 }
