Allow firestore to recover quicker when a network change occurs

firebase-firestore/CHANGELOG.md

@@ -1,6 +1,10 @@
 # Unreleased
+
+# 18.0.1
 - [fixed] Fixed an issue where Firestore would crash if handling write batches
   larger than 2 MB in size (#208).
+- [changed] Firestore now recovers more quickly from long periods without
+  network access (#217).
 
 # 18.0.0
 - [changed] The `timestampsInSnapshotsEnabled` setting is now enabled by

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/remote/RemoteStoreTest.java

@@ -0,0 +1,116 @@
+// Copyright 2019 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.firestore.remote;
+
+import static com.google.firebase.firestore.testutil.IntegrationTestUtil.waitFor;
+
+import android.support.test.InstrumentationRegistry;
+import android.support.test.runner.AndroidJUnit4;
+import com.google.firebase.database.collection.ImmutableSortedSet;
+import com.google.firebase.firestore.auth.User;
+import com.google.firebase.firestore.core.OnlineState;
+import com.google.firebase.firestore.local.LocalStore;
+import com.google.firebase.firestore.local.MemoryPersistence;
+import com.google.firebase.firestore.local.Persistence;
+import com.google.firebase.firestore.model.DocumentKey;
+import com.google.firebase.firestore.model.mutation.MutationBatchResult;
+import com.google.firebase.firestore.testutil.IntegrationTestUtil;
+import com.google.firebase.firestore.util.AsyncQueue;
+import com.google.firebase.firestore.util.Consumer;
+import io.grpc.Status;
+import java.util.concurrent.Semaphore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(AndroidJUnit4.class)
+public class RemoteStoreTest {
+  @Test
+  public void testRemoteStoreStreamStopsWhenNetworkUnreachable() {
+    AsyncQueue testQueue = new AsyncQueue();
+    Datastore datastore =
+        new Datastore(
+            IntegrationTestUtil.testEnvDatabaseInfo(),
+            testQueue,
+            null,
+            InstrumentationRegistry.getContext());
+    Semaphore networkChangeSemaphore = new Semaphore(0);
+    RemoteStore.RemoteStoreCallback callback =
+        new RemoteStore.RemoteStoreCallback() {
+          public void handleRemoteEvent(RemoteEvent remoteEvent) {}
+
+          public void handleRejectedListen(int targetId, Status error) {}
+
+          public void handleSuccessfulWrite(MutationBatchResult successfulWrite) {}
+
+          public void handleRejectedWrite(int batchId, Status error) {}
+
+          public void handleOnlineStateChange(OnlineState onlineState) {
+            networkChangeSemaphore.release();
+          }
+
+          public ImmutableSortedSet<DocumentKey> getRemoteKeysForTarget(int targetId) {
+            return null;
+          }
+        };
+
+    FakeConnectivityMonitor connectivityMonitor = new FakeConnectivityMonitor();
+    Persistence persistence = MemoryPersistence.createEagerGcMemoryPersistence();
+    persistence.start();
+    LocalStore localStore = new LocalStore(persistence, User.UNAUTHENTICATED);
+    RemoteStore remoteStore =
+        new RemoteStore(callback, localStore, datastore, testQueue, connectivityMonitor);
+
+    waitFor(testQueue.enqueue(() -> remoteStore.forceEnableNetwork()));
+    drain(testQueue);
+    networkChangeSemaphore.drainPermits();
+
+    connectivityMonitor.goOffline();
+    waitFor(networkChangeSemaphore);
+    drain(testQueue);
+
+    waitFor(testQueue.enqueue(() -> remoteStore.forceEnableNetwork()));
+    networkChangeSemaphore.drainPermits();
+    connectivityMonitor.goOnline();
+    waitFor(networkChangeSemaphore);
+  }
+
+  private void drain(AsyncQueue testQueue) {
+    waitFor(testQueue.enqueue(() -> {}));
+  }
+
+  class FakeConnectivityMonitor implements ConnectivityMonitor {
+    private Consumer<NetworkStatus> callback = null;
+
+    @Override
+    public void addCallback(Consumer<NetworkStatus> callback) {
+      this.callback = callback;
+    }
+
+    @Override
+    public void shutdown() {}
+
+    public void goOffline() {
+      if (callback != null) {
+        callback.accept(ConnectivityMonitor.NetworkStatus.UNREACHABLE);
+      }
+    }
+
+    public void goOnline() {
+      if (callback != null) {
+        callback.accept(ConnectivityMonitor.NetworkStatus.REACHABLE);
+      }
+    }
+  }
+}

firebase-firestore/src/main/java/com/google/firebase/firestore/core/FirestoreClient.java

@@ -44,6 +44,8 @@
 import com.google.firebase.firestore.model.NoDocument;
 import com.google.firebase.firestore.model.mutation.Mutation;
 import com.google.firebase.firestore.model.mutation.MutationBatchResult;
+import com.google.firebase.firestore.remote.AndroidConnectivityMonitor;
+import com.google.firebase.firestore.remote.ConnectivityMonitor;
 import com.google.firebase.firestore.remote.Datastore;
 import com.google.firebase.firestore.remote.RemoteEvent;
 import com.google.firebase.firestore.remote.RemoteSerializer;
@@ -241,7 +243,8 @@ private void initialize(Context context, User user, boolean usePersistence, long
     }
 
     Datastore datastore = new Datastore(databaseInfo, asyncQueue, credentialsProvider, context);
-    remoteStore = new RemoteStore(this, localStore, datastore, asyncQueue);
+    ConnectivityMonitor connectivityMonitor = new AndroidConnectivityMonitor(context);
+    remoteStore = new RemoteStore(this, localStore, datastore, asyncQueue, connectivityMonitor);
 
     syncEngine = new SyncEngine(localStore, remoteStore, user);
     eventManager = new EventManager(syncEngine);

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/AndroidConnectivityMonitor.java

@@ -0,0 +1,137 @@
+// Copyright 2019 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.firestore.remote;
+
+import static com.google.firebase.firestore.util.Assert.hardAssert;
+
+import android.annotation.TargetApi;
+import android.content.BroadcastReceiver;
+import android.content.Context;
+import android.content.Intent;
+import android.content.IntentFilter;
+import android.net.ConnectivityManager;
+import android.net.Network;
+import android.net.NetworkInfo;
+import android.os.Build;
+import com.google.firebase.firestore.util.Consumer;
+import java.util.ArrayList;
+import java.util.List;
+import javax.annotation.Nullable;
+
+/**
+ * Android implementation of ConnectivityMonitor. Parallel implementations exist for N+ and pre-N.
+ *
+ * <p>Implementation note: Most of the code here was shamelessly stolen from
+ * https://github.com/grpc/grpc-java/blob/master/android/src/main/java/io/grpc/android/AndroidChannelBuilder.java
+ */
+public final class AndroidConnectivityMonitor implements ConnectivityMonitor {
+
+  private final Context context;
+  @Nullable private final ConnectivityManager connectivityManager;
+  @Nullable private Runnable unregisterRunnable;
+  private final List<Consumer<NetworkStatus>> callbacks = new ArrayList<>();
+
+  public AndroidConnectivityMonitor(Context context) {
+    // This notnull restriction could be eliminated... the pre-N method doesn't
+    // require a Context, and we could use that even on N+ if necessary.
+    hardAssert(context != null, "Context must be non-null");
+    this.context = context;
+
+    connectivityManager =
+        (ConnectivityManager) context.getSystemService(Context.CONNECTIVITY_SERVICE);
+    configureNetworkMonitoring();
+  }
+
+  @Override
+  public void addCallback(Consumer<NetworkStatus> callback) {
+    callbacks.add(callback);
+  }
+
+  @Override
+  public void shutdown() {
+    if (unregisterRunnable != null) {
+      unregisterRunnable.run();
+      unregisterRunnable = null;
+    }
+  }
+
+  private void configureNetworkMonitoring() {
+    // Android N added the registerDefaultNetworkCallback API to listen to changes in the device's
+    // default network. For earlier Android API levels, use the BroadcastReceiver API.
+    if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N && connectivityManager != null) {
+      final DefaultNetworkCallback defaultNetworkCallback = new DefaultNetworkCallback();
+      connectivityManager.registerDefaultNetworkCallback(defaultNetworkCallback);
+      unregisterRunnable =
+          new Runnable() {
+            @Override
+            public void run() {
+              connectivityManager.unregisterNetworkCallback(defaultNetworkCallback);
+            }
+          };
+    } else {
+      NetworkReceiver networkReceiver = new NetworkReceiver();
+      IntentFilter networkIntentFilter = new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION);
+      context.registerReceiver(networkReceiver, networkIntentFilter);
+      unregisterRunnable =
+          new Runnable() {
+            @Override
+            public void run() {
+              context.unregisterReceiver(networkReceiver);
+            }
+          };
+    }
+  }
+
+  /** Respond to changes in the default network. Only used on API levels 24+. */
+  @TargetApi(Build.VERSION_CODES.N)
+  private class DefaultNetworkCallback extends ConnectivityManager.NetworkCallback {
+    @Override
+    public void onAvailable(Network network) {
+      for (Consumer<NetworkStatus> callback : callbacks) {
+        callback.accept(NetworkStatus.REACHABLE);
+      }
+    }
+
+    @Override
+    public void onLost(Network network) {
+      for (Consumer<NetworkStatus> callback : callbacks) {
+        callback.accept(NetworkStatus.UNREACHABLE);
+      }
+    }
+  }
+
+  /** Respond to network changes. Only used on API levels < 24. */
+  private class NetworkReceiver extends BroadcastReceiver {
+    private boolean isConnected = false;
+
+    @Override
+    public void onReceive(Context context, Intent intent) {
+      ConnectivityManager conn =
+          (ConnectivityManager) context.getSystemService(Context.CONNECTIVITY_SERVICE);
+      NetworkInfo networkInfo = conn.getActiveNetworkInfo();
+      boolean wasConnected = isConnected;
+      isConnected = networkInfo != null && networkInfo.isConnected();
+      if (isConnected && !wasConnected) {
+        for (Consumer<NetworkStatus> callback : callbacks) {
+          callback.accept(NetworkStatus.REACHABLE);
+        }
+      } else if (!isConnected && wasConnected) {
+        for (Consumer<NetworkStatus> callback : callbacks) {
+          callback.accept(NetworkStatus.UNREACHABLE);
+        }
+      }
+    }
+  }
+}

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/ConnectivityMonitor.java

@@ -0,0 +1,39 @@
+// Copyright 2019 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.firestore.remote;
+
+import com.google.firebase.firestore.util.Consumer;
+
+/** Interface for monitoring changes in network connectivity/reachability. */
+public interface ConnectivityMonitor {
+  enum NetworkStatus {
+    UNREACHABLE,
+    REACHABLE,
+    // TODO(rsgowman): REACHABLE_VIA_CELLULAR.
+    // Leaving this off for now, since (a) we don't need it, and (b) it's somewhat messy to
+    // determine, and (c) we need two parallel implementations (for N+ and pre-N).
+  };
+
+  // TODO(rsgowman): Skipping isNetworkReachable() until we need it.
+  // boolean isNetworkReachable();
+
+  void addCallback(Consumer<NetworkStatus> callback);
+
+  /**
+   * Stops monitoring connectivity. After this call completes, no further callbacks will be
+   * triggered. After shutdown() is called, no further calls are allowed on this instance.
+   */
+  void shutdown();
+}