Called everytime there is a response from listen. Collect changes
        and 'push' the changes in a batch to the customer when we receive
        'current' from the listen response.

        Args:
            listen_response(`google.cloud.firestore_v1beta1.types.ListenResponse`):
                Callback method that receives a object to

def on_snapshot(self, proto):
        """
        Called everytime there is a response from listen. Collect changes
        and 'push' the changes in a batch to the customer when we receive
        'current' from the listen response.

        Args:
            listen_response(`google.cloud.firestore_v1beta1.types.ListenResponse`):
                Callback method that receives a object to
        """
        TargetChange = firestore_pb2.TargetChange

        target_changetype_dispatch = {
            TargetChange.NO_CHANGE: self._on_snapshot_target_change_no_change,
            TargetChange.ADD: self._on_snapshot_target_change_add,
            TargetChange.REMOVE: self._on_snapshot_target_change_remove,
            TargetChange.RESET: self._on_snapshot_target_change_reset,
            TargetChange.CURRENT: self._on_snapshot_target_change_current,
        }

        target_change = proto.target_change
        if str(target_change):
            target_change_type = target_change.target_change_type
            _LOGGER.debug("on_snapshot: target change: " + str(target_change_type))
            meth = target_changetype_dispatch.get(target_change_type)
            if meth is None:
                _LOGGER.info(
                    "on_snapshot: Unknown target change " + str(target_change_type)
                )
                self.close(
                    reason="Unknown target change type: %s " % str(target_change_type)
                )
            else:
                try:
                    meth(proto)
                except Exception as exc2:
                    _LOGGER.debug("meth(proto) exc: " + str(exc2))
                    raise

            # NOTE:
            # in other implementations, such as node, the backoff is reset here
            # in this version bidi rpc is just used and will control this.

        elif str(proto.document_change):
            _LOGGER.debug("on_snapshot: document change")

            # No other target_ids can show up here, but we still need to see
            # if the targetId was in the added list or removed list.
            target_ids = proto.document_change.target_ids or []
            removed_target_ids = proto.document_change.removed_target_ids or []
            changed = False
            removed = False

            if WATCH_TARGET_ID in target_ids:
                changed = True

            if WATCH_TARGET_ID in removed_target_ids:
                removed = True

            if changed:
                _LOGGER.debug("on_snapshot: document change: CHANGED")

                # google.cloud.firestore_v1beta1.types.DocumentChange
                document_change = proto.document_change
                # google.cloud.firestore_v1beta1.types.Document
                document = document_change.document

                data = _helpers.decode_dict(document.fields, self._firestore)

                # Create a snapshot. As Document and Query objects can be
                # passed we need to get a Document Reference in a more manual
                # fashion than self._document_reference
                document_name = document.name
                db_str = self._firestore._database_string
                db_str_documents = db_str + "/documents/"
                if document_name.startswith(db_str_documents):
                    document_name = document_name[len(db_str_documents) :]

                document_ref = self._firestore.document(document_name)

                snapshot = self.DocumentSnapshot(
                    reference=document_ref,
                    data=data,
                    exists=True,
                    read_time=None,
                    create_time=document.create_time,
                    update_time=document.update_time,
                )
                self.change_map[document.name] = snapshot

            elif removed:
                _LOGGER.debug("on_snapshot: document change: REMOVED")
                document = proto.document_change.document
                self.change_map[document.name] = ChangeType.REMOVED

        # NB: document_delete and document_remove (as far as we, the client,
        # are concerned) are functionally equivalent

        elif str(proto.document_delete):
            _LOGGER.debug("on_snapshot: document change: DELETE")
            name = proto.document_delete.document
            self.change_map[name] = ChangeType.REMOVED

        elif str(proto.document_remove):
            _LOGGER.debug("on_snapshot: document change: REMOVE")
            name = proto.document_remove.document
            self.change_map[name] = ChangeType.REMOVED

        elif proto.filter:
            _LOGGER.debug("on_snapshot: filter update")
            if proto.filter.count != self._current_size():
                # We need to remove all the current results.
                self._reset_docs()
                # The filter didn't match, so re-issue the query.
                # TODO: reset stream method?
                # self._reset_stream();

        else:
            _LOGGER.debug("UNKNOWN TYPE. UHOH")
            self.close(reason=ValueError("Unknown listen response type: %s" % proto))