Fix an issue with client connection errors raised from initial request
The issue is not with the stream itself broken, but with the initial
streaming handshake broken, e.g. due to `ServerDisconnectedError`.
This kind of errors should be treated the same as the streaming error:
i.e., by ignoring them and restarting the stream request.
@@ -207,7 +207,9 @@ async def watch_objs(
if timeout is not None:
params['timeoutSeconds'] = str(timeout)
- # Talk to the API and initiate a streaming response.
+ # Stream the parsed events from the response until it is closed server-side,
+ # or until it is closed client-side by the freeze-waiting future's callbacks.
+ try:
response = await context.session.get(
url=resource.get_url(server=context.server, namespace=namespace, params=params),
timeout=aiohttp.ClientTimeout(
@@ -217,8 +219,6 @@ async def watch_objs(
)
response.raise_for_status()
- # Stream the parsed events from the response until it is closed server-side,
- # or until it is closed client-side by the freeze-waiting future's callbacks.
response_close_callback = lambda _: response.close()
freeze_waiter.add_done_callback(response_close_callback)
try:
@@ -226,11 +226,12 @@ async def watch_objs(
async for line in _iter_jsonlines(response.content):
raw_input = cast(bodies.RawInput, json.loads(line.decode("utf-8")))
yield raw_input
- except (aiohttp.ClientConnectionError, aiohttp.ClientPayloadError):
- pass
finally:
freeze_waiter.remove_done_callback(response_close_callback)
+ except (aiohttp.ClientConnectionError, aiohttp.ClientPayloadError):
+ pass
+
async def _iter_jsonlines(
content: aiohttp.StreamReader,
