Re-factoring Policy constructor. (#4535)

Doing so to make it clearer what is happening during construction.

Also

- removing the `_callback_requests` member (not needed)
- modifying some docstrings in the `pubsub_v1.subscriber._constructor`
  module so that it is valid Sphinx/rST

Author: dhermes

google/cloud/pubsub_v1/subscriber/_consumer.py

@@ -33,6 +33,7 @@
 :class:`concurrent.futures.Executor`:
 
 .. graphviz::
+
     digraph responses_only {
        "gRPC C Core" -> "gRPC Python" [label="queue", dir="both"]
        "gRPC Python" -> "Consumer" [label="responses", color="red"]
@@ -57,6 +58,7 @@
 a queue for that:
 
 .. graphviz::
+
     digraph response_flow {
         "gRPC C Core" -> "gRPC Python" [label="queue", dir="both"]
         "gRPC Python" -> "Consumer" [label="responses", color="red"]
@@ -71,6 +73,7 @@
 queue new requests:
 
 .. graphviz::
+
     digraph thread_only_requests {
         "gRPC C Core" -> "gRPC Python" [label="queue", dir="both"]
         "gRPC Python" -> "Consumer" [label="responses", color="red"]
@@ -92,6 +95,7 @@
 all together looks like this:
 
 .. graphviz::
+
     digraph responses_only {
         "gRPC C Core" -> "gRPC Python" [label="queue", dir="both"]
         "gRPC Python" -> "Consumer" [label="responses", color="red"]
@@ -110,7 +114,8 @@
     }
 
 This part is actually up to the Policy to enable. The consumer just provides a
-thread-safe queue for requests. The :cls:`QueueCallbackWorker` can be used by
+thread-safe queue for requests. The :class:`QueueCallbackWorker` can be used by
+
 the Policy implementation to spin up the worker thread to pump the
 concurrency-safe queue. See the Pub/Sub subscriber implementation for an
 example of this.
@@ -146,7 +151,7 @@ class Consumer(object):
     generate requests. This thread is called the *request generator thread*.
     Having the request generator thread allows the consumer to hold the stream
     open indefinitely. Now gRPC will send responses as fast as the consumer can
-    ask for them. The consumer hands these off to the :cls:`Policy` via
+    ask for them. The consumer hands these off to the :class:`Policy` via
     :meth:`Policy.on_response`, which should not block.
 
     Finally, we do not want to block the main thread, so the consumer actually
@@ -184,7 +189,7 @@ def __init__(self, policy):
 
         self.active = False
         self.helper_threads = _helper_threads.HelperThreadRegistry()
-        """:cls:`_helper_threads.HelperThreads`: manages the helper threads.
+        """:class:`_helper_threads.HelperThreads`: manages the helper threads.
             The policy may use this to schedule its own helper threads.
         """
 

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

@@ -67,60 +67,81 @@ class Policy(base.BasePolicy):
 
     This consumer handles the connection to the Pub/Sub service and all of
     the concurrency needs.
+
+    Args:
+        client (~.pubsub_v1.subscriber.client): The subscriber client used
+            to create this instance.
+        subscription (str): The name of the subscription. The canonical
+            format for this is
+            ``projects/{project}/subscriptions/{subscription}``.
+        flow_control (~google.cloud.pubsub_v1.types.FlowControl): The flow
+            control settings.
+        executor (~concurrent.futures.ThreadPoolExecutor): (Optional.) A
+            ThreadPoolExecutor instance, or anything duck-type compatible
+            with it.
+        queue (~queue.Queue): (Optional.) A Queue instance, appropriate
+            for crossing the concurrency boundary implemented by
+            ``executor``.
     """
+
     def __init__(self, client, subscription, flow_control=types.FlowControl(),
                  executor=None, queue=None):
-        """Instantiate the policy.
+        super(Policy, self).__init__(
+            client=client,
+            flow_control=flow_control,
+            subscription=subscription,
+        )
+        # Default the callback to a no-op; the **actual** callback is
+        # provided by ``.open()``.
+        self._callback = _do_nothing_callback
+        # Create a queue for keeping track of shared state.
+        self._request_queue = self._get_queue(queue)
+        # Also maintain an executor.
+        self._executor = self._get_executor(executor)
+
+    @staticmethod
+    def _get_queue(queue):
+        """Gets a queue for the constructor.
 
         Args:
-            client (~.pubsub_v1.subscriber.client): The subscriber client used
-                to create this instance.
-            subscription (str): The name of the subscription. The canonical
-                format for this is
-                ``projects/{project}/subscriptions/{subscription}``.
-            flow_control (~google.cloud.pubsub_v1.types.FlowControl): The flow
-                control settings.
-            executor (~concurrent.futures.ThreadPoolExecutor): (Optional.) A
-                ThreadPoolExecutor instance, or anything duck-type compatible
-                with it.
-            queue (~queue.Queue): (Optional.) A Queue instance, appropriate
+            queue (Optional[~queue.Queue]): A Queue instance, appropriate
                 for crossing the concurrency boundary implemented by
                 ``executor``.
-        """
-        # Default the callback to a no-op; it is provided by `.open`.
-        self._callback = _do_nothing_callback
 
-        # Default the future to None; it is provided by `.open`.
-        self._future = None
-
-        # Create a queue for keeping track of shared state.
+        Returns:
+            ~queue.Queue: Either ``queue`` if not :data:`None` or a default
+            queue.
+        """
         if queue is None:
-            queue = queue_mod.Queue()
-        self._request_queue = queue
+            return queue_mod.Queue()
+        else:
+            return queue
 
-        # Call the superclass constructor.
-        super(Policy, self).__init__(
-            client=client,
-            flow_control=flow_control,
-            subscription=subscription,
-        )
+    @staticmethod
+    def _get_executor(executor):
+        """Gets an executor for the constructor.
+
+        Args:
+            executor (Optional[~concurrent.futures.ThreadPoolExecutor]): A
+                ThreadPoolExecutor instance, or anything duck-type compatible
+                with it.
 
-        # Also maintain a request queue and an executor.
+        Returns:
+            ~concurrent.futures.ThreadPoolExecutor: Either ``executor`` if not
+            :data:`None` or a default thread pool executor with 10 workers
+            and a prefix (if supported).
+        """
         if executor is None:
             executor_kwargs = {}
             if sys.version_info[:2] == (2, 7) or sys.version_info >= (3, 6):
                 executor_kwargs['thread_name_prefix'] = (
                     'ThreadPoolExecutor-SubscriberPolicy')
-            executor = futures.ThreadPoolExecutor(
+            return futures.ThreadPoolExecutor(
                 max_workers=10,
                 **executor_kwargs
             )
-        self._executor = executor
-        _LOGGER.debug('Creating callback requests thread (not starting).')
-        self._callback_requests = _helper_threads.QueueCallbackWorker(
-            self._request_queue,
-            self.dispatch_callback,
-        )
+        else:
+            return executor
 
     def close(self):
         """Close the existing connection."""
@@ -158,10 +179,14 @@ def open(self, callback):
         # Start the thread to pass the requests.
         _LOGGER.debug('Starting callback requests worker.')
         self._callback = callback
+        dispatch_worker = _helper_threads.QueueCallbackWorker(
+            self._request_queue,
+            self.dispatch_callback,
+        )
         self._consumer.helper_threads.start(
             _CALLBACK_WORKER_NAME,
             self._request_queue.put,
-            self._callback_requests,
+            dispatch_worker,
         )
 
         # Actually start consuming messages.