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 ; } } } ) ; } } ; }
API entry point that starts the Watch stream .