Fix race condition in recv()'s usage of self.call.

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/bidi.py

@@ -207,21 +207,8 @@ def open(self):
  else:
  call.add_done_callback(self._on_call_done)
 
- import time
- import random
- from unittest import mock
- from google.cloud.pubsub_v1.proto import pubsub_pb2
-
- mock_call = mock.MagicMock(wraps=call)
-
- def mock_next():
- #time.sleep(random.uniform(0.0, 1.0))
- return pubsub_pb2.StreamingPullResponse()
-
- mock_call.__next__.side_effect = mock_next
-
  self._request_generator = request_generator
- self.call = mock_call
+ self.call = call
 
  def close(self):
  """Closes the stream."""
@@ -252,7 +239,6 @@ def send(self, request):
  # to mean something semantically different.
  if self.call.is_active():
  self._request_queue.put(request)
- pass
  else:
  # calling next should cause the call to raise.
  next(self.call)
@@ -356,7 +342,7 @@ def _reopen(self):
  # Another thread already managed to re-open this stream.
  if self.call is not None and self.call.is_active():
  _LOGGER.debug('Stream was already re-established.')
- #return
+ return
 
  self.call = None
  # Request generator should exit cleanly since the RPC its bound to
@@ -454,8 +440,7 @@ def _recv(self):
  return next(call)
 
  def recv(self):
- return self._recoverable(
- super(ResumableBidiRpc, self).recv)
+ return self._recoverable(self._recv)
 
  @property
  def is_active(self):
@@ -471,16 +456,6 @@ def is_active(self):
  return self.call is not None and not self._finalized
 
 
-def meanie_thread_main(rpc):
- import time
- import random
-
- while True:
- time.sleep(random.uniform(0, 0.1))
- print('Wahaha')
- rpc._reopen()
-
-
 class BackgroundConsumer(object):
  """A bi-directional stream consumer that runs in a separate thread.
 
@@ -536,13 +511,6 @@ def _thread_main(self):
  self._bidi_rpc.add_done_callback(self._on_call_done)
  self._bidi_rpc.open()
 
- import functools
- meanie_thread = threading.Thread(
- name='meanie thread',
- target=functools.partial(meanie_thread_main, self._bidi_rpc))
- meanie_thread.deamon = True
- meanie_thread.start()
-
  while self._bidi_rpc.is_active:
  # Do not allow the paused status to change at all during this
  # section. There is a condition where we could be resumed

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

@@ -263,8 +263,7 @@ def send(self, request):
  """Queue a request to be sent to the RPC."""
  if self._UNARY_REQUESTS:
  try:
- pass
- #self._send_unary_request(request)
+ self._send_unary_request(request)
  except exceptions.GoogleAPICallError as exc:
  _LOGGER.debug(
  'Exception while sending unary RPC. This is typically '