Closed DurandA closed 1 year ago
@DurandA I am not sure whether I understand your post properly, but I do not see the direct relation with sse-starlette. If you can be a bit more concrete, please feel free to reopen.
The issue was due to the EventSourceResponse cancelling the task from the asynchronous iterator.
The fix was to "shield" the iterator as follows:
async def event_publisher():
aiter = pubsub.__aiter__()
try:
while True:
task = asyncio.create_task(aiter.__anext__())
event = await asyncio.shield(task)
yield dict(data=event)
except asyncio.CancelledError as e:
print(f"Disconnected from client (via refresh/close) {request.client}")
# Do any other cleanup, if any
raise e
I suppose that the task is cancelled here: https://github.com/sysid/sse-starlette/blob/30ef55c08a8b1512b625752be136a1ea67df6030/sse_starlette/sse.py#L229-L236
Thanks for sharing your experience. I am glad that you could find a solution.
Is there anything that can be improved on sse-starlette's side?
I am using a simple publish/subscribe pattern with FastAPI in order to broadcast data to clients using SSE:
However, the task "my_task" is somehow killed as soon as the first client disconnects:
I also tried with other patterns, such as using
AsyncIteratorObserver
fromaioreactive
with the same result: the task linked to the async iterator ends up with anInvalidStateError
.