Add async_iterable_to_stream utility to avoid aiohttp compatibility issues

Create async_iterable_to_stream() in gns3server.utils.asyncio that
converts an async iterable to an aiohttp StreamReader via a background
feeder task. This bypasses aiohttp's AsyncIterablePayload which can
cause 'Connection reset by peer' with certain HTTP servers.

Use it in _run_http_query for the __aiter__ data path.
This commit is contained in:
YueGuobin 2026-06-09 23:55:59 +08:00
parent 8e340f0ce8
commit e659b64bf0
No known key found for this signature in database
2 changed files with 43 additions and 1 deletions

View File

@ -32,7 +32,7 @@ else:
from async_timeout import timeout as asynctimeout
from ..utils import parse_version
from ..utils.asyncio import locking
from ..utils.asyncio import locking, async_iterable_to_stream
from ..controller.controller_error import (
ControllerError,
ControllerBadRequestError,
@ -537,6 +537,7 @@ class Compute:
elif hasattr(data, "__aiter__"):
chunked = True
headers["content-type"] = "application/octet-stream"
data = await async_iterable_to_stream(data)
# If the data is an open file we will iterate on it
elif isinstance(data, io.BufferedIOBase):
chunked = True

View File

@ -136,3 +136,44 @@ def locking(f):
return await f(oself, *args, **kwargs)
return wrapper
async def async_iterable_to_stream(async_iter, limit=65536):
"""
Convert an async iterable into an aiohttp StreamReader.
This avoids passing async generators directly to aiohttp's payload
system, which can cause compatibility issues with certain HTTP servers.
:param async_iter: An async iterable that yields bytes
:param limit: Buffer limit for the StreamReader (default 64KB)
:returns: aiohttp.streams.StreamReader
"""
from aiohttp.streams import StreamReader
class _NoopProtocol:
_reading_paused = False
connected = True
def pause_reading(self):
self._reading_paused = True
def resume_reading(self, resume_parser=False):
self._reading_paused = False
reader = StreamReader(_NoopProtocol(), limit=limit)
async def _feed():
try:
async for chunk in async_iter:
reader.feed_data(chunk)
except GeneratorExit:
raise
except Exception:
pass
finally:
reader.feed_eof()
asyncio.ensure_future(_feed())
return reader