Checkout Flow. It provides a way to do this sort of “staged” and multi process streaming. Flow — Flow v1.2.4
benwilson512
Author of Craft GraphQL APIs in Elixir with Absinthe
lud
Thank you 
I was more searching a way to do that with the Stream module. I’m not ready to pull a library just for this task.
But I guess this is not natively supported.
0
D4no0
That library is released by elixir core team to cover exactly your use-case, this is the middle library between Stream and GenStage, I completely forgot about it.
2
lud
D4no0
Damn, you are right!
For some reason I always thought that Flow was released before GenStage.
0
benwilson512
Author of Craft GraphQL APIs in Elixir with Absinthe
I think this task is a lot more difficult than you’re saying. The main challenges are around managing buffer and minimizing stall. If you do it as you’re describing you’ll OOM with large inputs, and you’ll also limit yourself to a single core while it works on that phase.
There was extensive discussion around the use of Stream for this purpose around ~2015 (it was actually the subject of my first ElixirConf talk) and ultimately Flow was the answer. As @D4no0 notes Flow is less “a library” and more of an “external stdlib”, I definitely recommend it for this purpose.
4
dimitarvp
If you are adamant about hand-rolling this for your needs then you’ll very likely reinvent Flow at one point. 
But if you have the time and the will then it might be a precious learning experience.
2
lud
Well I just need the first stream to be ran as quickly as possible, so it’s actually very simple and boils down to that:
parent = self()
spawn_link(fn ->
source_of_items()
|> Stream.map(&computation_1/1)
|> Enum.each(fn result ->
send(parent, {:result, result})
end)
send(parent, :done)
end)
Stream.unfold(nil, fn _ ->
receive do
{:result, result} -> {result, nil}
:done -> nil
end
end)
|> Stream.map(&computation_2/1)
|> Stream.run()
0
benwilson512
Author of Craft GraphQL APIs in Elixir with Absinthe
The first stream isn’t doing anything, you could simply Enum.each, call the computation, and send it to get a bit less overhead.
And actually looking at it more closely, this seems like a less optimal version of Task.async_stream. Particularly if you use the ordered: false option, that’s going to do the same thing (compute operation 1 in other processes) but without the memory bloat of doing it all ahead of time whether operation 2 can consume it or not.
4
lud
Yeah that firt stream call is supposed to represent a pipeline :S
I do not want async stream, the computation is heavy, I wanted to do it as fast as possible but one at a time. There is a single spawn here, so order is preserved.
The second stream is supposed to consume it all but I agree if there was a take(10) down below it would be nice to use it. Though it would require more than one additional process.
Edit: hmm i wonder if async_stream is greedy within its max concurrency limit. I guess no because it still requires something to pull from the stream.
0