hey!
Normally it is not good to do any task on the supervisor.
Instead a valid approach would be using GenStage.
So, with this approach you could think about having a producer, the process that will receive the messages, and many consumers to process messages.
In the Producer you just will need to receive the messages and broadcast them. The filtering part can be done almost automatically using the selector option for the subscription of the consumers.
Hope it helps!

I see,

In that case GenStage can do the job with a consumer per code.

But my code is far from good:

• I did not use any buffering for the demands in the consumer, that should be handled properly
• Items are not handled in isolation, as you requested, the processes still receive a list of items.

That is why I would not use GenStage if I required process isolation for handling each item. Although, as it is a functional immutable language, I guess you are fine with handling lists of items.

defmodule ListUpdater do
  def start_link(producer \\ nil) do
    spawn_link(fn -> init(producer) end)
  end

  defp init(producer) do
    loop(producer)
  end

  defp loop(producer) do
    IO.puts("sending list")
    send(producer, {:new_list, generate_list()})
    # Send a list every minute, here every second
    Process.sleep(1000)
    loop(producer)
  end

  defp generate_list() do
    Stream.cycle(["A", "B", "C"])
    # The length of the list
    |> Enum.take(1000)
    |> Enum.shuffle()
    |> Enum.with_index()
    |> Enum.map(fn {key, index} -> %{key: key, id: index} end)
  end
end

defmodule GenstageExample.Producer do
  use GenStage

  def start_link() do
    GenStage.start_link(__MODULE__, [], name: __MODULE__)
  end

  def init(_arg) do
    {:producer, :some_state, dispatcher: GenStage.BroadcastDispatcher}
  end

  def handle_info({:new_list, list}, state) do
    {:noreply, list, state}
  end

  def handle_demand(_demand, state) do
    IO.puts("got demand")
    {:noreply, [], state}
  end
end

defmodule GenstageExample.Consumer do
  use GenStage

  # Starts consumer with his code
  def start_link(code) do
    GenStage.start_link(__MODULE__, code)
  end

  def init(code) do
    # Next line should filter messages sent from producer
    state = %{code: code}

    {:consumer, state,
     subscribe_to: [
       {GenstageExample.Producer, selector: fn %{key: key} -> code == key end, max_demand: 1000}
     ]}
  end

  def handle_events(events, _from, state) do
    # Prints every message received
    for event <- events do
      IO.puts(
        "[#{state.code}] #{inspect(self)} Received: #{inspect(event)}, Expected: #{state.code}"
      )
    end

    # As a consumer we never emit events
    {:noreply, [], state}
  end
end

GenstageExample.Producer.start_link()
GenstageExample.Consumer.start_link("A")
GenstageExample.Consumer.start_link("B")
GenstageExample.Consumer.start_link("C")
ListUpdater.start_link(GenstageExample.Producer)

Edit: You could also use the PartitionDispatcher, with the same code as above except for those differences:

  # GenstageExample.Producer
  def init(_arg) do
    {:producer, :some_state,
     dispatcher:
       {GenStage.PartitionDispatcher,
        partitions: ["A", "B", "C"], hash: fn event -> {event, event.key} end}}
  end


  # GenstageExample.Consumer
  def init(code) do
    # Next line should filter messages sent from producer
    state = %{code: code}

    {:consumer, state,
     subscribe_to: [
       {GenstageExample.Producer, partition: code, max_demand: 1000}
     ]}
  end

It makes more sense to me, but all the possible codes must be known beforehand. For example if you write partitions: ["A", "B"], then a list item with "C" will crash the producer. On the other hand, with the broadcast dispatcher, it is the opposite, as if you declare A,B,C, then a list item that would have D would be ignored (an maybe stay in the producer memory for ever ? I don’t know). So I would rather have the process to crash and use partitions that have unhandled events.

Hey lud, thanks for the code, I think that this will work with a few adjustments.

The only think is that I cant write different validate functions, because I dont know which code will be on the list(And even if I do, I wouldnt write 800 versions).

But with validate, I can get the code and the infos, and do what I need.

The final ideia is to do something like

defp validate({%{code: code, infos: infos}) do
    IO.puts("Checking code #{code}...")
    {age, name, nickname} = infos
    if age > 18 do
         IO.puts("#{nickname} is an adult ")
    end

end

Thats a silly demonstration, but the goal is to do specific commands according to properties inside “infos”.

Thanks for your time and effort