Merge pull request #1636 from tseaver/798-tmatsuo_strategy

#798: Add support for auto-acknowledging pulled messages.

docs/pubsub-usage.rst

@@ -158,6 +158,24 @@ Re-synchronize a subscription with the back-end:
    :start-after: [START subscription_reload]
    :end-before: [END subscription_reload]
 
+Fetch the IAM policy for a subscription
+
+.. literalinclude:: pubsub_snippets.py
+   :start-after: [START subscription_get_iam_policy]
+   :end-before: [END subscription_get_iam_policy]
+
+Update the IAM policy for a subscription:
+
+.. literalinclude:: pubsub_snippets.py
+   :start-after: [START subscription_set_iam_policy]
+   :end-before: [END subscription_set_iam_policy]
+
+Test permissions allowed by the current IAM policy on a subscription:
+
+.. literalinclude:: pubsub_snippets.py
+   :start-after: [START subscription_check_iam_permissions]
+   :end-before: [END subscription_check_iam_permissions]
+
 Delete a subscription:
 
 .. literalinclude:: pubsub_snippets.py
@@ -193,20 +211,15 @@ Update the acknowlegement deadline for pulled messages:
    :start-after: [START subscription_modify_ack_deadline]
    :end-before: [END subscription_modify_ack_deadline]
 
-Fetch the IAM policy for a subscription
-
-.. literalinclude:: pubsub_snippets.py
-   :start-after: [START subscription_get_iam_policy]
-   :end-before: [END subscription_get_iam_policy]
-
-Update the IAM policy for a subscription:
+Fetch pending messages, acknowledging those whose processing doesn't raise an
+error:
 
 .. literalinclude:: pubsub_snippets.py
-   :start-after: [START subscription_set_iam_policy]
-   :end-before: [END subscription_set_iam_policy]
+   :start-after: [START subscription_pull_autoack]
+   :end-before: [END subscription_pull_autoack]
 
-Test permissions allowed by the current IAM policy on a subscription:
+.. note::
 
-.. literalinclude:: pubsub_snippets.py
-   :start-after: [START subscription_check_iam_permissions]
-   :end-before: [END subscription_check_iam_permissions]
+   The ``pull`` API request occurs at entry to the ``with`` block, and the
+   ``acknowlege`` API request occurs at the end, passing only the ``ack_ids``
+   which haven't been deleted from ``ack``

docs/pubsub_snippets.py

@@ -178,21 +178,6 @@ def topic_publish_messages(client, to_delete):
     # [END topic_publish_message_with_attrs]
 
 
-@snippet
-def topic_batch(client, to_delete):
-    """Publish multiple messages in a single request."""
-    TOPIC_NAME = 'topic_batch-%d' % (_millis(),)
-    topic = client.topic(TOPIC_NAME)
-    topic.create()
-    to_delete.append(topic)
-
-    # [START topic_batch]
-    with topic.batch() as batch:
-        batch.publish(b'This is the message payload')
-        batch.publish(b'Another message payload', extra='EXTRA')
-    # [END topic_batch]  API request on block exit
-
-
 @snippet
 def topic_subscription(client, to_delete):
     """Create subscriptions to a topic."""
@@ -358,6 +343,52 @@ def log_exception(_):
         (extras,))
 
 
+@snippet
+def subscription_pull_w_autoack(client, to_delete):
+    """Pull messges from a topic, auto-acknowldging them"""
+    TOPIC_NAME = 'subscription_pull_autoack-%d' % (_millis(),)
+    SUB_NAME = 'subscription_pull_autoack-defaults-%d' % (_millis(),)
+    PAYLOAD1 = b'PAYLOAD1'
+    PAYLOAD2 = b'PAYLOAD2'
+    EXTRA = 'EXTRA'
+    topic = client.topic(TOPIC_NAME)
+    topic.create()
+    to_delete.append(topic)
+
+    subscription = topic.subscription(SUB_NAME)
+    subscription.create()
+    to_delete.append(subscription)
+
+    # [START topic_batch]
+    with topic.batch() as batch:
+        batch.publish(PAYLOAD1)
+        batch.publish(PAYLOAD2, extra=EXTRA)
+    # [END topic_batch]
+
+    time.sleep(1)  # eventually-consistent
+
+    payloads = []
+    extras = []
+
+    def do_something_with(message):  # pylint: disable=unused-argument
+        payloads.append(message.data)
+        if message.attributes:
+            extras.append(message.attributes)
+
+    # [START subscription_pull_autoack]
+    from gcloud.pubsub.subscription import AutoAck
+    with AutoAck(subscription, max_messages=10) as ack:
+        for ack_id, message in list(ack.items()):
+            try:
+                do_something_with(message)
+            except Exception:  # pylint: disable=broad-except
+                del ack[ack_id]
+    # [END subscription_pull_autoack]
+
+    assert set(payloads) == set(PAYLOAD1, PAYLOAD1), "eventual consistency"
+    assert extras == [{'extra': EXTRA}], "eventual consistency"
+
+
 @snippet
 def subscription_iam_policy(client, to_delete):
     """Fetch / set a subscription's IAM policy."""

gcloud/pubsub/subscription.py

@@ -123,6 +123,24 @@ def path(self):
         """URL path for the subscription's APIs"""
         return '/%s' % (self.full_name,)
 
+    def auto_ack(self, return_immediately=False, max_messages=1, client=None):
+        """:class:`AutoAck` factory
+
+        :type return_immediately: boolean
+        :param return_immediately: passed through to :meth:`Subscription.pull`
+
+        :type max_messages: int
+        :param max_messages: passed through to :meth:`Subscription.pull`
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: passed through to :meth:`Subscription.pull` and
+                      :meth:`Subscription.acknowledge`.
+
+        :rtype: :class:`AutoAck`
+        :returns: the instance created for the given ``ack_id`` and ``message``
+        """
+        return AutoAck(self, return_immediately, max_messages, client)
+
     def _require_client(self, client):
         """Check client or verify over-ride.
 
@@ -420,3 +438,53 @@ def check_iam_permissions(self, permissions, client=None):
         api = client.iam_policy_api
         return api.test_iam_permissions(
             self.full_name, list(permissions))
+
+
+class AutoAck(dict):
+    """Wrapper for :meth:`Subscription.pull` results.
+
+    Mapping, tracks messages still-to-be-acknowledged.
+
+    When used as a context manager, acknowledges all messages still in the
+    mapping on `__exit__`.  When processing the pulled messsages, application
+    code MUST delete messages from the :class:`AutoAck` mapping which are not
+    successfully processed, e.g.:
+
+    .. code-block: python
+
+       with AutoAck(subscription) as ack:  # calls ``subscription.pull``
+           for ack_id, message in ack.items():
+               try:
+                   do_something_with(message):
+               except:
+                   del ack[ack_id]
+
+    :type subscription: :class:`Subscription`
+    :param subscription: subcription to be pulled.
+
+    :type return_immediately: boolean
+    :param return_immediately: passed through to :meth:`Subscription.pull`
+
+    :type max_messages: int
+    :param max_messages: passed through to :meth:`Subscription.pull`
+
+    :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+    :param client: passed through to :meth:`Subscription.pull` and
+                   :meth:`Subscription.acknowledge`.
+    """
+    def __init__(self, subscription,
+                 return_immediately=False, max_messages=1, client=None):
+        super(AutoAck, self).__init__()
+        self._subscription = subscription
+        self._return_immediately = return_immediately
+        self._max_messages = max_messages
+        self._client = client
+
+    def __enter__(self):
+        items = self._subscription.pull(
+            self._return_immediately, self._max_messages, self._client)
+        self.update(items)
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self._subscription.acknowledge(list(self), self._client)

gcloud/pubsub/test_subscription.py

@@ -134,16 +134,40 @@ def test_from_api_repr_w_topics_w_topic_match(self):
 
     def test_full_name_and_path(self):
         PROJECT = 'PROJECT'
-        SUB_NAME = 'sub_name'
-        SUB_FULL = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        SUB_FULL = 'projects/%s/subscriptions/%s' % (PROJECT, self.SUB_NAME)
         SUB_PATH = '/%s' % (SUB_FULL,)
         TOPIC_NAME = 'topic_name'
         CLIENT = _Client(project=PROJECT)
         topic = _Topic(TOPIC_NAME, client=CLIENT)
-        subscription = self._makeOne(SUB_NAME, topic)
+        subscription = self._makeOne(self.SUB_NAME, topic)
         self.assertEqual(subscription.full_name, SUB_FULL)
         self.assertEqual(subscription.path, SUB_PATH)
 
+    def test_autoack_defaults(self):
+        from gcloud.pubsub.subscription import AutoAck
+        client = _Client(project=self.PROJECT)
+        topic = _Topic(self.TOPIC_NAME, client=client)
+        subscription = self._makeOne(self.SUB_NAME, topic)
+        auto_ack = subscription.auto_ack()
+        self.assertTrue(isinstance(auto_ack, AutoAck))
+        self.assertTrue(auto_ack._subscription is subscription)
+        self.assertEqual(auto_ack._return_immediately, False)
+        self.assertEqual(auto_ack._max_messages, 1)
+        self.assertTrue(auto_ack._client is None)
+
+    def test_autoack_explicit(self):
+        from gcloud.pubsub.subscription import AutoAck
+        client1 = _Client(project=self.PROJECT)
+        client2 = _Client(project=self.PROJECT)
+        topic = _Topic(self.TOPIC_NAME, client=client1)
+        subscription = self._makeOne(self.SUB_NAME, topic)
+        auto_ack = subscription.auto_ack(True, 10, client2)
+        self.assertTrue(isinstance(auto_ack, AutoAck))
+        self.assertTrue(auto_ack._subscription is subscription)
+        self.assertEqual(auto_ack._return_immediately, True)
+        self.assertEqual(auto_ack._max_messages, 10)
+        self.assertTrue(auto_ack._client is client2)
+
     def test_create_pull_wo_ack_deadline_w_bound_client(self):
         RESPONSE = {
             'topic': self.TOPIC_PATH,
@@ -642,6 +666,81 @@ def subscription_modify_ack_deadline(self, subscription_path, ack_ids,
         return self._subscription_modify_ack_deadline_response
 
 
+class TestAutoAck(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.pubsub.subscription import AutoAck
+        return AutoAck
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor_defaults(self):
+        subscription = _FauxSubscription(())
+        auto_ack = self._makeOne(subscription)
+        self.assertEqual(auto_ack._return_immediately, False)
+        self.assertEqual(auto_ack._max_messages, 1)
+        self.assertTrue(auto_ack._client is None)
+
+    def test_ctor_explicit(self):
+        CLIENT = object()
+        subscription = _FauxSubscription(())
+        auto_ack = self._makeOne(
+            subscription, return_immediately=True, max_messages=10,
+            client=CLIENT)
+        self.assertTrue(auto_ack._subscription is subscription)
+        self.assertEqual(auto_ack._return_immediately, True)
+        self.assertEqual(auto_ack._max_messages, 10)
+        self.assertTrue(auto_ack._client is CLIENT)
+
+    def test___enter___w_defaults(self):
+        subscription = _FauxSubscription(())
+        auto_ack = self._makeOne(subscription)
+
+        with auto_ack as returned:
+            pass
+
+        self.assertTrue(returned is auto_ack)
+        self.assertEqual(subscription._return_immediately, False)
+        self.assertEqual(subscription._max_messages, 1)
+        self.assertTrue(subscription._client is None)
+
+    def test___enter___w_explicit(self):
+        CLIENT = object()
+        subscription = _FauxSubscription(())
+        auto_ack = self._makeOne(
+            subscription, return_immediately=True, max_messages=10,
+            client=CLIENT)
+
+        with auto_ack as returned:
+            pass
+
+        self.assertTrue(returned is auto_ack)
+        self.assertEqual(subscription._return_immediately, True)
+        self.assertEqual(subscription._max_messages, 10)
+        self.assertTrue(subscription._client is CLIENT)
+
+    def test___exit___(self):
+        CLIENT = object()
+        ACK_ID1, MESSAGE1 = 'ACK_ID1', _FallibleMessage()
+        ACK_ID2, MESSAGE2 = 'ACK_ID2', _FallibleMessage()
+        ACK_ID3, MESSAGE3 = 'ACK_ID3', _FallibleMessage(True)
+        ITEMS = [
+            (ACK_ID1, MESSAGE1),
+            (ACK_ID2, MESSAGE2),
+            (ACK_ID3, MESSAGE3),
+        ]
+        subscription = _FauxSubscription(ITEMS)
+        auto_ack = self._makeOne(subscription, client=CLIENT)
+        with auto_ack:
+            for ack_id, message in list(auto_ack.items()):
+                if message.fail:
+                    del auto_ack[ack_id]
+        self.assertEqual(sorted(subscription._acknowledged),
+                         [ACK_ID1, ACK_ID2])
+        self.assertTrue(subscription._ack_client is CLIENT)
+
+
 class _FauxIAMPolicy(object):
 
     def get_iam_policy(self, target_path):
@@ -677,3 +776,30 @@ def __init__(self, project):
     def topic(self, name, timestamp_messages=False):
         from gcloud.pubsub.topic import Topic
         return Topic(name, client=self, timestamp_messages=timestamp_messages)
+
+
+class _FallibleMessage(object):
+
+    def __init__(self, fail=False):
+        self.fail = fail
+
+
+class _FauxSubscription(object):
+
+    def __init__(self, items):
+        self._items = items
+        self._mapping = dict(items)
+        self._acknowledged = set()
+
+    def pull(self, return_immediately=False, max_messages=1, client=None):
+        self._return_immediately = return_immediately
+        self._max_messages = max_messages
+        self._client = client
+        return self._items
+
+    def acknowledge(self, ack_ids, client=None):
+        self._ack_client = client
+        for ack_id in ack_ids:
+            message = self._mapping[ack_id]
+            assert not message.fail
+            self._acknowledged.add(ack_id)