This is probably the most clarifying answer. The key here is that handle_cast and other GenStage callbacks allow you to fulfill pending demand. That is/was not clear to most people, including me, even though it is in the docs.

You dispatch the events that you have available (and that do not exceed the demand) with the {:noreply, [event], state} tuple from handle_demand. Those events are processed immediately. But you still have to deliver the “pending demand” and as long as that pending demand has not been fulfilled, you cannot expect any further handle_demand calls.

But lets say there is a message that lets you know that more data for events has become available - you can process that message with handle_cast (for example). handle_cast (and other GenStage callbacks) also lets you return events with {:noreply, [event], state} to fulfill that pending demand.

Now if it you happen to process a handle_demand while you still have pending demand, you have to accumulated it - i.e.

new_pending_demand = demand + pending_demand

So I have been struggling with this same issue. My producer pulls events off of one or more kafka topics and sends the on to a consumer. This works great as long as there is more supply than demand. I’m using the kafka fetch api to pull events off the queue. If there are no events or not as many as demanded by the consumer the processing stops due to unmet demand. I was trying to find a good example of how to start a polling task to periodically consume from the kafka topic. Not much out there, but with a little help from the Slack channel I came up with the following example. It uses an erlang queue to mimic the kafka topic and Process.send_after to periodically pull from the queue.

defmodule QueueTest do
  # use GenStage

  def start(max_demand \\ 10, poll_interval \\ 10000) do
    string = "I am the very model of a modern Major-General, I've information vegetable, animal, and mineral, I know the kings of England, and I quote the fights historical From Marathon to Waterloo, in order categorical"
    {:ok, message_queue} = GenServer.start_link(QueueTest.MessageQueue, string)
    {:ok, producer} = GenStage.start_link(QueueTest.Producer, {message_queue, poll_interval})
    {:ok, consumer} = GenStage.start_link(QueueTest.Consumer, :ok)
    GenStage.sync_subscribe(consumer, [to: producer, max_demand: max_demand])
    message_queue
  end

  def push(message_queue, string), do: GenServer.cast(message_queue, {:push, string})
end

defmodule QueueTest.MessageQueue do
  use GenServer

  def init(string) do
    {:ok, :queue.from_list(String.split(string, " "))}
  end

  def handle_cast({:push, string}, queue) do
    queue = String.split(string) |> Enum.reduce(queue, fn(word, queue) -> :queue.in(word, queue) end)
    {:noreply, queue}
  end

  def handle_call({:fetch, items}, _from, queue) do
    dequeue(queue, items, [])
  end

  defp dequeue(queue, 0, words) do
    {:reply, Enum.reverse(words), queue}
  end

  defp dequeue(queue, demand, words) do
    case :queue.out(queue) do
      {{:value, word}, queue} ->
        dequeue(queue, demand - 1, [word | words])
      {:empty, queue} ->
        {:reply, Enum.reverse(words), queue}
    end
  end
end

defmodule QueueTest.Producer do
  use GenStage

  def init({queue, poll_interval}) do
    {:producer, {queue, 0, poll_interval}}
  end

  def handle_demand(demand, {queue, pending_demand, poll_interval}) do
    do_handle_demand(queue, demand+pending_demand, poll_interval)
  end

  def handle_info(:fetch_more, {queue, pending_demand, poll_interval}) do
    do_handle_demand(queue, pending_demand, poll_interval)
  end

  defp do_handle_demand(queue, demand, poll_interval) do
    words = GenServer.call(queue, {:fetch, demand})
    if Enum.count(words) < demand do
      poll_on_insufficient_supply(poll_interval)
    end
    {:noreply, words, {queue, demand - Enum.count(words), poll_interval}}
  end

  defp poll_on_insufficient_supply(poll_interval) do
    Process.send_after(self(), :fetch_more, poll_interval)
  end
end

defmodule QueueTest.Consumer do
  use GenStage

  def init(:ok) do
    {:consumer, :ok}
  end

  def handle_events(words, _from, state) do
    words |> Enum.join(" ") |> IO.write
    IO.puts ""
    {:noreply, [], state}
  end

  def handle_subscribe(:producer, _opts, from, _state) do
    IO.puts("subscribe from #{inspect from}")
    {:automatic, from}
  end
end

I believe there is a confusion here. If the consumer asks for 100 items, you don’t need to wait for 100 items in order to reply to the consumer. Instead, you should reply to it with whatever you have available. If you do not have enough items, then you should store how much you own the consumer and send more items as soon as you can.

It is likely that you are doing what you described above, I just wanted to clarify.

Just to clarify as well: there is no such thing as announcing demand can be fullfilled. Consumers will always ask for more items if they are ready to process more items. Then the producer should send what it can, when it can.

Oops, premature send.

In any case, the fixes you propose at the end seem to be correct. If you cannot fullfil the demand, you should either have the repository tell you when it has more items available OR you should have a timer that forces you to ask the repository again in 1 minute or so. Check Process.send_after/3 for the latter.