Pubsub batch autocommitting. #2966

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.

pubsub/google/cloud/pubsub/topic.py

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

pubsub/google/cloud/pubsub/topic.py

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


pubsub/google/cloud/pubsub/topic.py

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


pubsub/google/cloud/pubsub/topic.py

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


pubsub/google/cloud/pubsub/topic.py

+        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()


pubsub/google/cloud/pubsub/topic.py

+        # 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()


pubsub/unit_tests/test_topic.py

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


pubsub/unit_tests/test_topic.py

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


pubsub/unit_tests/test_topic.py

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


pubsub/unit_tests/test_topic.py

+        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())


pubsub/unit_tests/test_topic.py

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


pubsub/google/cloud/pubsub/topic.py


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


pubsub/google/cloud/pubsub/topic.py

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


pubsub/google/cloud/pubsub/topic.py

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


pubsub/unit_tests/test_topic.py

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


pubsub/unit_tests/test_topic.py

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

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

pubsub/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:master Jan 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

Pubsub batch autocommitting. #2966

Pubsub batch autocommitting. #2966

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