Add support for streaming pull receipts. (googleapis#4878)

chemelnucfin · web-flow · commit 60d4460d13c2 · 2018-02-21T13:31:35.000-08:00
* Closes (googleapis#4325) - PubSub: Implement StreamingPullRequest modify acknowledgement as message receipt * Review Changes * Minor fixups * Minor fixups
diff --git a/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py b/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py
@@ -329,8 +329,20 @@ def on_exception(self, exception):
     def on_response(self, response):
         """Process all received Pub/Sub messages.
 
-        For each message, schedule a callback with the executor.
+        For each message, send a modified acknowledgement request to the
+        server. This prevents expiration of the message due to buffering by
+        gRPC or proxy/firewall. This makes the server and client expiration
+        timer closer to each other thus preventing the message being
+        redelivered multiple times.
+
+        After the messages have all had their ack deadline updated, execute
+        the callback for each message using the executor.
         """
+        items = [
+            base.ModAckRequest(message.ack_id, self.histogram.percentile(99))
+            for message in response.received_messages
+        ]
+        self.modify_ack_deadline(items)
         for msg in response.received_messages:
             _LOGGER.debug(
                 'Using %s to process message with ack_id %s.',
diff --git a/pubsub/tests/unit/pubsub_v1/subscriber/test_policy_thread.py b/pubsub/tests/unit/pubsub_v1/subscriber/test_policy_thread.py
@@ -209,9 +209,17 @@ def test_on_response():
         ],
     )
 
-    # Actually run the method and prove that executor.submit and
-    # future.add_done_callback were called in the expected way.
-    policy.on_response(response)
+    # Actually run the method and prove that modack and executor.submit
+    # are called in the expected way.
+    modack_patch = mock.patch.object(
+        policy, 'modify_ack_deadline', autospec=True)
+    with modack_patch as modack:
+        policy.on_response(response)
+
+    modack.assert_called_once_with(
+        [base.ModAckRequest('fack', 10),
+         base.ModAckRequest('back', 10)]
+    )
 
     submit_calls = [m for m in executor.method_calls if m[0] == 'submit']
     assert len(submit_calls) == 2
