ream88
Flow into/from GenStage
I’m currently playing around with Flow and GenStage and wanted to know if something like this is possible:
def start_link(file) do
File.stream!(file)
|> Flow.from_enumerable()
|> Flow.partition()
|> Flow.into_stages([ProducerConsumer])
|> elem(1)
|> Flow.from_stage()
# more Flow stuff like |> Flow.filter
|> Flow.into_stages([Consumer])
end
This code does NOT work of course. My idea behind this was to have a clean and easy to understand flow and a bunch of GenStage modules which do the hard and complicated stuff.
Marked As Solved
ream88
I got the following to run, however @lackac could you be so kind to give me your opinion on it? (Still a GenStage/Flow novice here
)
def start_link(file) do
File.stream!(file)
|> Flow.from_enumerable()
# |> Flow.filter() ...
|> Flow.into_stages([ProducerConsumer])
ProducerConsumer
|> Flow.from_stage()
# more Flow stuff like |> Flow.filter
|> Flow.into_stages([Consumer])
end
Also Liked
josevalim
This feature is now in master: Provide a better mechanism for external GenStages to participate in Flows · Issue #52 · dashbitco/flow · GitHub
CptnKirk
https://github.com/elixir-lang/flow/issues/52
Feel free to tweak title. Naming things is hard.
lackac
Something like this could work, but the devil—as usually—is in the details.
First of all, there are some issues with the flow above, but I assume that’s because you didn’t want to include too much detail. For example, I don’t think calling Flow.into_stages/2 immediately after Flow.partition/1 makes sense.
Also consider if using Flow is really necessary. You’re already using GenStage for other tasks. While Flow is a simpler, higher level way of coordinating multiple stages of consumers, you might find that it’s unnecessary overhead if you don’t need that kind of coordination.
I assume you want to supervise this, so you’ll need to split it into separate modules too. So instead of
|> Flow.into_stages([ProducerConsumer])
|> elem(1)
|> Flow.from_stage()
you would have a supervisor with the following children:
children = [
ProducerConsumer,
Consumer,
{MyFlow1, {file, ProducerConsumer}},
{MyFlow2, {ProducerConsumer, Consumer}}
]
Then MyFlow1 and MyFlow2 could be modules that satisfy a behaviour that implements child_spec/1 and start_link/1 so that their main job would be to implement the flow specific aspects.
We have a similar kind of setup and so far it has worked out well. Let me know if you need more help with the specifics of the behaviour and the flow modules.
Popular in Questions
Other popular topics
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #javascript
- #code-sync
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #advent-of-code
- #elixirconf-us
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #performance








