docs: ListenerRegistration runWatch(
        final Executor userCallbackExecutor, EventListener<QuerySnapshot> listener) {
      boolean watchStarted = isActive.compareAndSet(false, true);
  
      Preconditions.checkState(watchStarted, "Can't restart an already active watch");
  
      this.userCallbackExecutor = userCallbackExecutor;
      this.listener = listener;
  
      this.stream = null;
      this.documentSet = DocumentSet.emptySet(comparator);
      this.changeMap = new HashMap<>();
      this.resumeToken = null;
      this.current = false;
  
      initStream();
  
      return new ListenerRegistration() {
        @Override
        public void remove() {
          isActive.set(false);
  
          firestore
              .getClient()
              .getExecutor()
              .execute(
                  new Runnable() {
                    @Override
                    public void run() {
                      synchronized (Watch.this) {
                        stream.onCompleted();
                        stream = null;
                      }
                    }
                  });
        }
      };
    }