Add support for auto-acknowledging pulled messages.

Follows @tmatsuo's suggested implementation in:
#798 (comment)

Closes #798.

Author: tseaver

docs/pubsub-usage.rst

@@ -272,3 +272,30 @@ Fetch messages for a pull subscription without blocking (none pending):
    >>> messages = [recv[1] for recv in received]
    >>> [message.message_id for message in messages]
    []
+
+Fetch pending messages, acknowledging those whose processing doesn't raise an
+error:
+
+.. doctest::
+
+   >>> from gcloud import pubsub
+   >>> client = pubsub.Client()
+   >>> topic = client.topic('topic_name')
+   >>> subscription = topic.subscription('subscription_name')
+   >>> with topic.batch() as batch:
+   ...     batch.publish('this is the first message_payload')
+   ...     batch.publish('this is the second message_payload',
+   ...                   attr1='value1', attr2='value2')
+   >>> from gcloud.pubsub.subscription import AutoAck
+   >>> with AutoAck(subscription, max_messages=10) as ack:
+   ...     for ack_id, message in ack.items():
+   ...        try:
+   ...            do_something_with(message)
+   ...        except:
+   ...            del ack[ack_id]
+
+.. note::
+
+   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``

gcloud/pubsub/subscription.py

@@ -263,3 +263,51 @@ def delete(self, client=None):
         """
         client = self._require_client(client)
         client.connection.api_request(method='DELETE', path=self.path)
+
+
+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__`.  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

@@ -485,6 +485,81 @@ def test_delete_w_alternate_client(self):
         self.assertEqual(req['path'], '/%s' % SUB_PATH)
 
 
+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 _Connection(object):
 
     def __init__(self, *responses):
@@ -522,3 +597,30 @@ def __init__(self, project, connection=None):
     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)