play.exs

defmodule A do
  use GenStage

  def init(counter) do
    {:producer, counter}
  end

  def handle_demand(demand, counter) when demand > 0 do
    IO.puts "DEMAND: #{demand}"

    # E.g. producer is querying some internet resource for new items, sometimes there are no new items
    events = case counter do
      0 -> [:a, :b, :c]
      1 -> []
      2 -> [:d, :e]
      _ -> [:oh]
    end
    {:noreply, events, counter + 1}
  end
end

defmodule C do
  use GenStage

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

  def handle_events(events, _from, state) do
    :timer.sleep(200)
    IO.inspect(events)

    # We are a consumer, so we would never emit items.
    {:noreply, [], state}
  end
end

{:ok, a} = GenStage.start_link(A, 0)
{:ok, c} = GenStage.start_link(C, :ok)

GenStage.sync_subscribe(c, to: a, max_demand: 1)
Process.sleep(:infinity)

# Output:
#
# ❯ mix run play.exs
# DEMAND: 1
# [:a]
# [:b]
# [:c]
# DEMAND: 1

# hangs there