Wait 90 seconds between ACK messages

google_nest_sdm/subscriber_client.py

@@ -27,7 +27,8 @@
 _T = TypeVar("_T")
 
 RPC_TIMEOUT_SECONDS = 10.0
-STREAM_ACK_TIMEOUT_SECONDS = 30
+STREAM_ACK_TIMEOUT_SECONDS = 180
+STREAM_ACK_FREQUENCY_SECONDS = 90
 
 
 def refresh_creds(creds: Credentials) -> Credentials:
@@ -50,13 +51,12 @@ def refresh_creds(creds: Credentials) -> Credentials:
     return creds
 
 
-def exception_handler[
-    _T: Any
-](func_name: str) -> Callable[..., Callable[..., Awaitable[_T]]]:
+def exception_handler[_T: Any](
+    func_name: str,
+) -> Callable[..., Callable[..., Awaitable[_T]]]:
     """Wrap a function with exception handling."""
 
     def wrapped(func: Callable[..., Awaitable[_T]]) -> Callable[..., Awaitable[_T]]:
-
         async def wrapped_func(*args: Any, **kwargs: Any) -> _T:
             try:
                 async with asyncio.timeout(RPC_TIMEOUT_SECONDS):
@@ -106,10 +106,13 @@ async def pull_request_generator(
         stream_ack_deadline_seconds=STREAM_ACK_TIMEOUT_SECONDS,
     )
     while True:
+        ids = ack_ids_generator()
+        _LOGGER.debug("Sending streaming pull request (acking %s messages)", len(ids))
         yield pubsub_v1.StreamingPullRequest(
             stream_ack_deadline_seconds=STREAM_ACK_TIMEOUT_SECONDS,
-            ack_ids=ack_ids_generator(),
+            ack_ids=ids,
         )
+        await asyncio.sleep(STREAM_ACK_FREQUENCY_SECONDS)
 
 
 async def aiter_exception_handler(iterable: AsyncIterable[_T]) -> AsyncIterable[_T]:

tests/test_subscriber_client.py

@@ -15,7 +15,7 @@
 from google_nest_sdm.subscriber_client import (
     refresh_creds,
     SubscriberClient,
-    pull_request_generator
+    pull_request_generator,
 )
 
 
@@ -70,6 +70,7 @@ async def test_ack_messages() -> None:
         subscription="test", ack_ids=["message1", "message2"]
     )
 
+
 async def test_streaming_pull() -> None:
     """Test streaming pull call."""
 
@@ -135,21 +136,24 @@ async def test_request_generator() -> None:
         ["ack-id-3", "ack-id-4"],
         [],
     ]
-    stream = pull_request_generator("projects/some-project-id/subscriptions/sub-1", lambda: ack_ids.pop(0))
-    stream_iter = aiter(stream)
-    request = await anext(stream_iter)
-    assert request.subscription == "projects/some-project-id/subscriptions/sub-1"
-    assert request.stream_ack_deadline_seconds == 30
-    assert not request.ack_ids
-
-    request = await anext(stream_iter)
-    assert request.subscription == ""
-    assert request.stream_ack_deadline_seconds == 30
-    assert request.ack_ids == ["ack-id-1", "ack-id-2"]
-
-    request = await anext(stream_iter)
-    assert request.subscription == ""
-    assert request.stream_ack_deadline_seconds == 30
-    assert request.ack_ids == ["ack-id-3", "ack-id-4"]
-
-    await stream.aclose()
+    with patch("asyncio.sleep", return_value=None):
+        stream = pull_request_generator(
+            "projects/some-project-id/subscriptions/sub-1", lambda: ack_ids.pop(0)
+        )
+        stream_iter = aiter(stream)
+        request = await anext(stream_iter)
+        assert request.subscription == "projects/some-project-id/subscriptions/sub-1"
+        assert request.stream_ack_deadline_seconds == 180
+        assert not request.ack_ids
+
+        request = await anext(stream_iter)
+        assert request.subscription == ""
+        assert request.stream_ack_deadline_seconds == 180
+        assert request.ack_ids == ["ack-id-1", "ack-id-2"]
+
+        request = await anext(stream_iter)
+        assert request.subscription == ""
+        assert request.stream_ack_deadline_seconds == 180
+        assert request.ack_ids == ["ack-id-3", "ack-id-4"]
+
+        await stream.aclose()