wip

24 files changed, 199 insertions, 126 deletions

diff --git a/AndroidAsync/src/com/koushikdutta/async/ByteBufferList.java b/AndroidAsync/src/com/koushikdutta/async/ByteBufferList.java
index 94394b6..43fd589 100644
--- a/AndroidAsync/src/com/koushikdutta/async/ByteBufferList.java
+++ b/AndroidAsync/src/com/koushikdutta/async/ByteBufferList.java
@@ -52,7 +52,7 @@ public class ByteBufferList {
         return remaining == 0;
     }
 
-    int remaining = 0;
+    private int remaining = 0;
     public int remaining() {
         return remaining;
     }
diff --git a/AndroidAsync/src/com/koushikdutta/async/FileDataEmitter.java b/AndroidAsync/src/com/koushikdutta/async/FileDataEmitter.java
index f76c99c..c181f57 100644
--- a/AndroidAsync/src/com/koushikdutta/async/FileDataEmitter.java
+++ b/AndroidAsync/src/com/koushikdutta/async/FileDataEmitter.java
@@ -17,7 +17,9 @@ public class FileDataEmitter implements DataEmitter {
     public FileDataEmitter(AsyncServer server, File file) {
         this.server = server;
         this.file = file;
-        doResume();
+        paused = !server.isAffinityThread();
+        if (!paused)
+            doResume();
     }
 
     DataCallback callback;
@@ -45,6 +47,7 @@ public class FileDataEmitter implements DataEmitter {
     @Override
     public void resume() {
         paused = false;
+        doResume();
     }
 
     private void report(Exception e) {
diff --git a/AndroidAsync/src/com/koushikdutta/async/Util.java b/AndroidAsync/src/com/koushikdutta/async/Util.java
index 230b7a7..0e30e8c 100644
--- a/AndroidAsync/src/com/koushikdutta/async/Util.java
+++ b/AndroidAsync/src/com/koushikdutta/async/Util.java
@@ -2,6 +2,7 @@ package com.koushikdutta.async;
 
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.DataCallback;
+import com.koushikdutta.async.callback.ProgressCallback;
 import com.koushikdutta.async.callback.WritableCallback;
 import com.koushikdutta.async.wrapper.AsyncSocketWrapper;
 import com.koushikdutta.async.wrapper.DataEmitterWrapper;
@@ -151,19 +152,24 @@ public class Util {
     }
 
     public static void writeAll(final DataSink sink, final ByteBufferList bb, final CompletedCallback callback) {
-        sink.setWriteableCallback(new WritableCallback() {
+        writeAll(sink, bb, null, callback);
+    }
+
+    public static void writeAll(final DataSink sink, final ByteBufferList bb, final ProgressCallback progress, final CompletedCallback callback) {
+        WritableCallback wc;
+        final int total = bb.remaining();
+        sink.setWriteableCallback(wc = new WritableCallback() {
             @Override
             public void onWriteable() {
-                if (bb.remaining() == 0)
-                    return;
-                sink.write(bb);
+                if (bb.remaining() > 0)
+                    sink.write(bb);
+                if (progress != null)
+                    progress.onProgress(total - bb.remaining(), total);
                 if (bb.remaining() == 0 && callback != null)
                     callback.onCompleted(null);
             }
         });
-        sink.write(bb);
-        if (bb.remaining() == 0 && callback != null)
-            callback.onCompleted(null);
+        wc.onWriteable();
     }
     public static void writeAll(DataSink sink, byte[] bytes, CompletedCallback callback) {
         ByteBuffer bb = ByteBuffer.wrap(bytes);
diff --git a/AndroidAsync/src/com/koushikdutta/async/callback/ProgressCallback.java b/AndroidAsync/src/com/koushikdutta/async/callback/ProgressCallback.java
new file mode 100644
index 0000000..ea382be
--- /dev/null
+++ b/AndroidAsync/src/com/koushikdutta/async/callback/ProgressCallback.java
@@ -0,0 +1,10 @@
+package com.koushikdutta.async.callback;
+
+import com.koushikdutta.async.http.AsyncHttpResponse;
+
+/**
+ * Created by koush on 5/27/13.
+ */
+public interface ProgressCallback {
+    public void onProgress(int progress, int total);
+}
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
index 6873961..3bd055a 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpClient.java
@@ -1,18 +1,16 @@
 package com.koushikdutta.async.http;
 
 import android.os.Handler;
-import android.util.Log;
 import com.koushikdutta.async.*;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.ConnectCallback;
-import com.koushikdutta.async.callback.DataCallback;
+import com.koushikdutta.async.callback.ProgressCallback;
 import com.koushikdutta.async.callback.RequestCallback;
 import com.koushikdutta.async.future.*;
 import com.koushikdutta.async.http.AsyncHttpClientMiddleware.OnRequestCompleteData;
 import com.koushikdutta.async.http.libcore.RawHeaders;
 import com.koushikdutta.async.parser.*;
 import com.koushikdutta.async.stream.OutputStreamDataCallback;
-import org.json.JSONException;
 import org.json.JSONObject;
 
 import java.io.*;
@@ -478,12 +476,7 @@ public class AsyncHttpClient {
 
                 final int contentLength = response.getHeaders().getContentLength();
 
-                Future<T> parsed = parser.parse(response, new ParserCallback() {
-                    @Override
-                    public void onProgress(int bytesParsed) {
-                        invokeProgress(callback, response, bytesParsed, contentLength);
-                    }
-                })
+                Future<T> parsed = parser.parse(response)
                 .setCallback(new FutureCallback<T>() {
                     @Override
                     public void onCompleted(Exception e, T result) {
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpRequestBody.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpRequestBody.java
index 8292098..ec17473 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpRequestBody.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpRequestBody.java
@@ -7,7 +7,5 @@ public interface AsyncHttpRequestBody<T> extends DataCallback {
     public String getContentType();
     public boolean readFullyOnRequest();
     public int length();
-
-    @Deprecated
     public T get();
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
index 677cfcd..964e20a 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/AsyncHttpResponseImpl.java
@@ -92,7 +92,7 @@ abstract class AsyncHttpResponseImpl extends FilteredDataEmitter implements Asyn
                     // socket may get detached after headers (websocket)
                     if (mSocket == null)
                         return;
-                    DataEmitter emitter = Util.getBodyDecoder(mSocket, mRawHeaders, false, mReporter);
+                    DataEmitter emitter = HttpUtil.getBodyDecoder(mSocket, mRawHeaders, false);
                     setDataEmitter(emitter);
                 }
             }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/HeaderMap.java b/AndroidAsync/src/com/koushikdutta/async/http/HeaderMap.java
deleted file mode 100644
index b73c1f8..0000000
--- a/AndroidAsync/src/com/koushikdutta/async/http/HeaderMap.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package com.koushikdutta.async.http;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-import com.koushikdutta.async.http.libcore.RawHeaders;
-
-public class HeaderMap {
-    public static Map<String,String> parse(RawHeaders headers, String header) {
-        HashMap<String, String> map = new HashMap<String, String>();
-        String value = headers.get(header);
-        String[] parts = value.split(";");
-        for (String part: parts) {
-            String[] pair = part.split("=", 2);
-            String key = pair[0].trim();
-            String v = null;
-            if (pair.length > 1)
-                v = pair[1];
-            if (v != null && v.endsWith("\"") && v.startsWith("\""))
-                v = v.substring(1, v.length() - 1);
-            map.put(key, v);
-        }
-        return Collections.unmodifiableMap(map);
-    }
-}
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/Util.java b/AndroidAsync/src/com/koushikdutta/async/http/HttpUtil.java
index 5c868a5..73027eb 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/Util.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/HttpUtil.java
@@ -11,7 +11,7 @@ import com.koushikdutta.async.http.filter.InflaterInputFilter;
 import com.koushikdutta.async.http.libcore.RawHeaders;
 import com.koushikdutta.async.http.server.UnknownRequestBody;
 
-public class Util {
+public class HttpUtil {
     public static AsyncHttpRequestBody getBody(DataEmitter emitter, CompletedCallback reporter, RawHeaders headers) {
         String contentType = headers.get("Content-Type");
         if (contentType != null) {
@@ -21,7 +21,7 @@ public class Util {
             }
             for (String ct: values) {
                 if (UrlEncodedFormBody.CONTENT_TYPE.equals(ct)) {
-                    UrlEncodedFormBody ret = new UrlEncodedFormBody(); 
+                    UrlEncodedFormBody ret = new UrlEncodedFormBody();
                     emitter.setDataCallback(ret);
                     return ret;
                 }
@@ -57,7 +57,7 @@ public class Util {
         }
     }
     
-    public static DataEmitter getBodyDecoder(DataEmitter emitter, RawHeaders headers, boolean server, final CompletedCallback reporter) {
+    public static DataEmitter getBodyDecoder(DataEmitter emitter, RawHeaders headers, boolean server) {
         int _contentLength;
         try {
             _contentLength = Integer.parseInt(headers.get("Content-Length"));
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/Multimap.java b/AndroidAsync/src/com/koushikdutta/async/http/Multimap.java
new file mode 100644
index 0000000..2af816d
--- /dev/null
+++ b/AndroidAsync/src/com/koushikdutta/async/http/Multimap.java
@@ -0,0 +1,98 @@
+package com.koushikdutta.async.http;
+
+import android.net.Uri;
+import com.koushikdutta.async.http.libcore.RawHeaders;
+import org.apache.http.NameValuePair;
+import org.apache.http.message.BasicNameValuePair;
+
+import java.util.ArrayList;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Created by koush on 5/27/13.
+ */
+public class Multimap extends Hashtable<String, ArrayList<String>> implements Iterable<NameValuePair> {
+    public Multimap() {
+    }
+
+    public String getString(String name) {
+        ArrayList<String> ret = get(name);
+        if (ret == null || ret.size() == 0)
+            return null;
+        return ret.get(0);
+    }
+
+    public void add(String name, String value) {
+        ArrayList<String> ret = get(name);
+        if (ret == null) {
+            ret = new ArrayList<String>();
+            put(name, ret);
+        }
+        ret.add(value);
+    }
+
+    public void put(String name, String value) {
+        ArrayList<String> ret = new ArrayList<String>();
+        ret.add(value);
+        put(name, ret);
+    }
+
+    public Multimap(RawHeaders headers) {
+        headers.toMultimap().putAll(this);
+    }
+
+    public Multimap(List<NameValuePair> pairs) {
+        for (NameValuePair pair: pairs)
+            add(pair.getName(), pair.getValue());
+    }
+
+    public static Multimap parseHeader(String header) {
+        Multimap map = new Multimap();
+        String[] parts = header.split(";");
+        for (String part: parts) {
+            String[] pair = part.split("=", 2);
+            String key = pair[0].trim();
+            String v = null;
+            if (pair.length > 1)
+                v = pair[1];
+            if (v != null && v.endsWith("\"") && v.startsWith("\""))
+                v = v.substring(1, v.length() - 1);
+            map.add(key, v);
+        }
+        return map;
+    }
+
+    public static Multimap parseHeader(RawHeaders headers, String header) {
+        return parseHeader(headers.get(header));
+    }
+
+    public static Multimap parseQuery(String query) {
+        Multimap map = new Multimap();
+        String[] pairs = query.split("&");
+        for (String p : pairs) {
+            String[] pair = p.split("=", 2);
+            if (pair.length == 0)
+                continue;
+            String name = Uri.decode(pair[0]);
+            String value = null;
+            if (pair.length == 2)
+                value = Uri.decode(pair[1]);
+            map.add(name, value);
+        }
+        return map;
+    }
+
+    @Override
+    public Iterator<NameValuePair> iterator() {
+        ArrayList<NameValuePair> ret = new ArrayList<NameValuePair>();
+        for (String name: keySet()) {
+            ArrayList<String> values = get(name);
+            for (String value: values) {
+                ret.add(new BasicNameValuePair(name, value));
+            }
+        }
+        return ret.iterator();
+    }
+}
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/MultipartFormDataBody.java b/AndroidAsync/src/com/koushikdutta/async/http/MultipartFormDataBody.java
index 6d0ce2f..4709a98 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/MultipartFormDataBody.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/MultipartFormDataBody.java
@@ -1,5 +1,6 @@
 package com.koushikdutta.async.http;
 
+import android.util.Log;
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
 import com.koushikdutta.async.LineEmitter;
@@ -57,6 +58,7 @@ public class MultipartFormDataBody extends BoundaryEmitter implements AsyncHttpR
             @Override
             public void onStringAvailable(String s) {
                 if (!"\r".equals(s)){
+                    Log.i("HEADER---", s);
                     headers.addLine(s);
                 }
                 else {
@@ -67,8 +69,10 @@ public class MultipartFormDataBody extends BoundaryEmitter implements AsyncHttpR
                     Part part = new Part(headers);
                     if (mCallback != null)
                         mCallback.onPart(part);
+                    Log.w("MULTIPART---", part.getName());
                     if (getDataCallback() == null) {
                         if (part.isFile()) {
+                            Log.w("MULTIPART", "UNHANDLED!");
                             setDataCallback(new NullDataCallback());
                             return;
                         }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/Part.java b/AndroidAsync/src/com/koushikdutta/async/http/Part.java
index eb54991..904b5ae 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/Part.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/Part.java
@@ -13,14 +13,14 @@ public class Part {
     public static final String CONTENT_DISPOSITION = "Content-Disposition";
     
     RawHeaders mHeaders;
-    Map<String, String> mContentDisposition;
+    Multimap mContentDisposition;
     public Part(RawHeaders headers) {
         mHeaders = headers;
-        mContentDisposition = HeaderMap.parse(mHeaders, CONTENT_DISPOSITION);
+        mContentDisposition = Multimap.parseHeader(mHeaders, CONTENT_DISPOSITION);
     }
     
     public String getName() {
-        return mContentDisposition.get("name");
+        return mContentDisposition.getString("name");
     }
     
     private int length = -1;
@@ -34,7 +34,7 @@ public class Part {
             }
         }
         mHeaders.set(CONTENT_DISPOSITION, builder.toString());
-        mContentDisposition = HeaderMap.parse(mHeaders, CONTENT_DISPOSITION);
+        mContentDisposition = Multimap.parseHeader(mHeaders, CONTENT_DISPOSITION);
     }
 
     public RawHeaders getRawHeaders() {
@@ -46,7 +46,7 @@ public class Part {
     }
 
     public String getFilename() {
-        String file = mContentDisposition.get("filename");
+        String file = mContentDisposition.getString("filename");
         if (file == null)
             return null;
         return new File(file).getName();
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpRequestBodyBase.java b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpRequestBodyBase.java
deleted file mode 100644
index 3cacfa8..0000000
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpRequestBodyBase.java
+++ /dev/null
@@ -1,38 +0,0 @@
-package com.koushikdutta.async.http.server;
-
-import com.koushikdutta.async.FilteredDataEmitter;
-import com.koushikdutta.async.http.AsyncHttpRequest;
-import com.koushikdutta.async.http.AsyncHttpRequestBody;
-import com.koushikdutta.async.http.AsyncHttpResponse;
-
-public abstract class AsyncHttpRequestBodyBase<T> extends FilteredDataEmitter implements AsyncHttpRequestBody<T> {
-    public AsyncHttpRequestBodyBase(String contentType) {
-        mContentType = contentType;
-    }
-
-    @Override
-    public void write(AsyncHttpRequest request, AsyncHttpResponse sink) {
-        assert false;
-    }
-
-    private String mContentType;
-    @Override
-    public String getContentType() {
-        return mContentType;
-    }
-
-//    @Override
-//    public void onCompleted(Exception ex) {
-//        report(ex);
-//    }
-
-    @Override
-    public boolean readFullyOnRequest() {
-        return false;
-    }
-
-    @Override
-    public int length() {
-        return -1;
-    }
-}
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
index 9baa6bb..4ae280f 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServer.java
@@ -141,11 +141,11 @@ public class AsyncHttpServer {
                 }
 
                 @Override
-                public Map<String, String> getQuery() {
+                public Multimap getQuery() {
                     String[] parts = fullPath.split("\\?", 2);
                     if (parts.length < 2)
-                        return new Hashtable<String, String>();
-                    return UrlEncodedFormBody.parse(parts[1]);
+                        return new Multimap();
+                    return Multimap.parseQuery(parts[1]);
                 }
             };
             req.setSocket(socket);
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequest.java b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequest.java
index 6b69012..8e4e8f6 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequest.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequest.java
@@ -6,6 +6,7 @@ import java.util.regex.Matcher;
 import com.koushikdutta.async.AsyncSocket;
 import com.koushikdutta.async.DataEmitter;
 import com.koushikdutta.async.http.AsyncHttpRequestBody;
+import com.koushikdutta.async.http.Multimap;
 import com.koushikdutta.async.http.libcore.RequestHeaders;
 
 public interface AsyncHttpServerRequest extends DataEmitter {
@@ -14,5 +15,5 @@ public interface AsyncHttpServerRequest extends DataEmitter {
     public AsyncHttpRequestBody getBody();
     public AsyncSocket getSocket();
     public String getPath();
-    public Map<String, String> getQuery();
+    public Multimap getQuery();
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequestImpl.java b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequestImpl.java
index 677a3f9..c2e18a9 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequestImpl.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/AsyncHttpServerRequestImpl.java
@@ -10,7 +10,7 @@ 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;
+import com.koushikdutta.async.http.HttpUtil;
 import com.koushikdutta.async.http.libcore.RawHeaders;
 import com.koushikdutta.async.http.libcore.RequestHeaders;
 
@@ -55,9 +55,9 @@ public abstract class AsyncHttpServerRequestImpl extends FilteredDataEmitter imp
                     mRawHeaders.addLine(s);
                 }
                 else {
-                    DataEmitter emitter = Util.getBodyDecoder(mSocket, mRawHeaders, true, mReporter);
+                    DataEmitter emitter = HttpUtil.getBodyDecoder(mSocket, mRawHeaders, true);
                     emitter.setEndCallback(mReporter);
-                    mBody = Util.getBody(emitter, mReporter, mRawHeaders);
+                    mBody = HttpUtil.getBody(emitter, mReporter, mRawHeaders);
                     mHeaders = new RequestHeaders(null, mRawHeaders);
                     onHeadersReceived();
                 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/http/server/BoundaryEmitter.java b/AndroidAsync/src/com/koushikdutta/async/http/server/BoundaryEmitter.java
index d4cf855..093ff69 100644
--- a/AndroidAsync/src/com/koushikdutta/async/http/server/BoundaryEmitter.java
+++ b/AndroidAsync/src/com/koushikdutta/async/http/server/BoundaryEmitter.java
@@ -1,5 +1,6 @@
 package com.koushikdutta.async.http.server;
 
+import android.util.Log;
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
 import com.koushikdutta.async.FilteredDataEmitter;
@@ -57,6 +58,7 @@ public class BoundaryEmitter extends FilteredDataEmitter {
     int state = 2;
     @Override
     public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
+        Log.d("BOUNDARY", "" + state);
 //        System.out.println(bb.getString());
 //        System.out.println("chunk: " + bb.remaining());
         
@@ -148,7 +150,7 @@ public class BoundaryEmitter extends FilteredDataEmitter {
         }
 
         if (last < buf.length) {
-//            System.out.println("amount left at boundary: " + (buf.length - last));
+            System.out.println("amount left at boundary: " + (buf.length - last));
 //            System.out.println(state);
             int keep = Math.max(state, 0);
             ByteBuffer b = ByteBuffer.wrap(buf, last, buf.length - last - keep);
diff --git a/AndroidAsync/src/com/koushikdutta/async/parser/AsyncParser.java b/AndroidAsync/src/com/koushikdutta/async/parser/AsyncParser.java
index ad84269..bcbb440 100644
--- a/AndroidAsync/src/com/koushikdutta/async/parser/AsyncParser.java
+++ b/AndroidAsync/src/com/koushikdutta/async/parser/AsyncParser.java
@@ -1,11 +1,14 @@
 package com.koushikdutta.async.parser;
 
 import com.koushikdutta.async.DataEmitter;
+import com.koushikdutta.async.DataSink;
+import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.future.Future;
 
 /**
  * Created by koush on 5/27/13.
  */
 public interface AsyncParser<T> {
-    Future<T> parse(DataEmitter emitter, ParserCallback callback);
+    Future<T> parse(DataEmitter emitter);
+    void write(DataSink sink, T value, CompletedCallback completed);
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/parser/ByteBufferListParser.java b/AndroidAsync/src/com/koushikdutta/async/parser/ByteBufferListParser.java
index 98b1a6d..224f151 100644
--- a/AndroidAsync/src/com/koushikdutta/async/parser/ByteBufferListParser.java
+++ b/AndroidAsync/src/com/koushikdutta/async/parser/ByteBufferListParser.java
@@ -2,8 +2,11 @@ package com.koushikdutta.async.parser;
 
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
+import com.koushikdutta.async.DataSink;
+import com.koushikdutta.async.Util;
 import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.callback.DataCallback;
+import com.koushikdutta.async.callback.ProgressCallback;
 import com.koushikdutta.async.future.Future;
 import com.koushikdutta.async.future.SimpleFuture;
 
@@ -12,7 +15,7 @@ import com.koushikdutta.async.future.SimpleFuture;
  */
 public class ByteBufferListParser implements AsyncParser<ByteBufferList> {
     @Override
-    public Future<ByteBufferList> parse(final DataEmitter emitter, final ParserCallback callback) {
+    public Future<ByteBufferList> parse(final DataEmitter emitter) {
         final ByteBufferList bb = new ByteBufferList();
         final SimpleFuture<ByteBufferList> ret = new SimpleFuture<ByteBufferList>() {
             @Override
@@ -21,13 +24,8 @@ public class ByteBufferListParser implements AsyncParser<ByteBufferList> {
             }
         };
         emitter.setDataCallback(new DataCallback() {
-            int parsed;
             @Override
             public void onDataAvailable(DataEmitter emitter, ByteBufferList data) {
-                if (callback != null) {
-                    parsed += data.remaining();
-                    callback.onProgress(parsed);
-                }
                 bb.add(data);
                 data.clear();
             }
@@ -52,4 +50,9 @@ public class ByteBufferListParser implements AsyncParser<ByteBufferList> {
 
         return ret;
     }
+
+    @Override
+    public void write(DataSink sink, ByteBufferList value, CompletedCallback completed) {
+        Util.writeAll(sink, value, completed);
+    }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/parser/JSONObjectParser.java b/AndroidAsync/src/com/koushikdutta/async/parser/JSONObjectParser.java
index 2052207..9441575 100644
--- a/AndroidAsync/src/com/koushikdutta/async/parser/JSONObjectParser.java
+++ b/AndroidAsync/src/com/koushikdutta/async/parser/JSONObjectParser.java
@@ -1,6 +1,8 @@
 package com.koushikdutta.async.parser;
 
 import com.koushikdutta.async.DataEmitter;
+import com.koushikdutta.async.DataSink;
+import com.koushikdutta.async.callback.CompletedCallback;
 import com.koushikdutta.async.future.Future;
 import com.koushikdutta.async.future.TransformFuture;
 import org.json.JSONObject;
@@ -10,13 +12,18 @@ import org.json.JSONObject;
  */
 public class JSONObjectParser implements AsyncParser<JSONObject> {
     @Override
-    public Future<JSONObject> parse(DataEmitter emitter, ParserCallback callback) {
+    public Future<JSONObject> parse(DataEmitter emitter) {
         return new TransformFuture<JSONObject, String>() {
             @Override
             protected void transform(String result) throws Exception {
                 setComplete(new JSONObject(result));
             }
         }
-        .from(new StringParser().parse(emitter, callback));
+        .from(new StringParser().parse(emitter));
+    }
+
+    @Override
+    public void write(DataSink sink, JSONObject value, CompletedCallback completed) {
+        new StringParser().write(sink, value.toString(), completed);
     }
 }
diff --git a/AndroidAsync/src/com/koushikdutta/async/parser/ParserCallback.java b/AndroidAsync/src/com/koushikdutta/async/parser/ParserCallback.java
deleted file mode 100644
index b78a9e7..0000000
--- a/AndroidAsync/src/com/koushikdutta/async/parser/ParserCallback.java
+++ /dev/null
@@ -1,10 +0,0 @@
-package com.koushikdutta.async.parser;
-
-import com.koushikdutta.async.http.AsyncHttpResponse;
-
-/**
- * Created by koush on 5/27/13.
- */
-public interface ParserCallback {
-    public void onProgress(int bytesParsed);
-}
diff --git a/AndroidAsync/src/com/koushikdutta/async/parser/StringParser.java b/AndroidAsync/src/com/koushikdutta/async/parser/StringParser.java
index a01aad1..496823a 100644
--- a/AndroidAsync/src/com/koushikdutta/async/parser/StringParser.java
+++ b/AndroidAsync/src/com/koushikdutta/async/parser/StringParser.java
@@ -2,6 +2,9 @@ package com.koushikdutta.async.parser;
 
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
+import com.koushikdutta.async.DataSink;
+import com.koushikdutta.async.callback.CompletedCallback;
+import com.koushikdutta.async.callback.ProgressCallback;
 import com.koushikdutta.async.future.Future;
 import com.koushikdutta.async.future.TransformFuture;
 
@@ -10,13 +13,18 @@ import com.koushikdutta.async.future.TransformFuture;
  */
 public class StringParser implements AsyncParser<String> {
     @Override
-    public Future<String> parse(DataEmitter emitter, ParserCallback callback) {
+    public Future<String> parse(DataEmitter emitter) {
         return new TransformFuture<String, ByteBufferList>() {
             @Override
             protected void transform(ByteBufferList result) throws Exception {
                 setComplete(result.readString());
             }
         }
-        .from(new ByteBufferListParser().parse(emitter, callback));
+        .from(new ByteBufferListParser().parse(emitter));
+    }
+
+    @Override
+    public void write(DataSink sink, String value, CompletedCallback completed) {
+        new ByteBufferListParser().write(sink, new ByteBufferList(value.getBytes()), completed);
     }
 }
diff --git a/AndroidAsyncTest/src/com/koushikdutta/async/test/FileTests.java b/AndroidAsyncTest/src/com/koushikdutta/async/test/FileTests.java
index 0f01f93..d062c6d 100644
--- a/AndroidAsyncTest/src/com/koushikdutta/async/test/FileTests.java
+++ b/AndroidAsyncTest/src/com/koushikdutta/async/test/FileTests.java
@@ -3,7 +3,10 @@ package com.koushikdutta.async.test;
 import com.koushikdutta.async.AsyncServer;
 import com.koushikdutta.async.FileDataEmitter;
 import com.koushikdutta.async.callback.CompletedCallback;
+import com.koushikdutta.async.future.Future;
+import com.koushikdutta.async.future.FutureCallback;
 import com.koushikdutta.async.http.StringBody;
+import com.koushikdutta.async.parser.StringParser;
 import junit.framework.TestCase;
 
 import java.io.File;
@@ -21,14 +24,14 @@ public class FileTests extends TestCase {
         StreamUtility.writeFile(f, "hello world");
         FileDataEmitter fdm = new FileDataEmitter(AsyncServer.getDefault(), f);
         final Md5 md5 = Md5.createInstance();
-        StringBody stringBody = new StringBody();
-        fdm.setDataCallback(stringBody);
-        fdm.setEndCallback(new CompletedCallback() {
+        Future<String> stringBody = new StringParser().parse(fdm)
+        .setCallback(new FutureCallback<String>() {
             @Override
-            public void onCompleted(Exception ex) {
+            public void onCompleted(Exception e, String result) {
                 semaphore.release();
             }
         });
+        fdm.resume();
 
         assertTrue("timeout", semaphore.tryAcquire(TIMEOUT, TimeUnit.MILLISECONDS));
         assertEquals("hello world", stringBody.get());
diff --git a/AndroidAsyncTest/src/com/koushikdutta/async/test/MultipartTests.java b/AndroidAsyncTest/src/com/koushikdutta/async/test/MultipartTests.java
index 81f537b..ad7e436 100644
--- a/AndroidAsyncTest/src/com/koushikdutta/async/test/MultipartTests.java
+++ b/AndroidAsyncTest/src/com/koushikdutta/async/test/MultipartTests.java
@@ -1,6 +1,7 @@
 package com.koushikdutta.async.test;
 
 import android.os.Environment;
+import android.util.Log;
 import com.koushikdutta.async.AsyncServer;
 import com.koushikdutta.async.ByteBufferList;
 import com.koushikdutta.async.DataEmitter;
@@ -47,6 +48,7 @@ public class MultipartTests extends TestCase {
                 body.setMultipartCallback(new MultipartCallback() {
                     @Override
                     public void onPart(Part part) {
+                        Log.i("MULTIPART", part.getName());
                         if (part.isFile()) {
                             body.setDataCallback(new DataCallback() {
                                 @Override
@@ -62,6 +64,12 @@ public class MultipartTests extends TestCase {
                 request.setEndCallback(new CompletedCallback() {
                     @Override
                     public void onCompleted(Exception ex) {
+                        try {
+                            throw new Exception();
+                        }
+                        catch (Exception e) {
+                            e.printStackTrace();
+                        }
                         response.send(body.getField("baz") + gotten + body.getField("foo"));
                     }
                 });