I am building a data processing pipeline which may sporadically fail. I would like to know when Producers/Consumers fail, so I can keep track of running pipelines. When I crash my producer, other processes crash with the same error message, which is confusing. Consider the following setup: defmodu…

I am now using Task.Supervisor.async_stream and a pipeline definition using Keyword lists. Something like this: stages = [ add_one: fn i -> i + 1 end, sub_one: fn i -> i - 1 end ] input_stream = 1..10 |> Enum.to_list() {:ok, sup} = Task.Supervisor.start_link(name: :tasksup) reduce_fn = &Enum.…

But this too will stop on the first return value that’s not {:ok, stuff}, is that your intention?

Sometimes yes, sometimes no :slight_smile: When failed items can be discarded the In others I would replace |> Stream.take_while(fn {:ok, _} -> true _ -> false end) |> Stream.map(fn {:ok, item} -> item end) with a filter dropping everything but {:ok, item}s. When I need to react to failures …

Elixir Programming Language Forum

Understanding GenStage/Flow Error Propagation

Questions & Help Questions

dimitarvp June 10, 2023, 5:25pm 5

My bad, I misread the function name. Apologies for the noise.