By extending Future, Future is allowed to remove itself from pendingFutures queue.

Problem:
- pendingFutures keeps the references of RemoteRequestRetry which is extended from Future. Each RemoteRequestRetry size after downloaded document is around 1MB. By storing 100 RemoteRequestRetry instances into pendingFutures causes out-of-memory problem.

Author: Hideki Itakura

src/main/java/com/couchbase/lite/replicator/PullerInternal.java

@@ -11,6 +11,7 @@
 import com.couchbase.lite.storage.SQLException;
 import com.couchbase.lite.support.BatchProcessor;
 import com.couchbase.lite.support.Batcher;
+import com.couchbase.lite.support.CustomFuture;
 import com.couchbase.lite.support.HttpClientFactory;
 import com.couchbase.lite.support.RemoteRequestCompletionBlock;
 import com.couchbase.lite.support.SequenceMap;
@@ -609,7 +610,7 @@ public void pullRemoteRevision(final RevisionInternal rev) {
         //create a final version of this variable for the log statement inside
         //FIXME find a way to avoid this
         final String pathInside = path.toString();
-        Future future = sendAsyncMultipartDownloaderRequest("GET", pathInside, null, db, new RemoteRequestCompletionBlock() {
+        CustomFuture future = sendAsyncMultipartDownloaderRequest("GET", pathInside, null, db, new RemoteRequestCompletionBlock() {
 
             @Override
             public void onCompletion(HttpResponse httpResponse, Object result, Throwable e) {
@@ -632,10 +633,12 @@ public void onCompletion(HttpResponse httpResponse, Object result, Throwable e)
 
                 // Note that we've finished this task:
                 --httpConnectionCount;
+
                 // Start another task if there are still revisions waiting to be pulled:
                 pullRemoteRevisions();
             }
         });
+        future.setQueue(pendingFutures);
         pendingFutures.add(future);
     }
 

src/main/java/com/couchbase/lite/replicator/ReplicationInternal.java

@@ -11,6 +11,7 @@
 import com.couchbase.lite.support.BatchProcessor;
 import com.couchbase.lite.support.Batcher;
 import com.couchbase.lite.support.BlockingQueueListener;
+import com.couchbase.lite.support.CustomFuture;
 import com.couchbase.lite.support.CustomLinkedBlockingQueue;
 import com.couchbase.lite.support.HttpClientFactory;
 import com.couchbase.lite.support.RemoteRequestCompletionBlock;
@@ -596,7 +597,7 @@ public Future sendAsyncMultipartRequest(String method, String relativePath, Mult
      * @exclude
      */
     @InterfaceAudience.Private
-    public Future sendAsyncMultipartDownloaderRequest(String method, String relativePath, Object body, Database db, RemoteRequestCompletionBlock onCompletion) {
+    public CustomFuture sendAsyncMultipartDownloaderRequest(String method, String relativePath, Object body, Database db, RemoteRequestCompletionBlock onCompletion) {
         try {
 
             String urlStr = buildRelativeURLString(relativePath);
@@ -617,7 +618,7 @@ public Future sendAsyncMultipartDownloaderRequest(String method, String relative
 
             request.setAuthenticator(getAuthenticator());
 
-            Future future = request.submit();
+            CustomFuture future = request.submit();
             return future;
         } catch (MalformedURLException e) {
             Log.e(Log.TAG_SYNC, "Malformed URL for async request", e);

src/main/java/com/couchbase/lite/support/CustomFuture.java

@@ -0,0 +1,11 @@
+package com.couchbase.lite.support;
+
+import java.util.Queue;
+import java.util.concurrent.Future;
+
+/**
+ * Created by hideki on 5/21/15.
+ */
+public interface CustomFuture<V> extends Future<V> {
+    void setQueue(Queue queue);
+}

src/main/java/com/couchbase/lite/support/RemoteRequestRetry.java

@@ -12,6 +12,7 @@
 import java.io.IOException;
 import java.net.URL;
 import java.util.Map;
+import java.util.Queue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -31,7 +32,7 @@
  * in between the retries.
  *
  */
-public class RemoteRequestRetry<T> implements Future<T> {
+public class RemoteRequestRetry<T> implements CustomFuture<T> {
 
     public static int MAX_RETRIES = 3;  // total number of attempts = 4 (1 initial + MAX_RETRIES)
     public static int RETRY_DELAY_MS = 4 * 1000; // 4 sec
@@ -64,9 +65,18 @@ public class RemoteRequestRetry<T> implements Future<T> {
 
     private RemoteRequestType requestType;
 
+
     // for Retry task
     ScheduledFuture retryFuture = null;
 
+    private Queue queue = null;
+
+    @Override
+    public void setQueue(Queue queue) {
+        this.queue = queue;
+    }
+
+
     /**
      * The kind of RemoteRequest that will be created on each retry attempt
      */
@@ -105,14 +115,14 @@ public RemoteRequestRetry(RemoteRequestType requestType,
 
     }
 
-    public Future submit() {
+    public CustomFuture submit() {
         return submit(false);
     }
 
     /**
      * @param gzip true - send gzipped request
      */
-    public Future submit(boolean gzip) {
+    public CustomFuture submit(boolean gzip) {
 
         RemoteRequest request = generateRemoteRequest();
 
@@ -185,6 +195,13 @@ private RemoteRequest generateRemoteRequest() {
         return request;
     }
 
+    void removeFromQueue() {
+        if (queue != null) {
+            queue.remove(this);
+            setQueue(null);
+        }
+    }
+
     RemoteRequestCompletionBlock onCompletionInner = new RemoteRequestCompletionBlock() {
 
         private void completed(HttpResponse httpResponse, Object result, Throwable e) {
@@ -194,11 +211,13 @@ private void completed(HttpResponse httpResponse, Object result, Throwable e) {
             completedSuccessfully.set(true);
 
             onCompletionCaller.onCompletion(requestHttpResponse, requestResult, requestThrowable);
-
+            
             // release unnecessary references to reduce memory usage as soon as called onComplete().
             requestHttpResponse = null;
             requestResult = null;
             requestThrowable = null;
+
+            removeFromQueue();
         }
 
         @Override