Abort snapshot listeners on terminate

firebase-firestore/CHANGELOG.md

@@ -1,5 +1,5 @@
 # Unreleased
-
+* [changed] Fail snapshot listeners when Firestore terminates. [#6136](//github.com/firebase/firebase-android-sdk/pull/6136)
 
 # 25.1.0
 * [feature] Add support for the VectorValue type. [#6154](//github.com/firebase/firebase-android-sdk/pull/6154)

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/FirestoreTest.java

@@ -1263,6 +1263,8 @@ public void testCanStopListeningAfterTerminate() {
 
  waitFor(instance.terminate());
 
+ assertEquals(eventAccumulator.awaitError().getCode(), Code.ABORTED);
+
  // This should proceed without error.
  registration.remove();
  // Multiple calls should proceed as an effectively no-op.

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/ListenerRegistrationTest.java

@@ -165,12 +165,13 @@ private void activityScopedListenerStopsListeningWhenActivityStops(Activity acti
  DocumentReference documentReference = collectionReference.document();
 
  Semaphore events = new Semaphore(0);
- collectionReference.addSnapshotListener(
- activity,
- (value, error) -> {
- assertNull(error);
- events.release();
- });
+ ListenerRegistration listener =
+ collectionReference.addSnapshotListener(
+ activity,
+ (value, error) -> {
+ assertNull(error);
+ events.release();
+ });
 
  // Initial events
  waitFor(events, 1);
@@ -188,6 +189,8 @@ private void activityScopedListenerStopsListeningWhenActivityStops(Activity acti
  // No listeners, therefore, there should be no events.
  waitFor(documentReference.set(map("foo", "new-bar")));
  assertEquals(0, events.availablePermits());
+
+ listener.remove();
  }
 
  /** @param activity Must be a TestActivity or a TestFragmentActivity */

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/QueryTest.java

@@ -147,7 +147,7 @@ public void testListenUnlistenRelistenSequenceOfMirrorQueries() {
 
  // Unlisten then re-listen limit query.
  limitRegistration.remove();
- limit.addSnapshotListener(limitAccumulator.listener());
+ limitRegistration = limit.addSnapshotListener(limitAccumulator.listener());
 
  // Verify `limit` query still works.
  data = querySnapshotToValues(limitAccumulator.await());
@@ -165,13 +165,16 @@ public void testListenUnlistenRelistenSequenceOfMirrorQueries() {
  // Unlisten to limitToLast, update a doc, then relisten to limitToLast
  limitToLastRegistration.remove();
  waitFor(collection.document("a").update(map("k", "a", "sort", -2)));
- limitToLast.addSnapshotListener(limitToLastAccumulator.listener());
+ limitToLastRegistration = limitToLast.addSnapshotListener(limitToLastAccumulator.listener());
 
  // Verify both query get expected result.
  data = querySnapshotToValues(limitAccumulator.await());
  assertEquals(asList(map("k", "a", "sort", -2L), map("k", "e", "sort", -1L)), data);
  data = querySnapshotToValues(limitToLastAccumulator.await());
  assertEquals(asList(map("k", "e", "sort", -1L), map("k", "a", "sort", -2L)), data);
+
+ limitRegistration.remove();
+ limitToLastRegistration.remove();
  }
 
  @Test
@@ -507,19 +510,22 @@ public void watchSurvivesNetworkDisconnect() {
 
  Semaphore receivedDocument = new Semaphore(0);
 
- collectionReference.addSnapshotListener(
- MetadataChanges.INCLUDE,
- (snapshot, error) -> {
- if (!snapshot.isEmpty() && !snapshot.getMetadata().isFromCache()) {
- receivedDocument.release();
- }
- });
+ ListenerRegistration listener =
+ collectionReference.addSnapshotListener(
+ MetadataChanges.INCLUDE,
+ (snapshot, error) -> {
+ if (!snapshot.isEmpty() && !snapshot.getMetadata().isFromCache()) {
+ receivedDocument.release();
+ }
+ });
 
  waitFor(firestore.disableNetwork());
  collectionReference.add(map("foo", FieldValue.serverTimestamp()));
  waitFor(firestore.enableNetwork());
 
  waitFor(receivedDocument);
+
+ listener.remove();
  }
 
  @Test

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/SourceTest.java

@@ -289,15 +289,17 @@ public void getDocumentWhileOfflineWithDifferentGetOptions() {
  // Create an initial listener for this query (to attempt to disrupt the gets below) and wait for
  // the listener to deliver its initial snapshot before continuing.
  TaskCompletionSource<Void> source = new TaskCompletionSource<>();
- docRef.addSnapshotListener(
- (docSnap, error) -> {
- if (error != null) {
- source.setException(error);
- } else {
- source.setResult(null);
- }
- });
+ ListenerRegistration listener =
+ docRef.addSnapshotListener(
+ (docSnap, error) -> {
+ if (error != null) {
+ source.setException(error);
+ } else {
+ source.setResult(null);
+ }
+ });
  waitFor(source.getTask());
+ listener.remove();
 
  Task<DocumentSnapshot> docTask = docRef.get(Source.CACHE);
  waitFor(docTask);
@@ -339,15 +341,17 @@ public void getCollectionWhileOfflineWithDifferentGetOptions() {
  // Create an initial listener for this query (to attempt to disrupt the gets below) and wait for
  // the listener to deliver its initial snapshot before continuing.
  TaskCompletionSource<Void> source = new TaskCompletionSource<>();
- colRef.addSnapshotListener(
- (qrySnap, error) -> {
- if (error != null) {
- source.setException(error);
- } else {
- source.setResult(null);
- }
- });
+ ListenerRegistration listener =
+ colRef.addSnapshotListener(
+ (qrySnap, error) -> {
+ if (error != null) {
+ source.setException(error);
+ } else {
+ source.setResult(null);
+ }
+ });
  waitFor(source.getTask());
+ listener.remove();
 
  Task<QuerySnapshot> qrySnapTask = colRef.get(Source.CACHE);
  waitFor(qrySnapTask);

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/testutil/EventAccumulator.java

@@ -19,6 +19,7 @@
 
 import com.google.firebase.firestore.DocumentSnapshot;
 import com.google.firebase.firestore.EventListener;
+import com.google.firebase.firestore.FirebaseFirestoreException;
 import com.google.firebase.firestore.QuerySnapshot;
 import com.google.firebase.firestore.util.Logger;
 import java.util.ArrayList;
@@ -30,28 +31,45 @@
 public class EventAccumulator<T> {
  private static final int MAX_EVENTS = 10;
 
- private final BlockingQueue<T> events;
+ private final BlockingQueue<Object> events;
  private boolean rejectAdditionalEvents;
 
  public EventAccumulator() {
- events = new ArrayBlockingQueue<T>(MAX_EVENTS);
+ events = new ArrayBlockingQueue<>(MAX_EVENTS);
  }
 
  public EventListener<T> listener() {
  return (value, error) -> {
- hardAssert(error == null, "Unexpected error: %s", error);
  hardAssert(
  !rejectAdditionalEvents, "Received event after `assertNoAdditionalEvents()` was called");
- Logger.debug("EventAccumulator", "Received new event: " + value);
- events.offer(value);
+ if (error == null) {
+ Logger.debug("EventAccumulator", "Received new event: " + value);
+ events.add(value);
+ } else {
+ Logger.debug("EventAccumulator", "Received error: " + error);
+ events.add(error);
+ }
  };
  }
 
+ public FirebaseFirestoreException awaitError() {
+ try {
+ return (FirebaseFirestoreException) events.take();
+ } catch (Exception e) {
+ Logger.debug("EventAccumulator", e.toString());
+ throw fail("Failed to receive error");
+ }
+ }
+
  public List<T> await(int numEvents) {
  try {
  List<T> result = new ArrayList<>(numEvents);
  for (int i = 0; i < numEvents; ++i) {
- result.add(events.take());
+ Object event = events.take();
+ if (event instanceof FirebaseFirestoreException) {
+ fail("Unexpected error: %s", event);
+ }
+ result.add((T) event);
  }
  return result;
  } catch (InterruptedException e) {

firebase-firestore/src/main/java/com/google/firebase/firestore/FirebaseFirestore.java

@@ -581,8 +581,8 @@ public Task<Void> runBatch(@NonNull WriteBatch.Function batchFunction) {
  * <p>To restart after termination, simply create a new instance of {@code FirebaseFirestore} with
  * {@link #getInstance()} or {@link #getInstance(FirebaseApp)}.
  *
- * <p>{@code terminate()} does not cancel any pending writes and any tasks that are awaiting a
- * response from the server will not be resolved. The next time you start this instance, it will
+ * <p>{@code terminate()} does not cancel any pending writes but any write tasks that are awaiting
+ * a response from the server will not be resolved. The next time you start this instance, it will
  * resume attempting to send these writes to the server.
  *
  * <p>Note: Under normal circumstances, calling {@code terminate()} is not required. This method

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

@@ -17,6 +17,7 @@
 import static com.google.firebase.firestore.util.Assert.hardAssert;
 
 import com.google.firebase.firestore.EventListener;
+import com.google.firebase.firestore.FirebaseFirestoreException;
 import com.google.firebase.firestore.ListenSource;
 import com.google.firebase.firestore.core.SyncEngine.SyncEngineCallback;
 import com.google.firebase.firestore.util.Util;
@@ -265,4 +266,14 @@ public void handleOnlineStateChange(OnlineState onlineState) {
  raiseSnapshotsInSyncEvent();
  }
  }
+
+ public void abortAllTargets() {
+ FirebaseFirestoreException error = Util.exceptionFromStatus(Status.ABORTED);
+ for (QueryListenersInfo info : queries.values()) {
+ for (QueryListener listener : info.listeners) {
+ listener.onError(error);
+ }
+ }
+ queries.clear();
+ }
 }

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

@@ -148,6 +148,7 @@ public Task<Void> enableNetwork() {
  public Task<Void> terminate() {
  authProvider.removeChangeListener();
  appCheckProvider.removeChangeListener();
+ asyncQueue.enqueueAndForgetEvenAfterShutdown(() -> eventManager.abortAllTargets());
  return asyncQueue.enqueueAndInitiateShutdown(
  () -> {
  remoteStore.shutdown();