FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

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

/ google-cloud-python Public

Pubsub subscriber - #3637

Merged
lukesneeringer merged 74 commits into
pubsub-publisherfrom
pubsub-subscriber
Aug 24, 2017
ConversationCommits (74)ChecksFiles changed
Merged

Pubsub subscriber#3637
lukesneeringer merged 74 commits into
pubsub-publisherfrom
pubsub-subscriber

Conversation

lukesneeringer commented Jul 20, 2017 •
edited
Loading

Copy link
Copy Markdown
Contributor

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

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
Comment thread pubsub/google/cloud/pubsub_v1/subscriber/policy/base.py Outdated
Args:
ack_id (str): The ack ID.
"""
request = types.StreamingPullRequest(ack_ids=[ack_id])

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/policy/base.py Outdated
time as there are).
"""
return self._client.api.streaming_pull(request_generator,
options=gax.CallOptions(timeout=600),

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/policy/base.py
time.sleep(snooze)
self.maintain_leases()

def modify_ack_deadline(self, ack_id, seconds):

This comment was marked as spam.

Sign in to view

theacodes suggested changes Jul 20, 2017
View reviewed changes
Comment thread pubsub/.flake8 Outdated
*.pyc,
conf.py

ignore =

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread 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

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/client.py Outdated
~.pubsub_v1.subscriber.consumer.base.BaseConsumer: An instance
of the defined ``consumer_class`` on the client.
"""
subscr = self._policy_class(self, subscription)

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

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

import six

__all__ = (

This comment was marked as spam.

Sign in to view

Comment thread 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

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/message.py Outdated
publish_time (datetime): The time that this message was originally
published.
"""
def __init__(self, policy, ack_id, message):

This comment was marked as spam.

Sign in to view

Comment thread 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

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py
logger = logging.getLogger(__name__)


class Policy(base.BasePolicy):

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/helper_threads.py
@@ -0,0 +1,121 @@
# Copyright 2017, Google Inc. All rights reserved.

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/histogram.py
@@ -0,0 +1,155 @@
# Copyright 2017, Google Inc. All rights reserved.

This comment was marked as spam.

Sign in to view

This was referenced Aug 21, 2017
Closed
Merged
theacodes suggested changes Aug 22, 2017
View reviewed changes
Comment thread pubsub/google/cloud/pubsub_v1/subscriber/_helper_threads.py Outdated
# 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.

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/consumer.py
@@ -0,0 +1,266 @@
# Copyright 2017, Google Inc. All rights reserved.

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/message.py
abbv_data = abbv_data[0:50] + b'...'

# Return a useful representation.
answer = 'Message {\n'

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread 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', {

This comment was marked as spam.

Sign in to view

Comment thread pubsub/google/cloud/pubsub_v1/subscriber/policy/base.py Outdated
snooze = random.uniform(0.0, p99 * 0.9)
logger.debug('Snoozing lease management for %f seconds.' % snooze)
time.sleep(snooze)
self.maintain_leases()

This comment was marked as spam.

Sign in to view

Comment thread 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()

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread 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:

This comment was marked as spam.

Sign in to view

lukesneeringer commented Aug 24, 2017

Copy link
Copy Markdown
Contributor Author

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

lukesneeringer commented Aug 24, 2017

Copy link
Copy Markdown
Contributor Author

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

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
Merged
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

2 more reviewers
theacodes theacodes requested changes
mdietz94 mdietz94 requested changes
Reviewers whose approvals may not affect merge requirements

Assignees

No one assigned

Labels

cla: yes This human has signed the Contributor License Agreement.

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

5 participants

Back | FazBrowse Home | New Git URL