def __init__(
        self,
        document_reference,
        firestore,
        target,
        comparator,
        snapshot_callback,
        document_snapshot_cls,
        document_reference_cls,
        BackgroundConsumer=None,  # FBO unit testing
        ResumableBidiRpc=None,  # FBO unit testing
    ):
        """
        Args:
            firestore:
            target:
            comparator:
            snapshot_callback: Callback method to process snapshots.
                Args:
                    docs (List(DocumentSnapshot)): A callback that returns the
                        ordered list of documents stored in this snapshot.
                    changes (List(str)): A callback that returns the list of
                        changed documents since the last snapshot delivered for
                        this watch.
                    read_time (string): The ISO 8601 time at which this
                        snapshot was obtained.

            document_snapshot_cls: instance of DocumentSnapshot
            document_reference_cls: instance of DocumentReference
        """
        self._document_reference = document_reference
        self._firestore = firestore
        self._api = firestore._firestore_api
        self._targets = target
        self._comparator = comparator
        self.DocumentSnapshot = document_snapshot_cls
        self.DocumentReference = document_reference_cls
        self._snapshot_callback = snapshot_callback
        self._closing = threading.Lock()
        self._closed = False

        def should_recover(exc):  # pragma: NO COVER
            return (
                isinstance(exc, grpc.RpcError)
                and exc.code() == grpc.StatusCode.UNAVAILABLE
            )

        initial_request = firestore_pb2.ListenRequest(
            database=self._firestore._database_string, add_target=self._targets
        )

        if ResumableBidiRpc is None:
            ResumableBidiRpc = self.ResumableBidiRpc  # FBO unit tests

        self._rpc = ResumableBidiRpc(
            self._api.transport.listen,
            initial_request=initial_request,
            should_recover=should_recover,
            metadata=self._firestore._rpc_metadata,
        )

        self._rpc.add_done_callback(self._on_rpc_done)

        # Initialize state for on_snapshot
        # The sorted tree of QueryDocumentSnapshots as sent in the last
        # snapshot. We only look at the keys.
        self.doc_tree = WatchDocTree()

        # A map of document names to QueryDocumentSnapshots for the last sent
        # snapshot.
        self.doc_map = {}

        # The accumulates map of document changes (keyed by document name) for
        # the current snapshot.
        self.change_map = {}

        # The current state of the query results.
        self.current = False

        # We need this to track whether we've pushed an initial set of changes,
        # since we should push those even when there are no changes, if there
        # aren't docs.
        self.has_pushed = False

        # The server assigns and updates the resume token.
        self.resume_token = None
        if BackgroundConsumer is None:  # FBO unit tests
            BackgroundConsumer = self.BackgroundConsumer

        self._consumer = BackgroundConsumer(self._rpc, self.on_snapshot)
        self._consumer.start()
def __init__ ( self , document_reference , firestore , target , comparator , snapshot_callback , document_snapshot_cls , document_reference_cls , BackgroundConsumer = None , ResumableBidiRpc = None , ) : self . _document_reference = document_reference self . _firestore = firestore self . _api = firestore . _firestore_api self . _targets = target self . _comparator = comparator self . DocumentSnapshot = document_snapshot_cls self . DocumentReference = document_reference_cls self . _snapshot_callback = snapshot_callback self . _closing = threading . Lock ( ) self . _closed = False def should_recover ( exc ) : return ( isinstance ( exc , grpc . RpcError ) and exc . code ( ) == grpc . StatusCode . UNAVAILABLE ) initial_request = firestore_pb2 . ListenRequest ( database = self . _firestore . _database_string , add_target = self . _targets ) if ResumableBidiRpc is None : ResumableBidiRpc = self . ResumableBidiRpc self . _rpc = ResumableBidiRpc ( self . _api . transport . listen , initial_request = initial_request , should_recover = should_recover , metadata = self . _firestore . _rpc_metadata , ) self . _rpc . add_done_callback ( self . _on_rpc_done ) self . doc_tree = WatchDocTree ( ) self . doc_map = { } self . change_map = { } self . current = False self . has_pushed = False self . resume_token = None if BackgroundConsumer is None : BackgroundConsumer = self . BackgroundConsumer self . _consumer = BackgroundConsumer ( self . _rpc , self . on_snapshot ) self . _consumer . start ( )
Args:
            firestore:
            target:
            comparator:
            snapshot_callback: Callback method to process snapshots.
                Args:
                    docs (List(DocumentSnapshot)): A callback that returns the
                        ordered list of documents stored in this snapshot.
                    changes (List(str)): A callback that returns the list of
                        changed documents since the last snapshot delivered for
                        this watch.
                    read_time (string): The ISO 8601 time at which this
                        snapshot was obtained.

            document_snapshot_cls: instance of DocumentSnapshot
            document_reference_cls: instance of DocumentReference