fix: Prevent watch stream from emitting events after close.

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

@@ -916,7 +916,7 @@ private void sendBatchLocked(final BulkCommitBatch batch, final boolean flush) {
               bulkWriterExecutor);
     } else {
       long delayMs = rateLimiter.getNextRequestDelayMs(batch.getMutationsSize());
-      logger.log(Level.FINE, String.format("Backing off for %d seconds", delayMs / 1000));
+      logger.log(Level.FINE, () -> String.format("Backing off for %d seconds", delayMs / 1000));
       bulkWriterExecutor.schedule(
           () -> {
             synchronized (lock) {

google-cloud-firestore/src/main/java/com/google/cloud/firestore/SuppressibleBidiStream.java

@@ -0,0 +1,85 @@
+package com.google.cloud.firestore;
+
+import com.google.api.gax.rpc.BidiStreamObserver;
+import com.google.api.gax.rpc.ClientStream;
+import com.google.api.gax.rpc.StreamController;
+import com.google.firestore.v1.ListenRequest;
+import java.util.function.Function;
+import java.util.logging.Logger;
+
+public class SuppressibleBidiStream<RequestT, ResponseT> implements BidiStreamObserver<RequestT, ResponseT> {
+
+  private final ClientStream<ListenRequest> stream;
+  private final BidiStreamObserver<RequestT, ResponseT> delegate;
+  private boolean silence = false;
+  private static final Logger LOGGER = Logger.getLogger(Watch.class.getName());
+
+  SuppressibleBidiStream(
+      BidiStreamObserver<RequestT, ResponseT> responseObserverT,
+      Function<BidiStreamObserver<RequestT, ResponseT>, ClientStream<ListenRequest>> streamSupplier
+  ) {
+    this.delegate = responseObserverT;
+    stream = streamSupplier.apply(this);
+  }
+
+  public void send(ListenRequest request) {
+    LOGGER.info(stream.toString());
+    stream.send(request);
+  }
+
+  public void close() {
+    LOGGER.info(stream.toString());
+    stream.closeSend();
+  }
+
+  public void closeAndSilence() {
+    LOGGER.info(stream.toString());
+    silence = true;
+    stream.closeSend();
+  }
+
+  @Override
+  public void onReady(ClientStream<RequestT> stream) {
+    if (silence) {
+      LOGGER.info(String.format("Silenced: %s", stream));
+    } else {
+      delegate.onReady(stream);
+    }
+  }
+
+  @Override
+  public void onStart(StreamController controller) {
+    if (silence) {
+      LOGGER.info(String.format("Silenced: %s", stream));
+    } else {
+      delegate.onStart(controller);
+    }
+  }
+
+  @Override
+  public void onResponse(ResponseT response) {
+    if (silence) {
+      LOGGER.info(String.format("Silenced: %s", stream));
+    } else {
+      delegate.onResponse(response);
+    }
+  }
+
+  @Override
+  public void onError(Throwable t) {
+    if (silence) {
+      LOGGER.info(String.format("Silenced: %s", stream));
+    } else {
+      delegate.onError(t);
+    }
+  }
+
+  @Override
+  public void onComplete() {
+    if (silence) {
+      LOGGER.info(String.format("Silenced: %s", stream));
+    } else {
+      delegate.onComplete();
+    }
+  }
+}

google-cloud-firestore/src/main/java/com/google/cloud/firestore/Watch.java

@@ -49,6 +49,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Logger;
 import javax.annotation.Nullable;
 
 /**
@@ -73,7 +74,7 @@ class Watch implements BidiStreamObserver<ListenRequest, ListenResponse> {
   private final ExponentialRetryAlgorithm backoff;
   private final Target target;
   private TimedAttemptSettings nextAttempt;
-  private ClientStream<ListenRequest> stream;
+  private SuppressibleBidiStream<ListenRequest, ListenResponse> stream;
 
   /** The sorted tree of DocumentSnapshots as sent in the last snapshot. */
   private DocumentSet documentSet;
@@ -115,6 +116,8 @@ static class ChangeSet {
     List<QueryDocumentSnapshot> updates = new ArrayList<>();
   }
 
+  private static final Logger LOGGER = Logger.getLogger(Watch.class.getName());
+
   /**
    * @param firestore The Firestore Database client.
    * @param query The query that is used to order the document snapshots returned by this watch.
@@ -246,7 +249,10 @@ && affectsTarget(change.getTargetIdsList(), WATCH_TARGET_ID)) {
         changeMap.put(ResourcePath.create(listenResponse.getDocumentRemove().getDocument()), null);
         break;
       case FILTER:
-        if (listenResponse.getFilter().getCount() != currentSize()) {
+        int filterCount = listenResponse.getFilter().getCount();
+        int currentSize = currentSize();
+        if (filterCount != currentSize) {
+          LOGGER.info(() -> String.format("filter: count mismatch filter count %d != current size %d", filterCount, currentSize));
           // We need to remove all the current results.
           resetDocs();
           // The filter didn't match, so re-issue the query.
@@ -297,7 +303,7 @@ ListenerRegistration runWatch(
           .execute(
               () -> {
                 synchronized (Watch.this) {
-                  stream.closeSend();
+                  stream.close();
                   stream = null;
                 }
               });
@@ -318,7 +324,7 @@ private void resetDocs() {
     resumeToken = null;
 
     for (DocumentSnapshot snapshot : documentSet) {
-      // Mark each document as deleted. If documents are not deleted, they  will be send again by
+      // Mark each document as deleted. If documents are not deleted, they  will be sent again by
       // the server.
       changeMap.put(snapshot.getReference().getResourcePath(), null);
     }
@@ -329,7 +335,7 @@ private void resetDocs() {
   /** Closes the stream and calls onError() if the stream is still active. */
   private void closeStream(final Throwable throwable) {
     if (stream != null) {
-      stream.closeSend();
+      stream.closeAndSilence();
       stream = null;
     }
 
@@ -371,7 +377,7 @@ private void maybeReopenStream(Throwable throwable) {
   /** Helper to restart the outgoing stream to the backend. */
   private void resetStream() {
     if (stream != null) {
-      stream.closeSend();
+      stream.closeAndSilence();
       stream = null;
     }
 
@@ -398,7 +404,10 @@ private void initStream() {
               nextAttempt = backoff.createNextAttempt(nextAttempt);
 
               Tracing.getTracer().getCurrentSpan().addAnnotation(TraceUtil.SPAN_NAME_LISTEN);
-              stream = firestore.streamRequest(Watch.this, firestore.getClient().listenCallable());
+              stream = new SuppressibleBidiStream<>(
+                  Watch.this,
+                  observer -> firestore.streamRequest(observer, firestore.getClient().listenCallable())
+              );
 
               ListenRequest.Builder request = ListenRequest.newBuilder();
               request.setDatabase(firestore.getDatabaseName());
@@ -459,6 +468,7 @@ private void pushSnapshot(final Timestamp readTime, ByteString nextResumeToken)
     if (!hasPushed || !changes.isEmpty()) {
       final QuerySnapshot querySnapshot =
           QuerySnapshot.withChanges(query, readTime, documentSet, changes);
+      LOGGER.info(querySnapshot.toString());
       userCallbackExecutor.execute(() -> listener.onEvent(querySnapshot, null));
       hasPushed = true;
     }

google-cloud-firestore/src/test/java/com/google/cloud/firestore/FirestoreSpy.java

@@ -0,0 +1,29 @@
+package com.google.cloud.firestore;
+
+import static org.mockito.Mockito.doCallRealMethod;
+
+import com.google.api.gax.rpc.BidiStreamObserver;
+import com.google.api.gax.rpc.BidiStreamingCallable;
+import com.google.firestore.v1.ListenRequest;
+import com.google.firestore.v1.ListenResponse;
+import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
+
+public final class FirestoreSpy {
+
+  public final FirestoreImpl spy;
+  public final ArgumentCaptor<BidiStreamObserver<ListenRequest, ListenResponse>> streamRequestBidiStreamObserverCaptor;
+
+  public FirestoreSpy(Firestore firestore) {
+    spy = Mockito.spy((FirestoreImpl) firestore);
+    streamRequestBidiStreamObserverCaptor = ArgumentCaptor.forClass(BidiStreamObserver.class);
+    doCallRealMethod()
+        .when(spy)
+        .streamRequest(
+          streamRequestBidiStreamObserverCaptor.capture(),
+          ArgumentMatchers.<BidiStreamingCallable>any()
+        );
+  }
+
+}

google-cloud-firestore/src/test/java/com/google/cloud/firestore/WatchTest.java

@@ -72,7 +72,9 @@
 import javax.annotation.Nullable;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.Timeout;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.ArgumentMatchers;
@@ -84,6 +86,10 @@
 
 @RunWith(MockitoJUnitRunner.class)
 public class WatchTest {
+
+  @Rule
+  public Timeout timeout = new Timeout(1, TimeUnit.SECONDS);
+
   /** The Target ID used by the Java Firestore SDK. */
   private static final int TARGET_ID = 0x1;
 

google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITBaseTest.java

@@ -18,6 +18,7 @@
 
 import com.google.cloud.firestore.Firestore;
 import com.google.cloud.firestore.FirestoreOptions;
+import com.google.cloud.firestore.FirestoreSpy;
 import com.google.common.base.Preconditions;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -30,6 +31,7 @@
 public abstract class ITBaseTest {
   private static final Logger logger = Logger.getLogger(ITBaseTest.class.getName());
   protected Firestore firestore;
+  protected FirestoreSpy firestoreSpy;
 
   @Before
   public void before() {
@@ -53,5 +55,14 @@ public void after() throws Exception {
         "Error instantiating Firestore. Check that the service account credentials were properly set.");
     firestore.close();
     firestore = null;
+    firestoreSpy = null;
+  }
+
+  public FirestoreSpy useFirestoreSpy() {
+    if (firestoreSpy == null) {
+      firestoreSpy = new FirestoreSpy(firestore);
+      firestore = firestoreSpy.spy;
+    }
+    return firestoreSpy;
   }
 }