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