Refactor ExceptionCallback into CompletedCallback. They were the same.

Add pause and resume to DataEmitter.
Add simple support for HTTP pipelining. Pause the socket. This will allow requests to be buffered to whatever the underlying OS buffer is.

22 files changed, 199 insertions, 116 deletions

diff --git a/AndroidAsync/src/com/koushikdutta/async/AsyncServer.java b/AndroidAsync/src/com/koushikdutta/async/AsyncServer.java
index ef77748..e077566 100644
--- a/AndroidAsync/src/com/koushikdutta/async/AsyncServer.java
+++ b/AndroidAsync/src/com/koushikdutta/async/AsyncServer.java
@@ -164,7 +164,7 @@ public class AsyncServer {
                     });
                 }
                 catch (Exception e) {
-                    handler.onException(e);
+                    handler.onCompleted(e);
                     e.printStackTrace();
                 }
             }
diff --git a/AndroidAsync/src/com/koushikdutta/async/AsyncSocket.java b/AndroidAsync/src/com/koushikdutta/async/AsyncSocket.java
index 521eb4d..48a422e 100644
--- a/AndroidAsync/src/com/koushikdutta/async/AsyncSocket.java
+++ b/AndroidAsync/src/com/koushikdutta/async/AsyncSocket.java
@@ -1,7 +1,7 @@
 package com.koushikdutta.async;
 
 
-public interface AsyncSocket extends DataExchange, CloseableData, ExceptionEmitter {
+public interface AsyncSocket extends DataExchange, CloseableData, CompletedEmitter {
     public boolean isConnected();
     public void pause();
     public void resume();
diff --git a/AndroidAsync/src/com/koushikdutta/async/AsyncSocketImpl.java b/AndroidAsync/src/com/koushikdutta/async/AsyncSocketImpl.java
index 282c10f..c594387 100644
--- a/AndroidAsync/src/com/koushikdutta/async/AsyncSocketImpl.java
+++ b/AndroidAsync/src/com/koushikdutta/async/AsyncSocketImpl.java
@@ -7,9 +7,9 @@ import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 
 import junit.framework.Assert;
-import android.util.Log;
 
 import com.koushikdutta.async.callback.ClosedCallback;
+import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.DataCallback;
 import com.koushikdutta.async.callback.WritableCallback;
 
@@ -91,8 +91,16 @@ class AsyncSocketImpl implements AsyncSocket {
         }
     }
 
+    private ByteBuffer pending;
+    private void spit(ByteBuffer b) {
+        ByteBufferList list = new ByteBufferList();
+        list.add(b);
+        Util.emitAllData(this, list);
+    }
+    
     int mToAlloc = 0;
     int onReadable() {
+        Assert.assertTrue(pending == null);
         // even if the socket is paused,
         // it may end up getting a queued readable event if it is
         // already in the selector's ready queue.
@@ -123,11 +131,11 @@ class AsyncSocketImpl implements AsyncSocket {
                 mToAlloc = read * 2;
                 b.limit(b.position());
                 b.position(0);
-                ByteBufferList list = new ByteBufferList();
-                list.add(b);
-                Util.emitAllData(this, list);
+                spit(b);
+                if (b.remaining() != 0)
+                    pending = b;
             }
-        
+
             if (closed)
                 reportClose();
         }
@@ -164,6 +172,13 @@ class AsyncSocketImpl implements AsyncSocket {
     DataCallback mDataHandler;
     @Override
     public void setDataCallback(DataCallback callback) {
+        try {
+            throw new Exception();
+        }
+        catch (Exception ex) {
+            System.out.println("data callback set " + this);
+            ex.printStackTrace();
+        }
         mDataHandler = callback;
     }
 
@@ -190,19 +205,19 @@ class AsyncSocketImpl implements AsyncSocket {
 
     void report(Exception e) {
         if (mExceptionCallback != null)
-            mExceptionCallback.onException(e);
+            mExceptionCallback.onCompleted(e);
         else
             e.printStackTrace();
     }
     
-    private ExceptionCallback mExceptionCallback;
+    private CompletedCallback mExceptionCallback;
     @Override
-    public void setExceptionCallback(ExceptionCallback callback) {
+    public void setCompletedCallback(CompletedCallback callback) {
         mExceptionCallback = callback;
     }
 
     @Override
-    public ExceptionCallback getExceptionCallback() {
+    public CompletedCallback getCompletedCallback() {
         return mExceptionCallback;
     }
     
@@ -239,6 +254,11 @@ class AsyncSocketImpl implements AsyncSocket {
         }
         catch (Exception ex) {
         }
+        if (pending != null) {
+            spit(pending);
+            if (pending.remaining() == 0)
+                pending = null;
+        }
     }
     
     @Override
diff --git a/AndroidAsync/src/com/koushikdutta/async/BufferedDataEmitter.java b/AndroidAsync/src/com/koushikdutta/async/BufferedDataEmitter.java
index f00ad63..2d4033f 100644
--- a/AndroidAsync/src/com/koushikdutta/async/BufferedDataEmitter.java
+++ b/AndroidAsync/src/com/koushikdutta/async/BufferedDataEmitter.java
@@ -7,7 +7,7 @@ public class BufferedDataEmitter implements DataEmitter, DataCallback {
     }
     
     public void onDataAvailable() {
-        if (mDataCallback != null)
+        if (mDataCallback != null && !mPaused && mBuffers.remaining() > 0)
             mDataCallback.onDataAvailable(this, mBuffers);
     }
     
@@ -36,4 +36,23 @@ public class BufferedDataEmitter implements DataEmitter, DataCallback {
 
         onDataAvailable();        
     }
+
+    private boolean mPaused;
+    @Override
+    public void pause() {
+        mPaused = true;
+    }
+
+    @Override
+    public void resume() {
+        if (!mPaused)
+            return;
+        mPaused = false;
+        onDataAvailable();
+    }
+
+    @Override
+    public boolean isPaused() {
+        return mPaused;
+    }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/ByteBufferList.java b/AndroidAsync/src/com/koushikdutta/async/ByteBufferList.java
index 8836e46..d3ff4b9 100644
--- a/AndroidAsync/src/com/koushikdutta/async/ByteBufferList.java
+++ b/AndroidAsync/src/com/koushikdutta/async/ByteBufferList.java
@@ -177,4 +177,12 @@ public class ByteBufferList implements Iterable<ByteBuffer> {
             }
         }
     }
+
+    public String debug() {
+        StringBuilder builder = new StringBuilder();
+        for (ByteBuffer bb: this) {
+            builder.append(new String(bb.array(), bb.arrayOffset() + bb.position(), bb.remaining()));
+        }
+        return builder.toString();
+    }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/CompletedEmitter.java b/AndroidAsync/src/com/koushikdutta/async/CompletedEmitter.java
new file mode 100644
index 0000000..3eb7bf7
--- /dev/null
+++ b/AndroidAsync/src/com/koushikdutta/async/CompletedEmitter.java
@@ -0,0 +1,8 @@
+package com.koushikdutta.async;
+
+import com.koushikdutta.async.callback.CompletedCallback;
+
+public interface CompletedEmitter {
+    public void setCompletedCallback(CompletedCallback callback);
+    public CompletedCallback getCompletedCallback();
+}
diff --git a/AndroidAsync/src/com/koushikdutta/async/DataEmitter.java b/AndroidAsync/src/com/koushikdutta/async/DataEmitter.java
index c868be6..b2ea176 100644
--- a/AndroidAsync/src/com/koushikdutta/async/DataEmitter.java
+++ b/AndroidAsync/src/com/koushikdutta/async/DataEmitter.java
@@ -6,4 +6,7 @@ public interface DataEmitter {
     public void setDataCallback(DataCallback callback);
     public DataCallback getDataCallback();
     public boolean isChunked();
+    public void pause();
+    public void resume();
+    public boolean isPaused();
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/ExceptionCallback.java b/AndroidAsync/src/com/koushikdutta/async/ExceptionCallback.java
deleted file mode 100644
index f33dcac..0000000
--- a/AndroidAsync/src/com/koushikdutta/async/ExceptionCallback.java
+++ /dev/null
@@ -1,5 +0,0 @@
-package com.koushikdutta.async;
-
-public interface ExceptionCallback {
-    public void onException(Exception error);
-}
diff --git a/AndroidAsync/src/com/koushikdutta/async/ExceptionEmitter.java b/AndroidAsync/src/com/koushikdutta/async/ExceptionEmitter.java
deleted file mode 100644
index 21d5e2f..0000000
--- a/AndroidAsync/src/com/koushikdutta/async/ExceptionEmitter.java
+++ /dev/null
@@ -1,6 +0,0 @@
-package com.koushikdutta.async;
-
-public interface ExceptionEmitter {
-    public void setExceptionCallback(ExceptionCallback callback);
-    public ExceptionCallback getExceptionCallback();
-}
diff --git a/AndroidAsync/src/com/koushikdutta/async/FilteredDataCallback.java b/AndroidAsync/src/com/koushikdutta/async/FilteredDataCallback.java
index fc33aeb..6ad3154 100644
--- a/AndroidAsync/src/com/koushikdutta/async/FilteredDataCallback.java
+++ b/AndroidAsync/src/com/koushikdutta/async/FilteredDataCallback.java
@@ -2,9 +2,10 @@ package com.koushikdutta.async;
 
 import junit.framework.Assert;
 
+import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.DataCallback;
 
-public class FilteredDataCallback implements DataEmitter, DataCallback, ExceptionEmitter {
+public class FilteredDataCallback implements DataEmitter, DataCallback, CompletedEmitter {
     public FilteredDataCallback() {
     }
 
@@ -25,24 +26,51 @@ public class FilteredDataCallback implements DataEmitter, DataCallback, Exceptio
     }
     
     protected void report(Exception e) {
-        if (mExceptionCallback != null)
-            mExceptionCallback.onException(e);
+        if (mCompletedCallback != null)
+            mCompletedCallback.onCompleted(e);
     }
     
     @Override
-    public ExceptionCallback getExceptionCallback() {
-        return mExceptionCallback;
+    public CompletedCallback getCompletedCallback() {
+        return mCompletedCallback;
     }
     
     @Override
-    public void setExceptionCallback(ExceptionCallback callback) {
-        mExceptionCallback = callback;
+    public void setCompletedCallback(CompletedCallback callback) {
+        mCompletedCallback = callback;
     }
-    ExceptionCallback mExceptionCallback;
+    CompletedCallback mCompletedCallback;
     
     @Override
     public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
         Assert.assertNotNull(mDataCallback);
         Util.emitAllData(this, bb);
+        if (bb.remaining() > 0)
+            pending = bb;
+    }
+
+    private ByteBufferList pending;
+    private boolean mPaused;
+    @Override
+    public void pause() {
+        mPaused = true;
+    }
+
+    @Override
+    public void resume() {
+        if (!mPaused)
+            return;
+        mPaused = false;
+        if (pending != null) {
+            Assert.assertNotNull(mDataCallback);
+            Util.emitAllData(this, pending);
+            if (pending.remaining() == 0)
+                pending = null;
+        }
+    }
+
+    @Override
+    public boolean isPaused() {
+        return mPaused;
     }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/Util.java b/AndroidAsync/src/com/koushikdutta/async/Util.java
index 524e0b3..a9c69ae 100644
--- a/AndroidAsync/src/com/koushikdutta/async/Util.java
+++ b/AndroidAsync/src/com/koushikdutta/async/Util.java
@@ -13,15 +13,18 @@ import com.koushikdutta.async.callback.WritableCallback;
 public class Util {
     public static void emitAllData(DataEmitter emitter, ByteBufferList list) {
         int remaining;
-        while (emitter.getDataCallback() != null && (remaining = list.remaining()) > 0) {
-            DataCallback handler = emitter.getDataCallback();
+        AsyncSocket socket = null;
+        if (emitter instanceof AsyncSocket)
+            socket = (AsyncSocket)emitter;
+        DataCallback handler;
+        while ((socket == null || !socket.isPaused()) && (handler = emitter.getDataCallback()) != null && (remaining = list.remaining()) > 0) {
             handler.onDataAvailable(emitter, list);
             if (remaining == list.remaining() && handler == emitter.getDataCallback()) {
                 Assert.fail("mDataHandler failed to consume data, yet remains the mDataHandler.");
                 break;
             }
         }
-        Assert.assertEquals(list.remaining(), 0);
+        Assert.assertTrue(list.remaining() == 0 || (socket != null && socket.isPaused()));
     }
 
     public static void emitAllData(DataEmitter emitter, ByteBuffer b) {
diff --git a/AndroidAsync/src/com/koushikdutta/async/callback/ListenCallback.java b/AndroidAsync/src/com/koushikdutta/async/callback/ListenCallback.java
index 0648f71..99ff78c 100644
--- a/AndroidAsync/src/com/koushikdutta/async/callback/ListenCallback.java
+++ b/AndroidAsync/src/com/koushikdutta/async/callback/ListenCallback.java
@@ -2,10 +2,9 @@ package com.koushikdutta.async.callback;
 
 import com.koushikdutta.async.AsyncServerSocket;
 import com.koushikdutta.async.AsyncSocket;
-import com.koushikdutta.async.ExceptionCallback;
 
 
-public interface ListenCallback extends ExceptionCallback {
+public interface ListenCallback extends CompletedCallback {
     public void onAccepted(AsyncSocket handler);
     public void onListening(AsyncServerSocket socket);
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
index 8da007c..0a36a72 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
@@ -11,8 +11,6 @@ import java.nio.ByteBuffer;
 import java.util.HashSet;
 import java.util.Hashtable;
 
-import junit.framework.Assert;
-
 import org.json.JSONException;
 import org.json.JSONObject;
 
@@ -24,8 +22,6 @@ import com.koushikdutta.async.AsyncSocket;
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
 import com.koushikdutta.async.DataExchange;
-import com.koushikdutta.async.ExceptionCallback;
-import com.koushikdutta.async.FilteredDataCallback;
 import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.SSLDataExchange;
 import com.koushikdutta.async.callback.ClosedCallback;
@@ -33,11 +29,7 @@ import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.ConnectCallback;
 import com.koushikdutta.async.callback.DataCallback;
 import com.koushikdutta.async.callback.RequestCallback;
-import com.koushikdutta.async.http.filter.ChunkedInputFilter;
-import com.koushikdutta.async.http.filter.GZIPInputFilter;
-import com.koushikdutta.async.http.filter.InflaterInputFilter;
 import com.koushikdutta.async.http.libcore.RawHeaders;
-import com.koushikdutta.async.http.libcore.ResponseHeaders;
 import com.koushikdutta.async.stream.OutputStreamDataCallback;
 
 public class AsyncHttpClient {
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponse.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponse.java
index 5e67256..05b184a 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponse.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponse.java
@@ -1,11 +1,11 @@
 package com.koushikdutta.async.http;
 
 import com.koushikdutta.async.DataExchange;
-import com.koushikdutta.async.ExceptionEmitter;
+import com.koushikdutta.async.CompletedEmitter;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.http.libcore.ResponseHeaders;
 
-public interface AsyncHttpResponse extends DataExchange, ExceptionEmitter {
+public interface AsyncHttpResponse extends DataExchange, CompletedEmitter {
     public void setCompletedCallback(CompletedCallback handler);
     public CompletedCallback getCompletedCallback();
     public ResponseHeaders getHeaders();
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
index 11601ac..f77cbce 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
@@ -8,7 +8,6 @@ import com.koushikdutta.async.AsyncSocket;
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
 import com.koushikdutta.async.DataExchange;
-import com.koushikdutta.async.ExceptionCallback;
 import com.koushikdutta.async.FilteredDataCallback;
 import com.koushikdutta.async.LineEmitter;
 import com.koushikdutta.async.LineEmitter.StringCallback;
@@ -50,12 +49,7 @@ public class AsyncHttpResponseImpl extends FilteredDataCallback implements Async
         LineEmitter liner = new LineEmitter(exchange);
         liner.setLineCallback(mHeaderCallback);
         
-        mSocket.setExceptionCallback(new ExceptionCallback() {
-            @Override
-            public void onException(Exception error) {
-                report(error);
-            }
-        });
+        mSocket.setCompletedCallback(mReporter);
         mSocket.setClosedCallback(new ClosedCallback() {
             @Override
             public void onClosed() {
@@ -66,9 +60,9 @@ public class AsyncHttpResponseImpl extends FilteredDataCallback implements Async
         });
     }
     
-    private ExceptionCallback mReporter = new ExceptionCallback() {
+    private CompletedCallback mReporter = new CompletedCallback() {
         @Override
-        public void onException(Exception error) {
+        public void onCompleted(Exception error) {
             report(error);
         }
     };
@@ -133,7 +127,7 @@ public class AsyncHttpResponseImpl extends FilteredDataCallback implements Async
         });
         mExchange.setWriteableCallback(null);
         mSocket.setClosedCallback(null);
-        mSocket.setExceptionCallback(null);
+        mSocket.setCompletedCallback(null);
         mCompleted = true;
 //        System.out.println("closing up shop");
         if (mCompletedCallback != null)
@@ -151,17 +145,6 @@ public class AsyncHttpResponseImpl extends FilteredDataCallback implements Async
         return mCompletedCallback;
     }
 
-    ExceptionCallback mErrorCallback;
-    @Override
-    public void setExceptionCallback(ExceptionCallback callback) {
-        mErrorCallback = callback;
-    }
-
-    @Override
-    public ExceptionCallback getExceptionCallback() {
-        return mErrorCallback;
-    }
-
     @Override
     public ResponseHeaders getHeaders() {
         return mHeaders;
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/Util.java b/AndroidAsync/src/com/koushikdutta/async/http/Util.java
index 39a069c..1f4b878 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/Util.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/Util.java
@@ -4,8 +4,8 @@ import junit.framework.Assert;
 
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
-import com.koushikdutta.async.ExceptionCallback;
 import com.koushikdutta.async.FilteredDataCallback;
+import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.DataCallback;
 import com.koushikdutta.async.http.filter.ChunkedInputFilter;
 import com.koushikdutta.async.http.filter.GZIPInputFilter;
@@ -18,16 +18,16 @@ public class Util {
         return new UnknownRequestBody(emitter, headers.get("Content-Type"));
     }
     
-    public static DataCallback getBodyDecoder(DataCallback callback, RawHeaders headers, final ExceptionCallback reporter) {
+    public static DataCallback getBodyDecoder(DataCallback callback, RawHeaders headers, final CompletedCallback reporter) {
         if ("gzip".equals(headers.get("Content-Encoding"))) {
             GZIPInputFilter gunzipper = new GZIPInputFilter();
             gunzipper.setDataCallback(callback);
-            gunzipper.setExceptionCallback(reporter);
+            gunzipper.setCompletedCallback(reporter);
             callback = gunzipper;
         }        
         else if ("deflate".equals(headers.get("Content-Encoding"))) {
             InflaterInputFilter inflater = new InflaterInputFilter();
-            inflater.setExceptionCallback(reporter);
+            inflater.setCompletedCallback(reporter);
             inflater.setDataCallback(callback);
             callback = inflater;
         }
@@ -40,9 +40,9 @@ public class Util {
             _contentLength = -1;
         }
         final int contentLength = _contentLength;
-        if (!"chunked".equalsIgnoreCase(headers.get("Transfer-Encoding"))) {
+        if (-1 != contentLength) {
             if (contentLength < 0) {
-                reporter.onException(new Exception("not using chunked encoding, and no content-length found."));
+                reporter.onCompleted(new Exception("not using chunked encoding, and no content-length found."));
             }
             FilteredDataCallback contentLengthWatcher = new FilteredDataCallback() {
                 int totalRead = 0;
@@ -52,24 +52,28 @@ public class Util {
                     Assert.assertTrue(totalRead <= contentLength);
                     super.onDataAvailable(emitter, bb);
                     if (totalRead == contentLength)
-                        reporter.onException(null);
+                        reporter.onCompleted(null);
                 }
             };
             contentLengthWatcher.setDataCallback(callback);
             callback = contentLengthWatcher;
         }
-        else {
+        else if ("chunked".equalsIgnoreCase(headers.get("Transfer-Encoding"))) {
             ChunkedInputFilter chunker = new ChunkedInputFilter() {
                 @Override
                 public void onCompleted(Exception ex) {
-                    reporter.onException(ex);
+                    reporter.onCompleted(ex);
                 }
             };
             
-            chunker.setExceptionCallback(reporter);
+            chunker.setCompletedCallback(reporter);
             chunker.setDataCallback(callback);
             callback = chunker;
         }
+        else {
+            // we're done I guess.
+            reporter.onCompleted(null);
+        }
         return callback;
     }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
index 5298e9d..ed34674 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
@@ -19,18 +19,16 @@ import android.content.Context;
 import com.koushikdutta.async.AsyncServer;
 import com.koushikdutta.async.AsyncServerSocket;
 import com.koushikdutta.async.AsyncSocket;
-import com.koushikdutta.async.ExceptionCallback;
-import com.koushikdutta.async.ExceptionEmitter;
+import com.koushikdutta.async.CompletedEmitter;
 import com.koushikdutta.async.NullDataCallback;
 import com.koushikdutta.async.Util;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.ListenCallback;
-import com.koushikdutta.async.http.AsyncHttpClient;
 import com.koushikdutta.async.http.AsyncHttpGet;
 import com.koushikdutta.async.http.AsyncHttpPost;
 import com.koushikdutta.async.http.libcore.RawHeaders;
 
-public class AsyncHttpServer implements ExceptionEmitter {
+public class AsyncHttpServer implements CompletedEmitter {
     AsyncServerSocket mListener;
     public void stop() {
         if (mListener != null)
@@ -41,12 +39,14 @@ public class AsyncHttpServer implements ExceptionEmitter {
             @Override
             public void onAccepted(final AsyncSocket socket) {
                 AsyncHttpServerRequestImpl req = new AsyncHttpServerRequestImpl() {
+                    boolean responseComplete;
+                    boolean requestComplete;
                     @Override
                     protected void onHeadersReceived() {
                         super.onHeadersReceived();
                         
-                        socket.setDataCallback(new NullDataCallback());
                         RawHeaders headers = getRawHeaders();
+                        System.out.println(headers.toHeaderString());
                         
                         String statusLine = headers.getStatusLine();
                         String[] parts = statusLine.split(" ");
@@ -70,8 +70,9 @@ public class AsyncHttpServer implements ExceptionEmitter {
                         AsyncHttpServerResponseImpl res = new AsyncHttpServerResponseImpl(socket) {
                             @Override
                             protected void onCompleted() {
+                                responseComplete = true;
                                 // reuse the socket for a subsequent request.
-                                onAccepted(socket);
+                                handleOnCompleted();
                             }
                         };
                         if (match == null) {
@@ -83,16 +84,26 @@ public class AsyncHttpServer implements ExceptionEmitter {
                         match.callback.onRequest(this, res);
                     }
                     @Override
-                    protected void report(Exception e) {
-                        super.report(e);
-                        AsyncHttpServer.this.report(e);
+                    protected void onCompleted(Exception e) {
+                        requestComplete = true;
+                        super.onCompleted(e);
+                        mSocket.setDataCallback(null);
+                        mSocket.pause();
+                        handleOnCompleted();
+                    }
+                    
+                    private void handleOnCompleted() {
+                        if (requestComplete && responseComplete) {
+                            onAccepted(socket);
+                        }
                     }
                 };
                 req.setSocket(socket);
+                socket.resume();
             }
 
             @Override
-            public void onException(Exception error) {
+            public void onCompleted(Exception error) {
                 report(error);
             }
 
@@ -104,23 +115,23 @@ public class AsyncHttpServer implements ExceptionEmitter {
     }
 
     private void report(Exception ex) {
-        if (mExceptionCallback != null)
-            mExceptionCallback.onException(ex);
+        if (mCompletedCallback != null)
+            mCompletedCallback.onCompleted(ex);
     }
     
     public AsyncHttpServer(int port) {
         this(AsyncServer.getDefault(), port);
     }
 
-    ExceptionCallback mExceptionCallback;
+    CompletedCallback mCompletedCallback;
     @Override
-    public void setExceptionCallback(ExceptionCallback callback) {
-        mExceptionCallback = callback;        
+    public void setCompletedCallback(CompletedCallback callback) {
+        mCompletedCallback = callback;        
     }
 
     @Override
-    public ExceptionCallback getExceptionCallback() {
-        return mExceptionCallback;
+    public CompletedCallback getCompletedCallback() {
+        return mCompletedCallback;
     }
     
     private static class Pair {
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequestImpl.java b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequestImpl.java
index 49b7c58..5a6a7c2 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequestImpl.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequestImpl.java
@@ -3,9 +3,9 @@ package com.koushikdutta.async.http.server;
 import java.util.regex.Matcher;
 
 import com.koushikdutta.async.AsyncSocket;
-import com.koushikdutta.async.ExceptionCallback;
 import com.koushikdutta.async.LineEmitter;
 import com.koushikdutta.async.LineEmitter.StringCallback;
+import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.DataCallback;
 import com.koushikdutta.async.http.AsyncHttpRequestBody;
 import com.koushikdutta.async.http.Util;
@@ -17,14 +17,14 @@ public class AsyncHttpServerRequestImpl implements AsyncHttpServerRequest {
     AsyncSocket mSocket;
     Matcher mMatcher;
 
-    private ExceptionCallback mReporter = new ExceptionCallback() {
+    private CompletedCallback mReporter = new CompletedCallback() {
         @Override
-        public void onException(Exception error) {
-            report(error);
+        public void onCompleted(Exception error) {
+            AsyncHttpServerRequestImpl.this.onCompleted(error);
         }
     };
 
-    protected void report(Exception e) {
+    protected void onCompleted(Exception e) {
         if (mBody != null)
             mBody.onCompleted(e);
     }
@@ -50,13 +50,13 @@ public class AsyncHttpServerRequestImpl implements AsyncHttpServerRequest {
                     mRawHeaders.addLine(s);
                 }
                 else {
-                    onHeadersReceived();
                     DataCallback callback = Util.getBodyDecoder(mBody = Util.getBody(mSocket, mRawHeaders), mRawHeaders, mReporter);
                     mSocket.setDataCallback(callback);
+                    onHeadersReceived();
                 }
             }
             catch (Exception ex) {
-                report(ex);
+                onCompleted(ex);
             }
         }
     };
@@ -103,4 +103,19 @@ public class AsyncHttpServerRequestImpl implements AsyncHttpServerRequest {
     public AsyncHttpRequestBody getBody() {
         return mBody;
     }
+
+    @Override
+    public void pause() {
+        mSocket.pause();
+    }
+
+    @Override
+    public void resume() {
+        mSocket.resume();
+    }
+
+    @Override
+    public boolean isPaused() {
+        return mSocket.isPaused();
+    }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/WebSocket.java b/AndroidAsync/src/com/koushikdutta/async/http/server/WebSocket.java
index bd3cb06..a4b9a55 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/WebSocket.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/WebSocket.java
@@ -1,10 +1,10 @@
 package com.koushikdutta.async.http.server;
 
 import com.koushikdutta.async.CloseableData;
-import com.koushikdutta.async.ExceptionEmitter;
+import com.koushikdutta.async.CompletedEmitter;
 
 
-public interface WebSocket extends CloseableData, ExceptionEmitter {
+public interface WebSocket extends CloseableData, CompletedEmitter {
     static public interface StringCallback {
         public void onStringAvailable(String s);
     }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/WebSocketImpl.java b/AndroidAsync/src/com/koushikdutta/async/http/server/WebSocketImpl.java
index 3c504f3..b4e0d1b 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/WebSocketImpl.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/WebSocketImpl.java
@@ -7,8 +7,8 @@ import android.util.Base64;
 
 import com.koushikdutta.async.AsyncSocket;
 import com.koushikdutta.async.BufferedDataSink;
-import com.koushikdutta.async.ExceptionCallback;
 import com.koushikdutta.async.callback.ClosedCallback;
+import com.koushikdutta.async.callback.CompletedCallback;
 
 public class WebSocketImpl implements WebSocket {
     private static String SHA1(String text) {
@@ -56,7 +56,7 @@ public class WebSocketImpl implements WebSocket {
             @Override
             protected void report(Exception ex) {
                 if (WebSocketImpl.this.mExceptionCallback != null)
-                    WebSocketImpl.this.mExceptionCallback.onException(ex);
+                    WebSocketImpl.this.mExceptionCallback.onCompleted(ex);
             }
             @Override
             protected void onMessage(byte[] payload) {
@@ -95,14 +95,14 @@ public class WebSocketImpl implements WebSocket {
         return mClosedCallback;
     }
 
-    ExceptionCallback mExceptionCallback;
+    CompletedCallback mExceptionCallback;
     @Override
-    public void setExceptionCallback(ExceptionCallback callback) {
-        mExceptionCallback = callback;        
+    public void setCompletedCallback(CompletedCallback callback) {
+        mExceptionCallback = callback;
     }
 
     @Override
-    public ExceptionCallback getExceptionCallback() {
+    public CompletedCallback getCompletedCallback() {
         return mExceptionCallback;
     }
 
diff --git a/AndroidAsync/src/com/koushikdutta/async/stream/OutputStreamDataCallback.java b/AndroidAsync/src/com/koushikdutta/async/stream/OutputStreamDataCallback.java
index d6f5747..4f3ffcb 100644
--- a/AndroidAsync/src/com/koushikdutta/async/stream/OutputStreamDataCallback.java
+++ b/AndroidAsync/src/com/koushikdutta/async/stream/OutputStreamDataCallback.java
@@ -6,10 +6,10 @@ import java.nio.ByteBuffer;
 
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
-import com.koushikdutta.async.ExceptionCallback;
+import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.DataCallback;
 
-public class OutputStreamDataCallback implements DataCallback, ExceptionCallback {
+public class OutputStreamDataCallback implements DataCallback, CompletedCallback {
     private OutputStream mOutput;
     public OutputStreamDataCallback(OutputStream os) {
         mOutput = os;
@@ -23,7 +23,7 @@ public class OutputStreamDataCallback implements DataCallback, ExceptionCallback
             }
         }
         catch (Exception ex) {
-            onException(ex);
+            onCompleted(ex);
         }
         bb.clear();
     }
@@ -33,12 +33,12 @@ public class OutputStreamDataCallback implements DataCallback, ExceptionCallback
             mOutput.close();
         }
         catch (IOException e) {
-            onException(e);
+            onCompleted(e);
         }
     }
 
     @Override
-    public void onException(Exception error) {
+    public void onCompleted(Exception error) {
         error.printStackTrace();       
     }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/test/TestActivity.java b/AndroidAsync/src/com/koushikdutta/test/TestActivity.java
index c276852..a93036f 100644
--- a/AndroidAsync/src/com/koushikdutta/test/TestActivity.java
+++ b/AndroidAsync/src/com/koushikdutta/test/TestActivity.java
@@ -366,8 +366,9 @@ public class TestActivity extends Activity {
                 }
 
                 @Override
-                public void onException(Exception error) {
-                    error.printStackTrace();
+                public void onCompleted(Exception error) {
+                    if (error != null)
+                        error.printStackTrace();
                 }
 
                 @Override