When using httpx.ASGITransport, aconnect_sse doesn't yield server sent events for an infinite AsyncGenerator

[will-afs]

Thanks for implementing this library, which might be very useful for testing SSE endpoints.

However, it seems the aconnect_sse context manager does not function as expected.

Let's consider this first example:

import asyncio

import httpx
from httpx_sse import aconnect_sse
from sse_starlette.sse import EventSourceResponse
from starlette.applications import Starlette
from starlette.routing import Route

async def auth_events(request):
    async def events():
        for i in range(5):
            yield {
                "event": "login",
                "data": '{"user_id": "4135"}',
            }
            await asyncio.sleep(1)

    return EventSourceResponse(events())

app = Starlette(routes=[Route("/sse/auth/", endpoint=auth_events)])

async def main():
    async with httpx.AsyncClient(app=app) as client:
        async with aconnect_sse(
            client, "http://localhost:8000/sse/auth/"
        ) as event_source:
            events = [sse async for sse in event_source.aiter_sse()]

asyncio.run(main())

The events = [sse async for sse in event_source.aiter_sse()] line will only be executed after 5s (i.e. the time the events() function gets executed).

Now, let's consider the following example, where the AsyncGenerator yields objects indefinitely (which should be the use case of any SSE endpoint):

import asyncio

import httpx
from httpx_sse import aconnect_sse
from sse_starlette.sse import EventSourceResponse
from starlette.applications import Starlette
from starlette.routing import Route

async def auth_events(request):
    async def events():
        while True:
            yield {
                "event": "login",
                "data": '{"user_id": "4135"}',
            }
            await asyncio.sleep(1)

    return EventSourceResponse(events())

app = Starlette(routes=[Route("/sse/auth/", endpoint=auth_events)])

async def main():
    async with httpx.AsyncClient(app=app) as client:
        async with aconnect_sse(
            client, "http://localhost:8000/sse/auth/"
        ) as event_source:
            events = [sse async for sse in event_source.aiter_sse()]

asyncio.run(main())

In this second example, the events = [sse async for sse in event_source.aiter_sse()] line will never be reached because we indefinitely yield from events().
I guess this behaviour is not expected since SSE endpoints are made by design to yield an infinite number of events.
Any solution to this?

[florimondmanca]

Hi,

My assumption would be that this is related to HTTPX's ASGI streaming support, which has not landed yet. encode/httpx#2186

As a result, the ASGITransport reads everything into memory and then yields a single body chunk which aiter_sse() consumes in parts.

If you start the app using Uvicorn and access it through a normal network connection, do you confirm this behavior does not reproduce?

[will-afs]

Hello, thanks for your answer!

Yes, I confirm this issue doesn't reproduce web the app is launched using Uvicorn and accessed via a web browser for instance. So indeed, it seems closely related to the issue you mentioned. Do you have any idea if these works are currently in progress? Let me know if I can help.

[florimondmanca]

You can find an open draft PR of mine on the HTTPX repo. I’d be curious to see if installing httpx from that branch resolves the issue you’re seeing!