Merge pull request #1583 from couchbase/fix/1581

Fixed #1581 - Replication failures when running Sync Gateway in distr…

Author: Hideki Itakura

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

@@ -476,7 +476,7 @@ private void queueDownloadedRevision(RevisionInternal rev) {
         // if queue memory size is more than maximum, force flush the queue.
         if (queuedMemorySize.get() > MAX_QUEUE_MEMORY_SIZE) {
             Log.d(TAG, "Flushing queued memory size at: " + queuedMemorySize);
-            downloadsToInsert.flushAll(true);
+            downloadsToInsert.flushAllAndWait();
         }
     }
 
@@ -744,7 +744,7 @@ public void onCompletion(Response httpResponse, Object result, Throwable e) {
                             // if queue memory size is more than maximum, force flush the queue.
                             if (queuedMemorySize.get() > MAX_QUEUE_MEMORY_SIZE) {
                                 Log.d(TAG, "Flushing  queued memory size at: " + queuedMemorySize);
-                                downloadsToInsert.flushAll(true);
+                                downloadsToInsert.flushAllAndWait();
                             }
                         }
 

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

@@ -53,6 +53,7 @@ public class Batcher<T> {
 
     private final Object mutex = new Object();
     private final Object processMutex = new Object();
+    private final Object flushAllMutext = new Object();
 
     ///////////////////////////////////////////////////////////////////////////
     // Constructors
@@ -161,6 +162,12 @@ public void queueObjects(List<T> objects) {
         }
     }
 
+    public void flushAllAndWait() {
+        synchronized (flushAllMutext) {
+            flushAll(true);
+        }
+    }
+
     /**
      * Sends _all_ the queued objects at once to the processor block.
      * After this method returns, all inbox objects will be processed.