PubSub Publisher #3463

lukesneeringer · 2017-06-01T14:40:00Z

This is what I assert to be a mostly complete publisher for Pub/Sub (minus error checking, which is a pretty big caveat, and tests, which is also a pretty big caveat). I would like feedback on this much before I move forward, since (as we all know) multiprocessing is tricky.

I am about to add some notes inline at specific places, but obviously feel free to comment on any of it.

…python into pubsub

pubsub/google/cloud/pubsub_v1/publisher/batch.py

+    def __init__(self, client, topic, settings, autocommit=True):
+        self._client = client
+
+        # Create a namespace that is owned by the client manager; this


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+            raise Exception('Empty queue')
+        response = self._client.api.publish(self._.topic, self._.messages)
+
+        # FIXME (lukesneeringer): Check for failures; retry.


pubsub/google/cloud/pubsub_v1/publisher/exceptions.py

@@ -1,4 +1,4 @@
-# Copyright 2016 Google Inc.


pubsub/google/cloud/pubsub_v1/publisher/future.py

+        Args:
+            message_id (str): The message ID, as a string.
+        """
+        for callback in self._callbacks:


pubsub/google/cloud/pubsub_v1/retry.py

+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+def retry(func, delay=0, count=0, err=None, **kwargs):


pubsub/google/cloud/pubsub_v1/subscriber.py

+__VERSION__ = pkg_resources.get_distribution('google-cloud-pubsub').version
+
+
+class SubscriberClient(subscriber_client.SubscriberClient):


pubsub/google/cloud/pubsub_v1/__init__.py

+
+__all__ = (
+    'PublisherClient',
+    'SubscriberClient',


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+    :class:`google.cloud.pubsub_v1.PublisherClient`.
+
+    Args:
+        client (:class:`google.cloud.pubsub_v1.PublisherClient`): The


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+            inferred from this.
+        topic (str): The topic. The format for this is
+            ``projects/{project}/topics/{topic}``.
+        settings (:class:`google.cloud.pubsub_v1.types.Batching`): The


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+        # Create a namespace that is owned by the client manager; this
+        # is necessary to be able to have these values be communicable between
+        # processes.
+        self._ = self.manager.Namespace()


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+        self._.messages = self.manager.list()
+        self._.message_ids = self.manager.dict()
+        self._.settings = settings
+        self._.status = 'accepting messages'


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+            Future: An object conforming to the ``concurrent.futures.Future``
+                interface.
+        """
+        # Sanity check: Is the data being sent as a bytestring?


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+        return f
+
+
+# Make a fake batch. This is used by the client to do single-op checks


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+    def commit(self):
+        """Actually publish all of the messages on the active batch.
+
+        This moves the batch out from being the active batch to an in-flight


pubsub/google/cloud/pubsub_v1/publisher/future.py

+        Args:
+            message_id (str): The message ID, as a string.
+        """
+        for callback in self._callbacks:


pubsub/google/cloud/pubsub_v1/_gapic.py

+import functools
+
+
+def add_methods(SourceClass, blacklist=()):


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+    :class:`google.cloud.pubsub_v1.PublisherClient`.
+
+    Args:
+        client (:class:`google.cloud.pubsub_v1.PublisherClient`): The


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+    message to be published is received; subsequent messages are added to
+    that batch until the process of actual publishing _starts_.
+
+    Once this occurs, any new messages sent to ``publish`` open a new batch.


pubsub/google/cloud/pubsub_v1/publisher/batch.py

+            Future: An object conforming to the ``concurrent.futures.Future``
+                interface.
+        """
+        # Sanity check: Is the data being sent as a bytestring?


pubsub/google/cloud/pubsub_v1/publisher/client.py

+            :class:`~gapic.pubsub.v1.publisher_client.PublisherClient`.
+            Generally, you should not need to set additional keyword arguments.
+    """
+    _gapic_class = publisher_client.PublisherClient


pubsub/google/cloud/pubsub_v1/publisher/client.py

+        """
+        # If there is no matching batch yet, then potentially create one
+        # and place it on the batches dictionary.
+        if self._batches.get(topic, FAKE).status != 'accepting messages':


pubsub/google/cloud/pubsub_v1/publisher/future.py

+
+import queue
+import uuid
+import time


Also rename base.BaseBatch to base.Batch.

theacodes

Looks mostly good, some small things and nits.

pubsub/google/cloud/pubsub_v1/publisher/batch/base.py

+        """
+        raise NotImplementedError
+
+    class Status(object):


pubsub/google/cloud/pubsub_v1/publisher/batch/base.py

@@ -0,0 +1,130 @@
+# Copyright 2017, Google Inc. All rights reserved.


pubsub/google/cloud/pubsub_v1/publisher/batch/base.py

+        """
+        raise NotImplementedError
+
+    @property


pubsub/google/cloud/pubsub_v1/publisher/batch/thread.py

+        # commit when the max latency is reached.
+        self._thread = None
+        if autocommit and self._settings.max_latency < float('inf'):
+            self._thread = threading.Thread(target=self.monitor)


pubsub/google/cloud/pubsub_v1/publisher/batch/thread.py

+            version, which calls this one.
+        """
+        # Update the status.
+        self._status = 'in-flight'


pubsub/google/cloud/pubsub_v1/publisher/client.py

+            if isinstance(v, six.binary_type):
+                attrs[k] = v.decode('utf-8')
+                continue
+            raise TypeError('All attributes being published to Pub/Sub must '


pubsub/google/cloud/pubsub_v1/publisher/exceptions.py

+    pass
+
+
+class TimeoutError(GoogleAPIError):


pubsub/google/cloud/pubsub_v1/publisher/futures.py

+        return False
+
+    def running(self):
+        """Publishes in Pub/Sub currently may not be canceled.


pubsub/google/cloud/pubsub_v1/publisher/futures.py

+
+        # Wait a little while and try again.
+        time.sleep(_wait)
+        return self.exception(


pubsub/google/cloud/pubsub_v1/types.py

+)
+
+
+names = ['BatchSettings', 'FlowControl']


#3463 (comment)

lukesneeringer · 2017-08-22T20:16:28Z

@jonparrott I think this is good other than the inner Status class that I asked about.

theacodes · 2017-08-22T20:19:03Z

@jonparrott I think this is good other than the inner Status class that I asked about.

I answered

lukesneeringer · 2017-08-22T23:11:21Z

Got it. GitHub obscured that and I missed it.

lukesneeringer · 2017-08-22T23:11:36Z

Also, I think this knocks out everything on this PR. Is that accurate?

theacodes · 2017-08-22T23:21:51Z

Two outstanding items:

lukesneeringer · 2017-08-24T03:48:55Z

Intentionally bypassing CircleCI because:

They already ran on pubsub-subscriber, which was branched off of this branch.
Same for pubsub-docs.
I am setting up a final check to merge pubsub to master, and CI can and will run from that tonight to hopefully merge tomorrow morning.

plamut · 2021-12-22T14:37:15Z

pubsub/google/cloud/pubsub_v1/_gapic.py

+def add_methods(source_class, blacklist=()):
+    """Add wrapped versions of the `api` member's methods to the class.


@lukesneeringer @theacodes @dhermes

Hey folks, happy 2022!

This is a long shot, as the PR is more than 4 years old, but does any of you remember what problem this decorator tried to solve? Why the generated client classes were not directly subclassed by the hand-written classes and instead kept an instance of the generated class in the .api attribute?

The thing is that type checkers issue a lot of complaints about non-existing methods, because they simply cannot know that a lot of these methods are dynamically injected at import time.

We would thus like to subclass the generated classes directly, already have a POC working, but we would like to know if there are any significant downsides to it, if anyone remembers?

Thanks!

…eCloudPlatform/python-docs-samples#3463)

Luke Sneeringer and others added 14 commits May 25, 2017 07:03

[WIP] Getting started on Pub/Sub.

f784121

stuff

c8a9fd6

WIP

9b09b8f

wip

8f0832e

WIP

02d7658

Merge branch 'pubsub' of github.com:GoogleCloudPlatform/google-cloud-…

3a3ebe9

…python into pubsub

WIP, fixing bugs.

503d11f

WIP

3b21b93

wip

1e879a1

wip

4bf0552

wip

c6dc098

wip

13821a7

wip

c2d7af8

Merge branch 'public-master' into pubsub

4c0bf84

lukesneeringer added api: pubsub Issues related to the Pub/Sub API. priority: p1 Important issue which blocks shipping the next release. Will be fixed prior to next release. do not merge Indicates a pull request not ready for merge, due to either quality or timing. labels Jun 1, 2017

lukesneeringer self-assigned this Jun 1, 2017

lukesneeringer requested review from theacodes and dhermes June 1, 2017 14:40

googlebot added the cla: yes This human has signed the Contributor License Agreement. label Jun 1, 2017

lukesneeringer commented Jun 1, 2017

View reviewed changes

pubsub/google/cloud/pubsub_v1/publisher/batch.py Outdated

def __init__(self, client, topic, settings, autocommit=True):

self._client = client

# Create a namespace that is owned by the client manager; this

This comment was marked as spam.

Sign in to view

lukesneeringer commented Jun 1, 2017

View reviewed changes

pubsub/google/cloud/pubsub_v1/publisher/batch.py Outdated

raise Exception('Empty queue')

response = self._client.api.publish(self._.topic, self._.messages)

# FIXME (lukesneeringer): Check for failures; retry.

This comment was marked as spam.

Sign in to view

lukesneeringer commented Jun 1, 2017

View reviewed changes

pubsub/google/cloud/pubsub_v1/publisher/exceptions.py Outdated

@@ -1,4 +1,4 @@

# Copyright 2016 Google Inc.

This comment was marked as spam.

Sign in to view

lukesneeringer commented Jun 1, 2017

View reviewed changes

pubsub/google/cloud/pubsub_v1/retry.py Outdated

# See the License for the specific language governing permissions and

# limitations under the License.

def retry(func, delay=0, count=0, err=None, **kwargs):

This comment was marked as spam.

Sign in to view

lukesneeringer commented Jun 1, 2017

View reviewed changes

pubsub/google/cloud/pubsub_v1/subscriber.py Outdated

__VERSION__ = pkg_resources.get_distribution('google-cloud-pubsub').version

class SubscriberClient(subscriber_client.SubscriberClient):

This comment was marked as spam.

Sign in to view

theacodes suggested changes Jun 1, 2017

View reviewed changes

dhermes reviewed Jun 1, 2017

View reviewed changes

Merge branch 'pubsub' into pubsub-publisher

783e9dd

Luke Sneeringer added 11 commits August 21, 2017 11:25

Improve the max latency thread comments.

356749a

Collapse property docstrings.

8242c9d

More @jonparrott feedback.

db87dab

Remove the client as a public item in the base batch.

58072b8

Also rename base.BaseBatch to base.Batch.

Remove the rejection batch.

8f67488

Lock batch acquisition.

a86d9b7

Alter exception superclass.

df18615

Inherit from google.api.core.future.Future.

5f0549b

Move to @jonparrott's Future interface.

101d9ca

Move Future off into its own module.

e6e58bb

Add is not None.

9fd490c

theacodes reviewed Aug 22, 2017

View reviewed changes

Luke Sneeringer added 5 commits August 22, 2017 10:31

Feedback.

469400e

#3463 (comment)

Use concurrent.futures.TimeoutError where possible.

31c0c81

Add a lock on commit to ensure no duplication.

04ac278

Move to an Event.

37ee908

Make _names private.

12d44be

theacodes approved these changes Aug 22, 2017

View reviewed changes

Pubsub subscriber (#3637)

6fe1309

lukesneeringer merged commit 6bf3359 into pubsub Aug 24, 2017

lukesneeringer deleted the pubsub-publisher branch August 24, 2017 03:49

lukesneeringer mentioned this pull request Aug 24, 2017

Pub/Sub API Redesign #3859

Merged

plamut reviewed Dec 22, 2021

View reviewed changes

parthea pushed a commit that referenced this pull request Oct 21, 2023

chore(deps): update dependency cryptography to v2.9.1 [(#3463)](Googl…

417789f

…eCloudPlatform/python-docs-samples#3463)

		__VERSION__ = pkg_resources.get_distribution('google-cloud-pubsub').version


		class SubscriberClient(subscriber_client.SubscriberClient):

		return f


		# Make a fake batch. This is used by the client to do single-op checks

		@@ -0,0 +1,130 @@
		# Copyright 2017, Google Inc. All rights reserved.

		def add_methods(source_class, blacklist=()):
		"""Add wrapped versions of the `api` member's methods to the class.

PubSub Publisher #3463

PubSub Publisher #3463

Conversation

lukesneeringer commented Jun 1, 2017

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.

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.

This comment was marked as spam.

This comment was marked as spam.

theacodes left a comment

Choose a reason for hiding this comment

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.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

lukesneeringer commented Aug 22, 2017

theacodes commented Aug 22, 2017

lukesneeringer commented Aug 22, 2017

lukesneeringer commented Aug 22, 2017

theacodes commented Aug 22, 2017

lukesneeringer commented Aug 24, 2017

plamut Dec 22, 2021

Choose a reason for hiding this comment