Add bi-directional stream consumption helpers. #5189

theacodes · 2018-04-12T18:15:30Z

This is the foundation of the Pub/Sub subscriber refactor. It exposes a socket-like interface for gRPC streams and builds a robust, resumable stream on top of that. Finally, it adds a class for consuming the stream in the background using callbacks.

This code will replace the code in the current google.cloud.pubsub_v1.subscriber._consumer module.

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

@@ -0,0 +1,481 @@
+# Copyright 2017, Google LLC All rights reserved.


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

+        # Note: there is a possibility that this starts *before* the call
+        # property is set. So we have to check if self.call is set before
+        # seeing if it's active.
+        if self.call is not None and not self.call.is_active():


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

+            self._request_queue.put(request)
+        else:
+            # calling next should cause the call to raise.
+            next(self.call)


This is the foundation of the Pub/Sub subscriber refactor. It exposes a socket-like interface for gRPC streams and builds a robust, resumable stream on top of that. Finally, it adds a class for consuming the stream in the background using callbacks.

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

+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0


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

+            if not self._is_active():
+                # We have an item, but the call is closed. We should put the
+                # item back on the queue so that the next call can consume it.
+                self._queue.put(item)


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

+                It will be provided with the same gRPC future as the underlying
+                stream which will also be a :class:`grpc.Call`.
+        """
+        self._callbacks.append(callback)


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

+    def open(self):
+        """Opens the stream."""
+        if self.is_active:
+            raise ValueError('Can not open an already open stream.')


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

+    def add_done_callback(self, callback):
+        """Adds a callback that will be called when the RPC terminates.
+
+        This occurs when the RPC errors or is successfully terminated.


lukesneeringer

Approved with comments. Addressing the comments is not a prerequisite to merge.

theacodes requested review from lukesneeringer and crwilcox April 12, 2018 18:15

googlebot added the cla: yes This human has signed the Contributor License Agreement. label Apr 12, 2018

crwilcox suggested changes Apr 16, 2018

View reviewed changes

crwilcox approved these changes Apr 16, 2018

View reviewed changes

Jon Wayne Parrott added 5 commits April 16, 2018 10:52

Fix tiny non-determinism

08a8d7b

Add missing __init__

406d495

Jon is super dumb and coded in a super dumb non-deterministic deadlock

58fcb53

Address review comments

d86adb3

theacodes force-pushed the pubsub-bidi branch from cf2bec3 to d86adb3 Compare April 16, 2018 17:52

Jon Wayne Parrott added 2 commits April 16, 2018 13:08

Fix another dumb key order thing

4baf2ac

Fix some more tiny non-determinism

e503d98

lukesneeringer reviewed Apr 17, 2018

View reviewed changes

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

def open(self):

"""Opens the stream."""

if self.is_active:

raise ValueError('Can not open an already open stream.')

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

lukesneeringer reviewed Apr 17, 2018

View reviewed changes

lukesneeringer approved these changes Apr 17, 2018

View reviewed changes

Fix copywrite

9eaa23b

theacodes merged commit dc3a8f9 into master Apr 17, 2018

theacodes deleted the pubsub-bidi branch April 17, 2018 19:05

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add bi-directional stream consumption helpers. #5189

Add bi-directional stream consumption helpers. #5189

theacodes commented Apr 12, 2018

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

lukesneeringer left a comment

		@@ -0,0 +1,481 @@
		# Copyright 2017, Google LLC All rights reserved.

Add bi-directional stream consumption helpers. #5189

Add bi-directional stream consumption helpers. #5189

Conversation

theacodes commented Apr 12, 2018

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

lukesneeringer left a comment

Choose a reason for hiding this comment