Bug 1291821 - Wrap local repositories in buffering middleware r=rnewman

History stage does not wrap history respository in a buffer, because we'd like to
use a high-water-mark and offset resuming later on, and using a persistent buffer
for this stage does not make sense.

MozReview-Commit-ID: FS1swml2bIC

--HG--
extra : rebase_source : be197e0459d86a320076174936cea8ee76e1dbed

mobile/android/services/src/main/java/org/mozilla/gecko/sync/repositories/android/AndroidBrowserHistoryRepositorySession.java

@@ -191,6 +191,11 @@ public class AndroidBrowserHistoryRepositorySession extends AndroidBrowserReposi
 
   @Override
   public void storeDone() {
+    storeDone(System.currentTimeMillis());
+  }
+
+  @Override
+  public void storeDone(final long end) {
     storeWorkQueue.execute(new Runnable() {
       @Override
       public void run() {
@@ -201,7 +206,7 @@ public class AndroidBrowserHistoryRepositorySession extends AndroidBrowserReposi
             Logger.warn(LOG_TAG, "Error flushing records to database.", e);
           }
         }
-        storeDone(System.currentTimeMillis());
+        AndroidBrowserHistoryRepositorySession.super.storeDone(end);
       }
     });
   }

mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/AndroidBrowserBookmarksServerSyncStage.java

@@ -7,6 +7,8 @@ package org.mozilla.gecko.sync.stage;
 import java.net.URISyntaxException;
 
 import org.mozilla.gecko.sync.MetaGlobalException;
+import org.mozilla.gecko.sync.middleware.BufferingMiddlewareRepository;
+import org.mozilla.gecko.sync.middleware.storage.MemoryBufferStorage;
 import org.mozilla.gecko.sync.repositories.ConstrainedServer11Repository;
 import org.mozilla.gecko.sync.repositories.RecordFactory;
 import org.mozilla.gecko.sync.repositories.Repository;
@@ -53,7 +55,11 @@ public class AndroidBrowserBookmarksServerSyncStage extends ServerSyncStage {
 
   @Override
   protected Repository getLocalRepository() {
-    return new AndroidBrowserBookmarksRepository();
+    return new BufferingMiddlewareRepository(
+            session.getSyncDeadline(),
+            new MemoryBufferStorage(),
+            new AndroidBrowserBookmarksRepository()
+    );
   }
 
   @Override

mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/FennecTabsServerSyncStage.java

@@ -4,6 +4,8 @@
 
 package org.mozilla.gecko.sync.stage;
 
+import org.mozilla.gecko.sync.middleware.BufferingMiddlewareRepository;
+import org.mozilla.gecko.sync.middleware.storage.MemoryBufferStorage;
 import org.mozilla.gecko.sync.repositories.RecordFactory;
 import org.mozilla.gecko.sync.repositories.Repository;
 import org.mozilla.gecko.sync.repositories.android.FennecTabsRepository;
@@ -30,7 +32,11 @@ public class FennecTabsServerSyncStage extends ServerSyncStage {
 
   @Override
   protected Repository getLocalRepository() {
-    return new FennecTabsRepository(session.getClientsDelegate());
+    return new BufferingMiddlewareRepository(
+            session.getSyncDeadline(),
+            new MemoryBufferStorage(),
+            new FennecTabsRepository(session.getClientsDelegate())
+    );
   }
 
   @Override

mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/FormHistoryServerSyncStage.java

@@ -7,6 +7,8 @@ package org.mozilla.gecko.sync.stage;
 import java.net.URISyntaxException;
 
 import org.mozilla.gecko.sync.CryptoRecord;
+import org.mozilla.gecko.sync.middleware.BufferingMiddlewareRepository;
+import org.mozilla.gecko.sync.middleware.storage.MemoryBufferStorage;
 import org.mozilla.gecko.sync.repositories.ConstrainedServer11Repository;
 import org.mozilla.gecko.sync.repositories.RecordFactory;
 import org.mozilla.gecko.sync.repositories.Repository;
@@ -54,7 +56,11 @@ public class FormHistoryServerSyncStage extends ServerSyncStage {
 
   @Override
   protected Repository getLocalRepository() {
-    return new FormHistoryRepositorySession.FormHistoryRepository();
+    return new BufferingMiddlewareRepository(
+            session.getSyncDeadline(),
+            new MemoryBufferStorage(),
+            new FormHistoryRepositorySession.FormHistoryRepository()
+    );
   }
 
   public class FormHistoryRecordFactory extends RecordFactory {

mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/PasswordsServerSyncStage.java

@@ -4,6 +4,8 @@
 
 package org.mozilla.gecko.sync.stage;
 
+import org.mozilla.gecko.sync.middleware.BufferingMiddlewareRepository;
+import org.mozilla.gecko.sync.middleware.storage.MemoryBufferStorage;
 import org.mozilla.gecko.sync.repositories.RecordFactory;
 import org.mozilla.gecko.sync.repositories.Repository;
 import org.mozilla.gecko.sync.repositories.android.PasswordsRepositorySession;
@@ -28,7 +30,11 @@ public class PasswordsServerSyncStage extends ServerSyncStage {
 
   @Override
   protected Repository getLocalRepository() {
-    return new PasswordsRepositorySession.PasswordsRepository();
+    return new BufferingMiddlewareRepository(
+            session.getSyncDeadline(),
+            new MemoryBufferStorage(),
+            new PasswordsRepositorySession.PasswordsRepository()
+    );
   }
 
   @Override

mobile/android/services/src/main/java/org/mozilla/gecko/sync/synchronizer/RecordsChannel.java

@@ -170,7 +170,16 @@ public class RecordsChannel implements
     this.consumer = new ConcurrentRecordConsumer(this);
     ThreadPool.run(this.consumer);
     waitingForQueueDone = true;
-    source.fetchSince(source.getLastSyncTimestamp(), this);
+
+    // Fetch all records that were modified since our previous flow. If our previous flow succeeded,
+    // we will use source's last-sync timestamp. If our previous flow didn't complete, resume it,
+    // starting from sink's high water mark timestamp.
+    // If there was no previous flow (first sync, or data was cleared...), fetch everything.
+    // Resuming a flow is supported for buffered RepositorySessions. We degrade gracefully otherwise.
+    final long highWaterMark = sink.getHighWaterMarkTimestamp();
+    final long lastSync = source.getLastSyncTimestamp();
+    final long sinceTimestamp = Math.max(highWaterMark, lastSync);
+    source.fetchSince(sinceTimestamp, this);
   }
 
   /**