Allow passing metadata as part of creating a bidi (#7514)

* allows providing rpc metadata for bidi streams

Author: crwilcox

packages/google-cloud-firestore/google/cloud/firestore_v1beta1/watch.py

@@ -213,9 +213,10 @@ def should_recover(exc):  # pragma: NO COVER
             ResumableBidiRpc = self.ResumableBidiRpc  # FBO unit tests
 
         self._rpc = ResumableBidiRpc(
-            self._api.transport._stubs["firestore_stub"].Listen,
+            self._api.transport.listen,
             initial_request=initial_request,
             should_recover=should_recover,
+            rpc_metadata=self._firestore._rpc_metadata,
         )
 
         self._rpc.add_done_callback(self._on_rpc_done)

packages/google-cloud-firestore/tests/unit/v1beta1/test_cross_language.py

@@ -342,12 +342,13 @@ def convert_precondition(precond):
 
 
 class DummyRpc(object):  # pragma: NO COVER
-    def __init__(self, listen, initial_request, should_recover):
+    def __init__(self, listen, initial_request, should_recover, rpc_metadata=None):
         self.listen = listen
         self.initial_request = initial_request
         self.should_recover = should_recover
         self.closed = False
         self.callbacks = []
+        self._rpc_metadata = rpc_metadata
 
     def add_done_callback(self, callback):
         self.callbacks.append(callback)

packages/google-cloud-firestore/tests/unit/v1beta1/test_watch.py

@@ -713,6 +713,7 @@ def _to_protobuf(self):
 class DummyFirestore(object):
     _firestore_api = DummyFirestoreClient()
     _database_string = "abc://bar/"
+    _rpc_metadata = None
 
     def document(self, *document_path):  # pragma: NO COVER
         if len(document_path) == 1:
@@ -781,12 +782,13 @@ def Thread(self, name, target, kwargs):
 
 
 class DummyRpc(object):
-    def __init__(self, listen, initial_request, should_recover):
+    def __init__(self, listen, initial_request, should_recover, rpc_metadata=None):
         self.listen = listen
         self.initial_request = initial_request
         self.should_recover = should_recover
         self.closed = False
         self.callbacks = []
+        self._rpc_metadata = rpc_metadata
 
     def add_done_callback(self, callback):
         self.callbacks.append(callback)