Pubsub batch autocommitting. by lukesneeringer · Pull Request #2966 · googleapis/google-cloud-python

lukesneeringer · 2017-01-26T21:36:46Z

This PR adds some functionality to the Batch object:

The ability to specify max_messages and have the batch
automatically call commit when the number of messages
gets that high.
The ability to specify max_interval and have the batch
automatically commit when a publish occurs and the batch
is at least as old as the specified interval.

This is one of two changes requested by the PubSub team.

This PR adds some functionality to the Batch object: * The ability to specify `max_messages` and have the batch automatically call `commit` when the number of messages gets that high. * The ability to specify `max_interval` and have the batch automatically commit when a publish occurs and the batch is at least as old as the specified interval. This is one of two changes requested by the PubSub team.

Sign in to view

- def __init__(self, topic, client):
+ INFINITY = float('inf')
+
+ def __init__(self, topic, client, max_interval=INFINITY,


dhermes

Mostly LG, ping me once addressed / we discuss?

Sign in to view

+ :type max_messages: float
 """
- def __init__(self, topic, client):
+ INFINITY = float('inf')


Sign in to view

+ self._max_messages = max_messages
+
+ # Set the initial starting timestamp (used against the interval).
+ self._start_timestamp = float(time.time())


Sign in to view

 :type attrs: dict (string -> string)
 :param attrs: key-value pairs to send as message attributes
+
+ :rtype: None


Sign in to view

+ if self._max_interval < self.INFINITY:
+ if float(time.time()) - self._start_timestamp > self._max_interval:
+ self._start_timestamp = float(time.time())
+ return self.commit()


Sign in to view

+ # If the number of messages on the list is greater than the
+ # maximum allowed, autocommit (with the batch's client).
+ if len(self.messages) >= self._max_messages:
+ return self.commit()


Sign in to view

+ def test_message_count_autocommit(self):
+ """Establish that if the batch is assigned to take a maximum
+ number of messages, that it commits when it reaches that maximum.
+ """


Sign in to view

+ topic = _Topic(name='TOPIC')
+
+ # Track commits, but do not perform them.
+ Batch = self._get_target_class()


Sign in to view

+ for i in range(0, 4):
+ batch.publish({
+ 'attributes': {},
+ 'data': 'Batch message %d.' % i,


Sign in to view

+ Batch = self._get_target_class()
+ with mock.patch.object(Batch, 'commit') as commit:
+ with self._make_one(topic, client=client, max_messages=5) as batch:
+ self.assertIsInstance(batch, self._get_target_class())


Sign in to view

+ # Track commits, but do not perform them.
+ Batch = self._get_target_class()
+ with mock.patch.object(Batch, 'commit') as commit:
+ mock_time.return_value = 0.0


Sign in to view


+ # If too much time has elapsed since the first message
+ # was added, autocommit.
+ if self._max_interval < self.INFINITY:


Sign in to view

+
+ # If the number of messages on the list is greater than the
+ # maximum allowed, autocommit (with the batch's client).
+ if len(self.messages) >= self._max_messages:


Sign in to view

+ # maximum allowed, autocommit (with the batch's client).
+ if len(self.messages) >= self._max_messages:
+ self.commit()
+ return


Sign in to view

+ topic = _Topic(name='TOPIC')
+
+ # Track commits, but do not perform them.
+ Batch = self._get_target_class()


Sign in to view

+ # Track commits, but do not perform them.
+ Batch = self._get_target_class()
+ with mock.patch.object(Batch, 'commit') as commit:
+ mock_time.return_value = 0.0


dhermes

LGTM once CI goes green

@dhermes

* Pubsub batch autocommitting. This PR adds some functionality to the Batch object: * The ability to specify `max_messages` and have the batch automatically call `commit` when the number of messages gets that high. * The ability to specify `max_interval` and have the batch automatically commit when a publish occurs and the batch is at least as old as the specified interval. This is one of two changes requested by the PubSub team. * Addressing comments from @dhermes. * Remove unneeded -lt check @dhermes. * Make INFINITY have a leading underscore. @dhermes

@dhermes

* Pubsub batch autocommitting. This PR adds some functionality to the Batch object: * The ability to specify `max_messages` and have the batch automatically call `commit` when the number of messages gets that high. * The ability to specify `max_interval` and have the batch automatically commit when a publish occurs and the batch is at least as old as the specified interval. This is one of two changes requested by the PubSub team. * Addressing comments from @dhermes. * Remove unneeded -lt check @dhermes. * Make INFINITY have a leading underscore. @dhermes

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

lukesneeringer requested a review from dhermes January 26, 2017 21:38

theacodes reviewed Jan 26, 2017
View reviewed changes

Comment threadpubsub/google/cloud/pubsub/topic.py Outdated
def __init__(self, topic, client):
INFINITY = float('inf')

def __init__(self, topic, client, max_interval=INFINITY,
This comment was marked as spam.
Sign in to view
This comment was marked as spam.
Sign in to view

dhermes reviewed Jan 27, 2017
View reviewed changes

Addressing comments from@dhermes.
eab0b7b

dhermes reviewed Jan 27, 2017
View reviewed changes

lukesneeringer added 2 commits January 27, 2017 10:06

Remove unneeded -lt check@dhermes.
d35b490

Make INFINITY have a leading underscore.@dhermes
7cc6ea7

dhermes approved these changes Jan 27, 2017
View reviewed changes

lukesneeringer merged commit 50c8e88 into googleapis:masterJan 27, 2017

lukesneeringer deleted the pubsub-batching branch January 27, 2017 19:11

dhermes mentioned this pull request Feb 24, 2017
Upgrading all versions for umbrella release. #3068
Merged
19 tasks

Uh oh!

Conversation

lukesneeringer commented Jan 26, 2017

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

dhermes left a comment

Choose a reason for hiding this comment

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!

dhermes left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants