Adds a ready event to BackgroundConsumer to allow waiting for the bac…

…kground thread to start

api_core/google/api_core/bidi.py

@@ -499,6 +499,7 @@ def __init__(self, bidi_rpc, on_response):
         self._wake = threading.Condition()
         self._thread = None
         self._operational_lock = threading.Lock()
+        self._ready = threading.Event()
 
     def _on_call_done(self, future):
         # Resume the thread if it's paused, this prevents blocking forever
@@ -507,6 +508,7 @@ def _on_call_done(self, future):
 
     def _thread_main(self):
         try:
+            self._ready.set()
             self._bidi_rpc.add_done_callback(self._on_call_done)
             self._bidi_rpc.open()
 
@@ -560,6 +562,10 @@ def start(self):
             )
             thread.daemon = True
             thread.start()
+            # In other places in the code, we rely on `thread.is_alive` which 
+            # isn't sufficient to know if a thread is going to run. To protect
+            # against races, we use a ready event and wait on it to be set.
+            self.is_ready.wait()
             self._thread = thread
             _LOGGER.debug("Started helper thread %s", thread.name)
 
@@ -598,3 +604,11 @@ def resume(self):
     def is_paused(self):
         """bool: True if the response stream is paused."""
         return self._paused
+
+    @property
+    def is_ready(self):
+        """
+        threading.Event: Returns an event indicating if this
+            consumer has started
+        """
+        return self._ready