How to make Task.async_stream return results as they are ready?

How to make Task.async_stream return results as they are ready?

I have the following code:

defmodule Demo do
  def run do
    child =
      spawn(fn ->
        Stream.resource(
          fn -> :ok end,
          fn _ ->
            receive do
              {:msg, msg} ->
                IO.puts("# received #{inspect(msg)}")
                {[msg], nil}

              _ ->
                {[], nil}
            end
          end,
          fn _ -> :ok end
        )
        |> Stream.each(fn x ->
          IO.puts("## before #{inspect(x)}")
          x
        end)
        |> Task.async_stream(
          fn x ->
            IO.puts("### during #{inspect(x)}")
            x
          end,
          ordered: false,
          max_concurrency: 2
        )
        |> Stream.each(fn x ->
          IO.puts("#### after #{inspect(x)}")
          x
        end)
        |> Stream.run()
      end)

    1..3
    |> Enum.each(fn x -> send(child, {:msg, x}) end)
  end
end

When run, I get the following output:

iex(6)> Demo.run
# received 1
## before 1
# received 2
### during 1
## before 2
#### after {:ok, 1}
### during 2
# received 3
## before 3
### during 3
#### after {:ok, 2}

The Stream.each after the Task.async_stream is never called with the last message. Async_stream seems to withhold output until it has max_concurrency results ready.

Is there any way to make to avoid the buffering?

Background: I’m building a data processing pipeline that processes endless stream of data and cleanup after each task has to happen immediately after its done. I could do it in the task itself, but that would break encapsulation and force me into complicating things with process messaging. I previously used GenStage, which worked well at small scale, but could use performance improvement as no backpressure is needed.