Pubsub subscriber #3637

lukesneeringer · 2017-07-20T13:55:31Z

Uses #3463 as a base.

This PR implements the Pub/Sub subscriber, which by default uses threading.

The following is yet to be implemented:

Automatic pause/resume based on flow control settings.
Tests of any kind (working on them in parallel with the Pub/Sub team's review)
API Documentation

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

+        Args:
+            ack_id (str): The ack ID.
+        """
+        request = types.StreamingPullRequest(ack_ids=[ack_id])


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

+                time as there are).
+        """
+        return self._client.api.streaming_pull(request_generator,
+            options=gax.CallOptions(timeout=600),


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

+        time.sleep(snooze)
+        self.maintain_leases()
+
+    def modify_ack_deadline(self, ack_id, seconds):


pubsub/.flake8

  *.pyc,
  conf.py
+
+ignore =


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

+        flow_control (~.pubsub_v1.types.FlowControl): The flow control
+            settings. Use this to prevent situations where you are
+            inundated with too many messages at once.
+        policy_class (class): A class that describes how to handle


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

+            ~.pubsub_v1.subscriber.consumer.base.BaseConsumer: An instance
+                of the defined ``consumer_class`` on the client.
+        """
+        subscr = self._policy_class(self, subscription)


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

+
+import six
+
+__all__ = (


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

+    are free to use a different formula.
+
+    The precision of data stored is to the nearest integer. Additionally,
+    values outside the range of ``10 <= x <= 600`` are stored as ``10`` or


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

+        publish_time (datetime): The time that this message was originally
+            published.
+    """
+    def __init__(self, policy, ack_id, message):


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

+        # Actually start consuming messages.
+        self._consumer.start_consuming()
+
+        # Spawn a helper thread that maintains all of the leases for


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

+logger = logging.getLogger(__name__)
+
+
+class Policy(base.BasePolicy):


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

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


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

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


It is still in nox.py in the cover session.

…ogle-cloud-python into pubsub-subscriber

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

+            # This doesn't presently deal with exceptions that bubble up
+            # through the callback. If there is an error here, the thread will
+            # exit and no further queue items will be processed. We could
+            # potentially capture errors, log them, and then continue on.


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

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


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

+            abbv_data = abbv_data[0:50] + b'...'
+
+        # Return a useful representation.
+        answer = 'Message {\n'


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

+            receive any given message more than once.
+        """
+        time_to_ack = math.ceil(time.time() - self._received_timestamp)
+        self._request_queue.put(('ack', {


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

+        snooze = random.uniform(0.0, p99 * 0.9)
+        logger.debug('Snoozing lease management for %f seconds.' % snooze)
+        time.sleep(snooze)
+        self.maintain_leases()


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

+        # If we do, we need to stop the stream.
+        if self._load >= 1.0:
+            self._paused = True
+            self.close()


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

+        # In order to not thrash too much, require us to have passed below
+        # the resume threshold (80% by default) of each flow control setting
+        # before restarting.
+        if self._paused and self._load < self.flow_control.resume_threshold:


lukesneeringer · 2017-08-24T03:46:41Z

Merging this is blocked by CircleCI being hosed. Will try tomorrow morning.

lukesneeringer · 2017-08-24T03:47:06Z

Nevermind -- Pub/Sub passed. It failed on Spanner. Merging now.

lukesneeringer and others added 21 commits June 2, 2017 21:11

wip

0b55ff5

Wrote some docs; not much else.

210ef3b

subscriber wip

7cd6156

WIP

fc4ead3

wip

890de3a

wip

12ace0e

wip

7305000

Fix a couple minor lint issues.

e5a27ae

Adapting a subscriber that will work.

d50a22e

WIP

14f200a

Implement lease management.

6a7e846

WIP

3bb130b

WIP

f97dc23

WIP

303436c

WIP

933d2f3

WIP

1df0ccf

WIP

acb4534

Update subscriber client config to be sane.

2fb2785

Start adding unit tests.

ef178e9

Beginning work on unit tests.

147ad18

Merge branch 'pubsub-publisher' into pubsub-subscriber

c96367a

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

mdietz94 suggested changes Jul 20, 2017

View reviewed changes

theacodes suggested changes Jul 20, 2017

View reviewed changes

Luke Sneeringer added 6 commits July 25, 2017 09:02

Publisher tests complete.

9c701e3

subscriber/client.py tests

de38b83

Consumer tests

faeaa8e

Fix minor linting error.

d467719

Histogram tests

c821d33

Minor fix based on Max feedback.

ed750b2

Luke Sneeringer added 2 commits August 21, 2017 13:56

Fix a publish failure test.

8cb8f98

Fix final test.

47678c3

This was referenced Aug 21, 2017

PubSub Documentation #3848

Closed

Pub/Sub Docs #3849

Merged

Luke Sneeringer added 6 commits August 21, 2017 14:43

Sane max_workers default for 2.7 and 3.4

90ef40f

Mock credentials appropriately.

34c8273

Remove fail_under from .coveragerc.

831fe75

It is still in nox.py in the cover session.

Make histogram and helper_threads private.

02fa81f

Add a publishing system test.

2458b55

Merge branch 'pubsub-subscriber' of github.com:GoogleCloudPlatform/go…

aa56a9b

…ogle-cloud-python into pubsub-subscriber

theacodes suggested changes Aug 22, 2017

View reviewed changes

Luke Sneeringer and others added 14 commits August 22, 2017 10:17

Subscription system test.

17b6544

Merge branch 'pubsub-publisher' into pubsub-subscriber

f469381

Update tests.

a24c0a7

Make the wait test work on 2.7

a054324

Discarding unused mocks.

3735da5

Make _consumer a private module.

ac9f182

Switch from recursion to while for maintain_leases.

f272eca

Add exception logging in the callback.

e6bcbe7

Fix a long line. Whups.

b9115ca

Accept an executor.

6ae46cb

Fix a minor flake8 complaint.

852438e

No longer need to use inf for exception timeout.

13c2205

Fixes discussed in chat with @jonparrott.

6a03f48

Pub/Sub Docs (#3849)

f1dde8f

lukesneeringer merged commit 6fe1309 into pubsub-publisher Aug 24, 2017

lukesneeringer deleted the pubsub-subscriber branch August 24, 2017 03:47

lukesneeringer mentioned this pull request Aug 24, 2017

Pub/Sub API Redesign #3859

Merged

		logger = logging.getLogger(__name__)


		class Policy(base.BasePolicy):

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

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

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


		import six

		__all__ = (

Pubsub subscriber #3637

Pubsub subscriber #3637

Uh oh!

Conversation

lukesneeringer commented Jul 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

lukesneeringer commented Jul 20, 2017 •

edited

Loading