Fix crash bug when cancelling requests being cached.

koush/ion#252 (comment)

Author: koush

AndroidAsync/src/com/koushikdutta/async/FilteredDataEmitter.java

@@ -45,6 +45,12 @@ public void setDataTracker(DataTracker tracker) {
     int totalRead;
     @Override
     public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
+        if (closed) {
+            // this emitter was closed but for some reason data is still being spewed...
+            // eat it, nom nom.
+            bb.recycle();
+            return;
+        }
         if (bb != null)
             totalRead += bb.remaining();
         Util.emitAllData(this, bb);
@@ -81,9 +87,12 @@ public AsyncServer getServer() {
         return mEmitter.getServer();
     }
 
+    boolean closed;
     @Override
     public void close() {
-        mEmitter.close();
+        closed = true;
+        if (mEmitter != null)
+            mEmitter.close();
     }
 
     @Override

AndroidAsync/src/com/koushikdutta/async/http/ResponseCacheMiddleware.java

@@ -325,7 +325,7 @@ protected void report(Exception e) {
         @Override
         public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
             if (cached != null) {
-                com.koushikdutta.async.Util.emitAllData(this, cached);
+                super.onDataAvailable(emitter, cached);
                 // couldn't emit it all, so just wait for another day...
                 if (cached.remaining() > 0)
                     return;
@@ -369,6 +369,12 @@ public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
             }
         }
 
+        @Override
+        public void close() {
+            abort();
+            super.close();
+        }
+
         public void abort() {
             if (editor != null) {
                 editor.abort();

AndroidAsync/test/src/com/koushikdutta/async/test/CacheTests.java

@@ -4,6 +4,11 @@
 
 import com.koushikdutta.async.AsyncServer;
 import com.koushikdutta.async.AsyncServerSocket;
+import com.koushikdutta.async.ByteBufferList;
+import com.koushikdutta.async.DataEmitter;
+import com.koushikdutta.async.FilteredDataEmitter;
+import com.koushikdutta.async.Util;
+import com.koushikdutta.async.callback.DataCallback;
 import com.koushikdutta.async.http.AsyncHttpClient;
 import com.koushikdutta.async.http.AsyncHttpGet;
 import com.koushikdutta.async.http.ResponseCacheMiddleware;
@@ -16,7 +21,10 @@
 import junit.framework.TestCase;
 
 import java.io.File;
+import java.nio.ByteBuffer;
 import java.util.Date;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Created by koush on 6/13/13.
@@ -55,29 +63,37 @@ public void onRequest(AsyncHttpServerRequest request, AsyncHttpServerResponse re
         }
     }
 
-//    static public boolean deleteDirectory(File path) {
-//        if (path.exists()) {
-//            File[] files = path.listFiles();
-//            if (files != null) {
-//                for (int i = 0; i < files.length; i++) {
-//                    if (files[i].isDirectory()) {
-//                        deleteDirectory(files[i]);
-//                    } else {
-//                        files[i].delete();
-//                    }
-//                }
-//            }
-//        }
-//        return (path.delete());
-//    }
-
-//    public void testDiskLruCache() throws Exception {
-//        File dir = new File(Environment.getExternalStorageDirectory(), "AndroidAsyncTest/cache-test");
-//        deleteDirectory(dir);
-//        DiskLruCache cache = DiskLruCache.open(dir, 0, 1000, 10000000);
-//        DiskLruCache.Editor editor = cache.edit("stuff");
-//
-//        DiskLruCache cache2 = DiskLruCache.open(dir, 0, 2, 10000000);
-//        DiskLruCache.Snapshot snapshot = cache2.get("stuff");
-//    }
+    private static final long TIMEOUT = 1000L;
+    public void testFilteredDataEmitter() throws Exception {
+        final Semaphore semaphore = new Semaphore(0);
+
+        FilteredDataEmitter f = new FilteredDataEmitter() {
+            @Override
+            public boolean isPaused() {
+                return false;
+            }
+        };
+
+        f.setDataCallback(new DataCallback() {
+            @Override
+            public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
+                assertEquals(bb.readString(), "hello");
+                bb.recycle();
+                semaphore.release();
+            }
+        });
+
+        f.onDataAvailable(f, new ByteBufferList().add(ByteBuffer.wrap("hello".getBytes())));
+        assertTrue("timeout", semaphore.tryAcquire(TIMEOUT, TimeUnit.MILLISECONDS));
+
+        f.setDataCallback(new DataCallback() {
+            @Override
+            public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
+                fail();
+            }
+        });
+        f.close();
+
+        f.onDataAvailable(f, new ByteBufferList().add(ByteBuffer.wrap("hello".getBytes())));
+    }
 }