This feature is now in master: 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.

I would also be interested in Flow getting better first class support for flows/graphs that incorporate external GenStages (or maybe just GenStage MFAs if control over materialization is necessary).

These days I end up manually composing the few custom GenStages that I need, plus a custom Enum GenStage that is a passthrough :producer_consumer + side effects.

The system works, but feels backwards. I’d much rather use a high level Flow/Graph DSL. And would welcome any enhancements to Flow that would make it easier to BYO GenStage in a complete execution graph (not just as flow source/sink).

Note: Some may argue that the need for custom GenStages ought to be rare. Maybe. But the need is still there. There are times you want/need an async boundary. There are times you want/need to deal with differences in flow rates. There are times you want/need to manage state and would rather do it in a GenServery way instead of a functional reducer way.

I’d love for Flow to incorporate more support for these types of data flows. The GenStage implementation of these things tends to then influence the DSL. More mature flow DSLs support the notion of explicit async boundaries, sub-flows, explicit buffers, timers and other rate control mechanisms, among other things. All of these can be implemented via a graph of GenStages. Optimizations can then be made to avoid extra stages and communication when unnecessary. For example, a series of Flow.map transformations could be made as a series of :producer_consumer GenStages, but usually it is more efficient to compose all those functions together. Flow makes that optimization already today.

Yes, this should work. However, I am curious to know why you need to drop Flow in favor of GenStage?