Make sure Pub / Sub BIDI consumer doesn't try to join itself. (#4540)

This functionality was removed in #4537, so this puts it back.

Author: dhermes

google/cloud/pubsub_v1/subscriber/_consumer.py

@@ -321,7 +321,8 @@ def _blocking_consume(self):
                 if recover:
                     recover = self._stop_request_generator(request_generator)
                 if not recover:
-                    self.stop_consuming()
+                    self._stop_no_join()
+                    return
 
     def start_consuming(self):
         """Start consuming the stream."""
@@ -336,11 +337,35 @@ def start_consuming(self):
         _LOGGER.debug('Started helper thread %s', thread.name)
         self._consumer_thread = thread
 
-    def stop_consuming(self):
-        """Signal the stream to stop and block until it completes."""
+    def _stop_no_join(self):
+        """Signal the request stream to stop.
+
+        To actually stop the worker ("consumer thread"), a ``STOP`` is
+        sent to the request queue.
+
+        The ``_consumer_thread`` member is removed from the current instance
+        and returned.
+
+        Returns:
+            threading.Thread: The worker ("consumer thread") that is being
+            stopped.
+        """
         self.active = False
         self._exiting.set()
         _LOGGER.debug('Stopping helper thread %s', self._consumer_thread.name)
         self.send_request(_helper_threads.STOP)
-        self._consumer_thread.join()
+        thread = self._consumer_thread
         self._consumer_thread = None
+        return thread
+
+    def stop_consuming(self):
+        """Signal the stream to stop and block until it completes.
+
+        To actually stop the worker ("consumer thread"), a ``STOP`` is
+        sent to the request queue.
+
+        This **assumes** that the caller is not in the same thread
+        (since a thread cannot ``join()`` itself).
+        """
+        thread = self._stop_no_join()
+        thread.join()

tests/unit/pubsub_v1/subscriber/test_consumer.py

@@ -77,6 +77,18 @@ def test_blocking_consume():
             assert on_res.mock_calls[1][1][1] == mock.sentinel.B
 
 
+@mock.patch.object(_consumer, '_LOGGER')
+def test_blocking_consume_when_exiting(_LOGGER):
+    consumer = create_consumer()
+    assert consumer._exiting.is_set() is False
+    consumer._exiting.set()
+
+    # Make sure method cleanly exits.
+    assert consumer._blocking_consume() is None
+
+    _LOGGER.debug.assert_called_once_with('Event signalled consumer exit.')
+
+
 class OnException(object):
 
     def __init__(self, acceptable=None):
@@ -149,6 +161,23 @@ def test_start_consuming():
     assert consumer._consumer_thread is Thread.return_value
 
 
+def test_stop_consuming():
+    consumer = create_consumer()
+    consumer.active = True
+    assert consumer._exiting.is_set() is False
+    thread = mock.Mock(spec=threading.Thread)
+    consumer._consumer_thread = thread
+
+    assert consumer.stop_consuming() is None
+
+    # Make sure state was updated.
+    assert consumer.active is False
+    assert consumer._exiting.is_set() is True
+    assert consumer._consumer_thread is None
+    # Check mocks.
+    thread.join.assert_called_once_with()
+
+
 def basic_queue_generator(queue, received):
     while True:
         value = queue.get()