Removing redundant Consumer.active member. (#4549)

Also renaming `Consumer._exiting` as `Consumer.stopped` (and
making it a public member).

Author: dhermes

google/cloud/pubsub_v1/subscriber/_consumer.py

@@ -184,10 +184,8 @@ def __init__(self, policy):
         """
         self._policy = policy
         self._request_queue = queue.Queue()
-        self._exiting = threading.Event()
+        self.stopped = threading.Event()
         self._put_lock = threading.Lock()
-
-        self.active = False
         self._consumer_thread = None
 
     def send_request(self, request):
@@ -299,7 +297,7 @@ def _blocking_consume(self):
             # exit cleanly when the user has called stop_consuming(). This
             # checks to make sure we're not exiting before opening a new
             # stream.
-            if self._exiting.is_set():
+            if self.stopped.is_set():
                 _LOGGER.debug('Event signalled consumer exit.')
                 break
 
@@ -326,8 +324,7 @@ def _blocking_consume(self):
 
     def start_consuming(self):
         """Start consuming the stream."""
-        self.active = True
-        self._exiting.clear()
+        self.stopped.clear()
         thread = threading.Thread(
             name=_BIDIRECTIONAL_CONSUMER_NAME,
             target=self._blocking_consume,
@@ -350,8 +347,7 @@ def _stop_no_join(self):
             threading.Thread: The worker ("consumer thread") that is being
             stopped.
         """
-        self.active = False
-        self._exiting.set()
+        self.stopped.set()
         _LOGGER.debug('Stopping helper thread %s', self._consumer_thread.name)
         self.send_request(_helper_threads.STOP)
         thread = self._consumer_thread

google/cloud/pubsub_v1/subscriber/policy/base.py

@@ -177,7 +177,7 @@ def ack(self, ack_id, time_to_ack=None, byte_size=None):
         # However, if the consumer is inactive, then queue the ack_id here
         # instead; it will be acked as part of the initial request when the
         # consumer is started again.
-        if self._consumer.active:
+        if self._consumer.stopped.is_set():
             request = types.StreamingPullRequest(ack_ids=[ack_id])
             self._consumer.send_request(request)
         else:
@@ -311,7 +311,7 @@ def maintain_leases(self):
         """
         while True:
             # Sanity check: Should this infinite loop quit?
-            if not self._consumer.active:
+            if not self._consumer.stopped.is_set():
                 _LOGGER.debug('Consumer inactive, ending lease maintenance.')
                 return
 
@@ -326,7 +326,7 @@ def maintain_leases(self):
             # because it is more efficient to make a single request.
             ack_ids = list(self.managed_ack_ids)
             _LOGGER.debug('Renewing lease for %d ack IDs.', len(ack_ids))
-            if ack_ids and self._consumer.active:
+            if ack_ids and self._consumer.stopped.is_set():
                 request = types.StreamingPullRequest(
                     modify_deadline_ack_ids=ack_ids,
                     modify_deadline_seconds=[p99] * len(ack_ids),

tests/unit/pubsub_v1/subscriber/test_consumer.py

@@ -80,8 +80,8 @@ def test_blocking_consume():
 @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()
+    assert consumer.stopped.is_set() is False
+    consumer.stopped.set()
 
     # Make sure method cleanly exits.
     assert consumer._blocking_consume() is None
@@ -152,8 +152,7 @@ def test_start_consuming():
     with mock.patch.object(threading, 'Thread', autospec=True) as Thread:
         consumer.start_consuming()
 
-    assert consumer._exiting.is_set() is False
-    assert consumer.active is True
+    assert consumer.stopped.is_set() is False
     Thread.assert_called_once_with(
         name=_consumer._BIDIRECTIONAL_CONSUMER_NAME,
         target=consumer._blocking_consume,
@@ -163,16 +162,14 @@ def test_start_consuming():
 
 def test_stop_consuming():
     consumer = create_consumer()
-    consumer.active = True
-    assert consumer._exiting.is_set() is False
+    assert consumer.stopped.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.stopped.is_set() is True
     assert consumer._consumer_thread is None
     # Check mocks.
     thread.join.assert_called_once_with()

tests/unit/pubsub_v1/subscriber/test_policy_base.py

@@ -85,7 +85,7 @@ def test_subscription():
 
 def test_ack():
     policy = create_policy()
-    policy._consumer.active = True
+    policy._consumer.stopped.set()
     with mock.patch.object(policy._consumer, 'send_request') as send_request:
         policy.ack('ack_id_string', 20)
         send_request.assert_called_once_with(types.StreamingPullRequest(
@@ -97,7 +97,7 @@ def test_ack():
 
 def test_ack_no_time():
     policy = create_policy()
-    policy._consumer.active = True
+    policy._consumer.stopped.set()
     with mock.patch.object(policy._consumer, 'send_request') as send_request:
         policy.ack('ack_id_string')
         send_request.assert_called_once_with(types.StreamingPullRequest(
@@ -109,7 +109,7 @@ def test_ack_no_time():
 def test_ack_paused():
     policy = create_policy()
     policy._paused = True
-    policy._consumer.active = False
+    policy._consumer.stopped.clear()
     with mock.patch.object(policy, 'open') as open_:
         policy.ack('ack_id_string')
         open_.assert_called()
@@ -198,20 +198,20 @@ def test_modify_ack_deadline():
 
 def test_maintain_leases_inactive_consumer():
     policy = create_policy()
-    policy._consumer.active = False
+    policy._consumer.stopped.clear()
     assert policy.maintain_leases() is None
 
 
 def test_maintain_leases_ack_ids():
     policy = create_policy()
-    policy._consumer.active = True
+    policy._consumer.stopped.set()
     policy.lease('my ack id', 50)
 
     # Mock the sleep object.
     with mock.patch.object(time, 'sleep', autospec=True) as sleep:
         def trigger_inactive(seconds):
             assert 0 < seconds < 10
-            policy._consumer.active = False
+            policy._consumer.stopped.clear()
         sleep.side_effect = trigger_inactive
 
         # Also mock the consumer, which sends the request.
@@ -226,11 +226,11 @@ def trigger_inactive(seconds):
 
 def test_maintain_leases_no_ack_ids():
     policy = create_policy()
-    policy._consumer.active = True
+    policy._consumer.stopped.set()
     with mock.patch.object(time, 'sleep', autospec=True) as sleep:
         def trigger_inactive(seconds):
             assert 0 < seconds < 10
-            policy._consumer.active = False
+            policy._consumer.stopped.clear()
         sleep.side_effect = trigger_inactive
         policy.maintain_leases()
         sleep.assert_called()