This may just be my misunderstanding on appropriate design, but is there a way for a producer to yield intermediate results where the count is much lower than what is demanded? If it is an expensive operation to produce the results or you want to limit the actual number of items in flight at any given time, what is the best way to do this?
I saw in the docs that the consumer can set max_demand and min_demand, but is there a way for the producer to specify this? Or for the producer to give back intermediate results while still fullfilling up to the requested demand?
Some example code below.
alias Experimental.GenStage
defmodule GenStageEx do
# ----------------------
# Producer -------------
# ----------------------
defmodule Producer do
use GenStage
def init(_ok) do
{:producer, []}
end
def handle_demand(demand, leftovers) do
IO.puts("# producer was asked for #{demand}")
items = case leftovers do
[] -> request_more()
_ -> leftovers
end
{to_return, next_leftovers} = Enum.split(items, demand)
{:noreply, to_return, next_leftovers}
end
def request_more() do
# some expensive operation that only returns 10 items at a time
0..9
end
end
# ----------------------
# Consumer -------------
# ----------------------
defmodule Consumer do
use GenStage
def init(_ok) do
{:consumer, %{}}
end
def handle_events(events, _from, state) do
events
|> Enum.map(&(IO.puts/1))
{:noreply, [], state}
end
end
def run_me do
{:ok, p} = GenStage.start_link(GenStageEx.Producer, :ok)
{:ok, c} = GenStage.start_link(GenStageEx.Consumer, :ok)
GenStage.sync_subscribe(c, to: p)
Process.sleep(2000)
end
end
