Pretty Nice, with a BroadcastDispatcher, the selector of the Consumer is now working(and thats great!)
Now I`m kinda confused abot how the handle_demand of producer works.
I needed to send the list(or and update of it) to the consumer from times to time(not always, but not with Process.sleep(), is there a way to only send it through a method call?
Thanks for helping!
Well, as soon as a message is published by the producer, the consumer will digest it when it is idle. In case you need to add a delay to this consumption you would have to do it on the producer some way.
is “Not B” related to be, or does it mean anything besides “A” or “B”? Please explain your problem fully, if you add new concepts on each message we will not be able to help you properly 
Well, I though of sending it separated for performance improvement…
It may work, but spawning processes for each item will definitely not improve performances to just run “a quick validation” and a print on a list of 1000 items.
For now I would stick to the last proposal I made above, spawning a process for each item as required.
First, a producer that receive the whole list (as in the code I have posted before). You socket receives the whole list, and now you have that in memory. With any solution you will have to copy this data to other processes memory anyway, so you just send it to the producer and you can garbage collect on the socket. If you want to improve, you can stream events from the socket to the producer, but that is optimization, so for a future version, if ever needed.
So, you have this producer that receives the whole list. It uses the default dispatcher (not the broadcast or partition ones).
Then you add a ConsumerSupervisor that will receive the items according to its max_demand, and will spawn a process for each list item. In the handler function, you check the code and call the appropriate validator for this code.
defmodule SocketListUpdater 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.map(fn code -> %{code: code, infos: generate_infos()} end)
end
defp generate_infos do
[
%{name: "mary"},
%{name: "john"},
%{name: "robert"},
%{name: "ava"},
%{name: "simon"},
%{name: "simon"}
]
|> Enum.take(Enum.random(2..6))
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}
end
def handle_info({:new_list, list}, state) do
{:noreply, list, state}
end
def handle_demand(_demand, state) do
{:noreply, [], state}
end
end
defmodule GenstageExample.Consumer do
use ConsumerSupervisor
# Starts consumer with his code
def start_link(arg) do
ConsumerSupervisor.start_link(__MODULE__, arg)
end
def init(_arg) do
# Note: By default the restart for a child is set to :permanent
# which is not supported in ConsumerSupervisor. You need to explicitly
# set the :restart option either to :temporary or :transient.
children = [
%{
id: GenstageExample.Validator,
start: {GenstageExample.Validator, :start_link, []},
restart: :transient
}
]
opts = [strategy: :one_for_one, subscribe_to: [{GenstageExample.Producer, max_demand: 50}]]
ConsumerSupervisor.init(children, opts)
end
end
defmodule GenstageExample.Validator do
def start_link(event) do
Task.start_link(fn -> validate(event) end)
end
# Here you can call any validator function depending on the code. In this
# example I implemented the validators directly in the function body
defp validate(%{code: "A", infos: infos}) do
if length(infos) < 3 do
IO.puts("item with code A is not valid")
end
end
defp validate(%{code: "B", infos: infos}) do
if length(infos) == 2 do
IO.puts("item with code B is not valid")
end
end
defp validate(%{code: code}) do
IO.puts("received unknown code #{code}")
end
end
GenstageExample.Producer.start_link()
GenstageExample.Consumer.start_link([])
SocketListUpdater.start_link(GenstageExample.Producer)
So you do not have to start a validator for “B” or “Not B”, you just have a single validate/1 function that will check the code and do the appropriate validation. It is simpler.
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