This feature is now in master: Provide a better mechanism for external GenStages to participate in Flows · Issue #52 · dashbitco/flow · GitHub

https://github.com/elixir-lang/flow/issues/52

Feel free to tweak title. Naming things is hard.

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.