Is there a reason why Task.async_stream would not work? It seems to support all the things you’ve mentioned.
Task.async_stream