Pubsub subscriber by lukesneeringer · Pull Request #3637 · googleapis/google-cloud-python

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

Sign in to view

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


Sign in to view

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


Sign in to view

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


Sign in to view

 *.pyc,
 conf.py
+
+ignore =


Sign in to view

+ 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


Sign in to view

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


Sign in to view

+
+import six
+
+__all__ = (


Sign in to view

+ 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


Sign in to view

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


Sign in to view

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


Sign in to view

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


Sign in to view

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


Sign in to view

@@ -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

Sign in to view

+ # 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.


Sign in to view

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


Sign in to view

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


Sign in to view

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


Sign in to view

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


Sign in to view

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


Sign in to view

+ # 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.

lukesneeringerand 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 Sneeringerand 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-publisherAug 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__ = (

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.

Uh oh!

This comment was marked as spam.

lukesneeringer commented Jul 20, 2017•
edited
Loading