move NullDataCallback.

change how socket ownership works. fix possible bugs around that.
throw assertion error if connect callback is invoked twice. Switch to future?

15 files changed, 60 insertions, 44 deletions

diff --git a/AndroidAsync/src/com/koushikdutta/async/AsyncSSLSocketWrapper.java b/AndroidAsync/src/com/koushikdutta/async/AsyncSSLSocketWrapper.java
index 0f777e6..8760bdd 100644
--- a/AndroidAsync/src/com/koushikdutta/async/AsyncSSLSocketWrapper.java
+++ b/AndroidAsync/src/com/koushikdutta/async/AsyncSSLSocketWrapper.java
@@ -411,7 +411,7 @@ public class AsyncSSLSocketWrapper implements AsyncSocketWrapper, AsyncSSLSocket
         final HandshakeCallback hs = handshakeCallback;
         if (hs != null) {
             handshakeCallback = null;
-            mSocket.setDataCallback(new NullDataCallback());
+            mSocket.setDataCallback(new DataCallback.NullDataCallback());
             mSocket.end();
             mSocket.close();
             hs.onHandshakeCompleted(e, null);
diff --git a/AndroidAsync/src/com/koushikdutta/async/NullDataCallback.java b/AndroidAsync/src/com/koushikdutta/async/NullDataCallback.java
deleted file mode 100644
index 3f9be24..0000000
--- a/AndroidAsync/src/com/koushikdutta/async/NullDataCallback.java
+++ /dev/null
@@ -1,10 +0,0 @@
-package com.koushikdutta.async;
-
-import com.koushikdutta.async.callback.DataCallback;
-
-public class NullDataCallback implements DataCallback {
-    @Override
-    public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
-        bb.recycle();
-    }
-}
diff --git a/AndroidAsync/src/com/koushikdutta/async/Util.java b/AndroidAsync/src/com/koushikdutta/async/Util.java
index 5e26662..d9808ae 100644
--- a/AndroidAsync/src/com/koushikdutta/async/Util.java
+++ b/AndroidAsync/src/com/koushikdutta/async/Util.java
@@ -130,22 +130,30 @@ public class Util {
             }
         });
 
-        CompletedCallback wrapper = new CompletedCallback() {
+        final CompletedCallback wrapper = new CompletedCallback() {
             boolean reported;
             @Override
             public void onCompleted(Exception ex) {
                 if (reported)
                     return;
+                reported = true;
+                emitter.setDataCallback(null);
                 emitter.setEndCallback(null);
                 sink.setClosedCallback(null);
                 sink.setWriteableCallback(null);
-                reported = true;
                 callback.onCompleted(ex);
             }
         };
 
         emitter.setEndCallback(wrapper);
-        sink.setClosedCallback(wrapper);
+        sink.setClosedCallback(new CompletedCallback() {
+            @Override
+            public void onCompleted(Exception ex) {
+                if (ex == null)
+                    ex = new IOException("sink was closed before emitter ended");
+                wrapper.onCompleted(ex);
+            }
+        });
     }
     
     public static void stream(AsyncSocket s1, AsyncSocket s2, CompletedCallback callback) {
diff --git a/AndroidAsync/src/com/koushikdutta/async/callback/CompletedCallback.java b/AndroidAsync/src/com/koushikdutta/async/callback/CompletedCallback.java
index d6c0342..a5b4d64 100644
--- a/AndroidAsync/src/com/koushikdutta/async/callback/CompletedCallback.java
+++ b/AndroidAsync/src/com/koushikdutta/async/callback/CompletedCallback.java
@@ -1,5 +1,12 @@
 package com.koushikdutta.async.callback;
 
 public interface CompletedCallback {
+    public class NullCompletedCallback implements CompletedCallback {
+        @Override
+        public void onCompleted(Exception ex) {
+
+        }
+    }
+
     public void onCompleted(Exception ex);
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/callback/DataCallback.java b/AndroidAsync/src/com/koushikdutta/async/callback/DataCallback.java
index 564e48b..54da7ea 100644
--- a/AndroidAsync/src/com/koushikdutta/async/callback/DataCallback.java
+++ b/AndroidAsync/src/com/koushikdutta/async/callback/DataCallback.java
@@ -5,5 +5,12 @@ import com.koushikdutta.async.DataEmitter;
 
 
 public interface DataCallback {
+    public class NullDataCallback implements DataCallback {
+        @Override
+        public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
+            bb.recycle();
+        }
+    }
+
     public void onDataAvailable(DataEmitter emitter, ByteBufferList bb);
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
index 956fb26..943cee9 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
@@ -10,9 +10,9 @@ import com.koushikdutta.async.AsyncServer;
 import com.koushikdutta.async.AsyncSocket;
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
-import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.ConnectCallback;
+import com.koushikdutta.async.callback.DataCallback;
 import com.koushikdutta.async.future.Cancellable;
 import com.koushikdutta.async.future.Future;
 import com.koushikdutta.async.future.FutureCallback;
@@ -141,7 +141,7 @@ public class AsyncHttpClient {
                 return false;
 
             if (socket != null) {
-                socket.setDataCallback(new NullDataCallback());
+                socket.setDataCallback(new DataCallback.NullDataCallback());
                 socket.close();
             }
 
@@ -172,7 +172,7 @@ public class AsyncHttpClient {
 
         if (response != null) {
             // the request was cancelled, so close up shop, and eat any pending data
-            response.setDataCallback(new NullDataCallback());
+            response.setDataCallback(new DataCallback.NullDataCallback());
             response.close();
         }
     }
@@ -251,8 +251,19 @@ public class AsyncHttpClient {
 
         // 2) wait for a connect
         data.connectCallback = new ConnectCallback() {
+            boolean reported;
             @Override
             public void onConnectCompleted(Exception ex, AsyncSocket socket) {
+                if (reported) {
+                    if (socket != null) {
+                        socket.setDataCallback(new DataCallback.NullDataCallback());
+                        socket.setEndCallback(new CompletedCallback.NullCompletedCallback());
+                        socket.close();
+                        throw new AssertionError("double connect callback");
+                    }
+                }
+                reported = true;
+
                 request.logv("socket connected");
                 if (cancel.isCancelled()) {
                     if (socket != null)
@@ -264,14 +275,14 @@ public class AsyncHttpClient {
                 if (cancel.timeoutRunnable != null)
                     mServer.removeAllCallbacks(cancel.scheduled);
 
-                data.socket = socket;
-                cancel.socket = socket;
-
                 if (ex != null) {
                     reportConnectedCompleted(cancel, ex, null, request, callback);
                     return;
                 }
 
+                data.socket = socket;
+                cancel.socket = socket;
+
                 executeSocket(request, redirectCount, cancel, callback, data);
             }
         };
@@ -563,7 +574,7 @@ public class AsyncHttpClient {
             @Override
             public void cancelCleanup() {
                 try {
-                    cancel.get().setDataCallback(new NullDataCallback());
+                    cancel.get().setDataCallback(new DataCallback.NullDataCallback());
                     cancel.get().close();
                 }
                 catch (Exception e) {
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
index cd6ffff..c161f30 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
@@ -6,16 +6,10 @@ import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
 import com.koushikdutta.async.DataSink;
 import com.koushikdutta.async.FilteredDataEmitter;
-import com.koushikdutta.async.LineEmitter;
-import com.koushikdutta.async.LineEmitter.StringCallback;
-import com.koushikdutta.async.NullDataCallback;
-import com.koushikdutta.async.Util;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.WritableCallback;
 import com.koushikdutta.async.http.body.AsyncHttpRequestBody;
-import com.koushikdutta.async.http.filter.ChunkedOutputFilter;
 
-import java.io.IOException;
 import java.nio.charset.Charset;
 
 abstract class AsyncHttpResponseImpl extends FilteredDataEmitter implements AsyncSocket, AsyncHttpResponse, AsyncHttpClientMiddleware.ResponseHead {
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncSocketMiddleware.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncSocketMiddleware.java
index 35ce3dd..12dd28a 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncSocketMiddleware.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncSocketMiddleware.java
@@ -6,10 +6,10 @@ import com.koushikdutta.async.ArrayDeque;
 import com.koushikdutta.async.AsyncSocket;
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
-import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.ConnectCallback;
 import com.koushikdutta.async.callback.ContinuationCallback;
+import com.koushikdutta.async.callback.DataCallback;
 import com.koushikdutta.async.future.Cancellable;
 import com.koushikdutta.async.future.Continuation;
 import com.koushikdutta.async.future.SimpleCancellable;
@@ -127,6 +127,8 @@ public class AsyncSocketMiddleware extends SimpleMiddleware {
             return null;
         }
 
+        data.state.put("socket-owner", this);
+
         final String lookup = computeLookup(uri, port, data.request.getProxyHost(), data.request.getProxyPort());
         ConnectionInfo info = getOrCreateConnectionInfo(lookup);
         synchronized (AsyncSocketMiddleware.this) {
@@ -139,7 +141,6 @@ public class AsyncSocketMiddleware extends SimpleMiddleware {
 
             info.openCount++;
 
-            data.state.put(getClass().getCanonicalName() + ".owned", true);
 
             while (!info.sockets.isEmpty()) {
                 IdleSocketHolder idleSocketHolder = info.sockets.pop();
@@ -239,7 +240,7 @@ public class AsyncSocketMiddleware extends SimpleMiddleware {
                                     }
 
                                     if (setComplete(null, socket)) {
-                                        data.connectCallback.onConnectCompleted(ex, socket);
+                                        data.connectCallback.onConnectCompleted(null, socket);
                                     }
                                 }
                             }));
@@ -313,7 +314,7 @@ public class AsyncSocketMiddleware extends SimpleMiddleware {
         socket.setWriteableCallback(null);
         // should not get any data after this point...
         // if so, eat it and disconnect.
-        socket.setDataCallback(new NullDataCallback() {
+        socket.setDataCallback(new DataCallback.NullDataCallback() {
             @Override
             public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
                 super.onDataAvailable(emitter, bb);
@@ -346,9 +347,8 @@ public class AsyncSocketMiddleware extends SimpleMiddleware {
 
     @Override
     public void onResponseComplete(final OnResponseCompleteDataOnRequestSentData data) {
-        if (!data.state.get(getClass().getCanonicalName() + ".owned", false)) {
+        if (data.state.get("socket-owner") != this)
             return;
-        }
 
         try {
             idleSocket(data.socket);
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/body/MultipartFormDataBody.java b/AndroidAsync/src/com/koushikdutta/async/http/body/MultipartFormDataBody.java
index 4cf41f2..0a8a26d 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/body/MultipartFormDataBody.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/body/MultipartFormDataBody.java
@@ -5,7 +5,6 @@ import com.koushikdutta.async.DataEmitter;
 import com.koushikdutta.async.DataSink;
 import com.koushikdutta.async.LineEmitter;
 import com.koushikdutta.async.LineEmitter.StringCallback;
-import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.ContinuationCallback;
 import com.koushikdutta.async.callback.DataCallback;
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/cache/ResponseCacheMiddleware.java b/AndroidAsync/src/com/koushikdutta/async/http/cache/ResponseCacheMiddleware.java
index 40aa530..ae84173 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/cache/ResponseCacheMiddleware.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/cache/ResponseCacheMiddleware.java
@@ -173,7 +173,10 @@ public class ResponseCacheMiddleware extends SimpleMiddleware {
                 }
             });
             cacheHitCount++;
-            return new SimpleCancellable();
+            data.state.put("socket-owner", this);
+            SimpleCancellable ret = new SimpleCancellable();
+            ret.setComplete();
+            return ret;
         }
         else if (responseSource == ResponseSource.CONDITIONAL_CACHE) {
             data.request.logi("Response may be served from conditional cache");
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/filter/GZIPInputFilter.java b/AndroidAsync/src/com/koushikdutta/async/http/filter/GZIPInputFilter.java
index 83beef7..81b9c1d 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/filter/GZIPInputFilter.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/filter/GZIPInputFilter.java
@@ -2,7 +2,6 @@ package com.koushikdutta.async.http.filter;
 
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
-import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.PushParser;
 import com.koushikdutta.async.PushParser.ParseCallback;
 import com.koushikdutta.async.callback.DataCallback;
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
index 3d65669..c18c5d5 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
@@ -13,7 +13,6 @@ import com.koushikdutta.async.AsyncServerSocket;
 import com.koushikdutta.async.AsyncSocket;
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
-import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.Util;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.ListenCallback;
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/UnknownRequestBody.java b/AndroidAsync/src/com/koushikdutta/async/http/server/UnknownRequestBody.java
index 14a1424..a3fde6d 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/UnknownRequestBody.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/UnknownRequestBody.java
@@ -2,7 +2,6 @@ package com.koushikdutta.async.http.server;
 
 import com.koushikdutta.async.DataEmitter;
 import com.koushikdutta.async.DataSink;
-import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.Util;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.DataCallback;
@@ -64,6 +63,6 @@ public class UnknownRequestBody implements AsyncHttpRequestBody<Void> {
     public void parse(DataEmitter emitter, CompletedCallback completed) {
         this.emitter = emitter;
         emitter.setEndCallback(completed);
-        emitter.setDataCallback(new NullDataCallback());
+        emitter.setDataCallback(new DataCallback.NullDataCallback());
     }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/socketio/transport/WebSocketTransport.java b/AndroidAsync/src/com/koushikdutta/async/http/socketio/transport/WebSocketTransport.java
index 15928a3..5514ecd 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/socketio/transport/WebSocketTransport.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/socketio/transport/WebSocketTransport.java
@@ -1,8 +1,8 @@
 package com.koushikdutta.async.http.socketio.transport;
 
 import com.koushikdutta.async.AsyncServer;
-import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.callback.CompletedCallback;
+import com.koushikdutta.async.callback.DataCallback;
 import com.koushikdutta.async.http.WebSocket;
 
 public class WebSocketTransport implements SocketIOTransport {
@@ -12,7 +12,7 @@ public class WebSocketTransport implements SocketIOTransport {
     public WebSocketTransport(WebSocket webSocket) {
         this.webSocket = webSocket;
 
-        this.webSocket.setDataCallback(new NullDataCallback());
+        this.webSocket.setDataCallback(new DataCallback.NullDataCallback());
     }
 
     @Override
diff --git a/AndroidAsync/test/src/com/koushikdutta/async/test/ByteUtilTests.java b/AndroidAsync/test/src/com/koushikdutta/async/test/ByteUtilTests.java
index 0d2f5df..386c31a 100644
--- a/AndroidAsync/test/src/com/koushikdutta/async/test/ByteUtilTests.java
+++ b/AndroidAsync/test/src/com/koushikdutta/async/test/ByteUtilTests.java
@@ -2,10 +2,10 @@ package com.koushikdutta.async.test;
 
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.FilteredDataEmitter;
-import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.PushParser;
 import com.koushikdutta.async.TapCallback;
 import com.koushikdutta.async.Util;
+import com.koushikdutta.async.callback.DataCallback;
 
 import junit.framework.TestCase;
 
@@ -23,7 +23,7 @@ public class ByteUtilTests extends TestCase {
             }
         };
         new PushParser(mock)
-            .until((byte)0, new NullDataCallback())
+            .until((byte)0, new DataCallback.NullDataCallback())
             .readInt(new PushParser.ParseCallback<Integer>() {
                 public void parsed(Integer arg) {
                     valRead = arg;
@@ -43,7 +43,7 @@ public class ByteUtilTests extends TestCase {
             }
         };
         new PushParser(mock)
-                .until((byte)0, new NullDataCallback())
+                .until((byte)0, new DataCallback.NullDataCallback())
                 .readInt()
                 .tap(new TapCallback() {
                     public void parsed(int arg) {