Gen Stage

Skill GenStage - Guia Completo para Desenvolvedores Elixir

O que e GenStage?

GenStage e uma especificacao Elixir para troca de eventos entre producers (produtores) e consumers (consumidores) com back-pressure (contrapressao) automatica. Ele resolve o problema fundamental de: "Como processar dados de forma concorrente sem sobrecarregar o sistema?"

A ideia central: o consumidor pede dados ao produtor (modelo pull), nao o produtor empurrando dados para o consumidor (modelo push). Isso garante que nenhum componente do pipeline fique sobrecarregado.

Instalacao

# mix.exs
defp deps do
  [
    {:gen_stage, "~> 1.0"}
  ]
end

Requisito: Elixir >= 1.11

Conceitos Fundamentais

Os 3 Tipos de Stage

+------------+       +---------------------+       +------------+
|  PRODUCER  | ----> | PRODUCER_CONSUMER   | ----> |  CONSUMER  |
| (Source)   |       | (Transformador)     |       | (Sink)     |
+------------+       +---------------------+       +------------+
     ^                        ^                          |
     |                        |                          |
  Gera eventos         Recebe E emite            Consome eventos
  sob demanda          eventos                   e pede mais

1. :producer (source) - Apenas gera eventos. Implementa handle_demand/2.
2. :consumer (sink) - Apenas consome eventos. Implementa handle_events/3.
3. :producer_consumer - Recebe eventos, transforma e emite novos. Implementa handle_events/3.

Como funciona o Back-Pressure

Consumer: "Quero 1000 eventos" (max_demand: 1000)
    |
    v
Producer: Envia ate 1000 eventos
    |
    v
Consumer: Processa eventos...
    Quando demanda cai para 750 (min_demand), pede mais 250
    |
    v
Producer: Envia mais 250
    ... ciclo continua ...

• max_demand - Maximo de eventos solicitados por vez (padrao: 1000)
• min_demand - Limiar para pedir mais eventos (padrao: max_demand * 3/4, ou seja, 750)
• O consumidor NUNCA recebe mais do que pediu
• Quando o producer produz em lotes de 100 e max_demand=1000/min_demand=750: apos 3 lotes a demanda cai para 700 (abaixo de 750), entao o consumer pede 300 para voltar a 1000

Padrao 1: Pipeline Basico (Producer -> Consumer)

O caso mais simples: um produtor gera numeros e um consumidor os processa.

defmodule NumberProducer do
  use GenStage

  def start_link(initial) do
    GenStage.start_link(__MODULE__, initial, name: __MODULE__)
  end

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

  @impl true
  def handle_demand(demand, counter) when demand > 0 do
    # Se counter=3 e demand=2, emite [3, 4] e estado vira 5
    events = Enum.to_list(counter..(counter + demand - 1))
    {:noreply, events, counter + demand}
  end
end

defmodule NumberConsumer do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok)
  end

  @impl true
  def init(:ok) do
    # subscribe_to conecta automaticamente ao producer
    {:consumer, :ok, subscribe_to: [{NumberProducer, max_demand: 10}]}
  end

  @impl true
  def handle_events(events, _from, state) do
    Enum.each(events, fn event ->
      IO.puts("Processando: #{event}")
    end)

    # Consumer SEMPRE retorna lista vazia de eventos
    {:noreply, [], state}
  end
end

Supervision Tree

children = [
  {NumberProducer, 0},
  Supervisor.child_spec({NumberConsumer, []}, id: :c1),
  Supervisor.child_spec({NumberConsumer, []}, id: :c2),
  Supervisor.child_spec({NumberConsumer, []}, id: :c3),
  Supervisor.child_spec({NumberConsumer, []}, id: :c4)
]

# rest_for_one: se o producer reiniciar, consumers tambem reiniciam
Supervisor.start_link(children, strategy: :rest_for_one)

IMPORTANTE: Use strategy: :rest_for_one para que se o producer reiniciar, os consumers tambem reiniciem (eles precisam se reinscrever). Com :one_for_one, se o producer A morre, os consumers morrem tambem (perdem a inscricao) e o supervisor pode interpretar como muitas falhas simultaneas.

Padrao 2: Pipeline com Transformacao (Producer -> ProducerConsumer -> Consumer)

Quando voce precisa transformar dados entre a fonte e o destino.

defmodule DataSource do
  use GenStage

  def start_link(opts) do
    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(_opts) do
    {:producer, :no_state}
  end

  @impl true
  def handle_demand(demand, state) do
    events = fetch_from_database(demand)
    {:noreply, events, state}
  end

  defp fetch_from_database(count) do
    Enum.map(1..count, fn i -> %{id: i, raw_data: "data_#{i}"} end)
  end
end

defmodule DataTransformer do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  @impl true
  def init(:ok) do
    # producer_consumer tem buffer_size: :infinity por padrao!
    {:producer_consumer, :ok, subscribe_to: [{DataSource, max_demand: 100}]}
  end

  @impl true
  def handle_events(events, _from, state) do
    transformed =
      events
      |> Enum.map(&transform/1)
      |> Enum.filter(&valid?/1)

    # producer_consumer PODE filtrar: emitir menos eventos do que recebeu
    {:noreply, transformed, state}
  end

  defp transform(%{raw_data: data} = event) do
    Map.put(event, :processed_data, String.upcase(data))
  end

  defp valid?(%{processed_data: data}), do: String.length(data) > 0
end

defmodule DataSink do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok)
  end

  @impl true
  def init(:ok) do
    {:consumer, :ok, subscribe_to: [{DataTransformer, max_demand: 50}]}
  end

  @impl true
  def handle_events(events, _from, state) do
    Enum.each(events, &persist/1)
    {:noreply, [], state}
  end

  defp persist(event) do
    IO.inspect(event, label: "Persistindo")
  end
end

Anti-pattern: NAO crie um stage por passo logico do seu dominio. Stages servem para modelar propriedades de runtime (concorrencia, transferencia de dados), NAO organizacao de codigo. Se voce tem 3 passos logicos, considere combina-los em um unico consumer e escalar com multiplas instancias:

                 [Consumer: Step1 + Step2 + Step3]
                /
[Producer] ---->[Consumer: Step1 + Step2 + Step3]
                \
                 [Consumer: Step1 + Step2 + Step3]

Padrao 3: BroadcastDispatcher (1 Producer -> N Consumers, todos recebem tudo)

Quando TODOS os consumers precisam receber TODOS os eventos (fan-out).

defmodule EventBroadcaster do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  @doc "Envia evento e so retorna apos o despacho"
  def sync_notify(event, timeout \\ 5000) do
    GenStage.call(__MODULE__, {:notify, event}, timeout)
  end

  @impl true
  def init(:ok) do
    # BroadcastDispatcher envia para TODOS os consumers
    # Acumula demanda ate TODOS terem pedido
    {:producer, {:queue.new(), 0}, dispatcher: GenStage.BroadcastDispatcher}
  end

  @impl true
  def handle_call({:notify, event}, from, {queue, pending_demand}) do
    queue = :queue.in({from, event}, queue)
    dispatch_events(queue, pending_demand, [])
  end

  @impl true
  def handle_demand(incoming_demand, {queue, pending_demand}) do
    dispatch_events(queue, incoming_demand + pending_demand, [])
  end

  defp dispatch_events(queue, 0, events) do
    {:noreply, Enum.reverse(events), {queue, 0}}
  end

  defp dispatch_events(queue, demand, events) do
    case :queue.out(queue) do
      {{:value, {from, event}}, queue} ->
        GenStage.reply(from, :ok)
        dispatch_events(queue, demand - 1, [event | events])

      {:empty, queue} ->
        {:noreply, Enum.reverse(events), {queue, demand}}
    end
  end
end

Consumer com filtro (selector)

# Consumer que recebe APENAS eventos do tipo :email
defmodule EmailNotifier do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok)
  end

  @impl true
  def init(:ok) do
    selector = fn event -> event.type == :email end

    {:consumer, :ok,
     subscribe_to: [
       {EventBroadcaster, selector: selector}
     ]}
  end

  @impl true
  def handle_events(events, _from, state) do
    Enum.each(events, fn event ->
      IO.puts("Enviando email: #{inspect(event)}")
    end)
    {:noreply, [], state}
  end
end

# Consumer que recebe TODOS os eventos (auditoria)
defmodule AuditLogger do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok)
  end

  @impl true
  def init(:ok) do
    {:consumer, :ok, subscribe_to: [EventBroadcaster]}
  end

  @impl true
  def handle_events(events, _from, state) do
    Enum.each(events, fn event ->
      IO.puts("[AUDIT] #{inspect(event)}")
    end)
    {:noreply, [], state}
  end
end

Cuidado com o BroadcastDispatcher durante setup: O primeiro lote de eventos pode ser entregue antes de todos os consumers se inscreverem. Use demand: :accumulate no init do producer e chame GenStage.demand(producer, :forward) depois que todos os consumers estiverem conectados.

def init(:ok) do
  {:producer, state, dispatcher: GenStage.BroadcastDispatcher, demand: :accumulate}
end

# Depois de todos os consumers se inscreverem:
GenStage.demand(EventBroadcaster, :forward)

Padrao 4: PartitionDispatcher (Roteamento por Chave)

Quando voce precisa rotear eventos para consumers especificos baseado em uma chave. Apenas UM consumer por particao.

defmodule OrderProducer do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  @impl true
  def init(:ok) do
    {:producer, {:queue.new(), 0},
     dispatcher: {
       GenStage.PartitionDispatcher,
       partitions: [:standard, :express, :international],
       hash: fn event ->
         # Retorna {evento, nome_da_particao}
         {event, event.shipping_type}
       end
     }}
  end

  # API publica
  def new_order(order) do
    GenStage.cast(__MODULE__, {:order, order})
  end

  @impl true
  def handle_cast({:order, order}, {queue, demand}) do
    queue = :queue.in(order, queue)
    dispatch(queue, demand, [])
  end

  @impl true
  def handle_demand(incoming, {queue, demand}) do
    dispatch(queue, incoming + demand, [])
  end

  defp dispatch(queue, 0, events),
    do: {:noreply, Enum.reverse(events), {queue, 0}}

  defp dispatch(queue, demand, events) do
    case :queue.out(queue) do
      {{:value, order}, queue} ->
        dispatch(queue, demand - 1, [order | events])
      {:empty, queue} ->
        {:noreply, Enum.reverse(events), {queue, demand}}
    end
  end
end

# Um consumer POR particao (obrigatorio)
defmodule OrderProcessor do
  use GenStage

  def start_link(partition) do
    GenStage.start_link(__MODULE__, partition)
  end

  @impl true
  def init(partition) do
    {:consumer, partition,
     subscribe_to: [
       {OrderProducer, partition: partition, max_demand: 20}
     ]}
  end

  @impl true
  def handle_events(orders, _from, partition) do
    Enum.each(orders, fn order ->
      IO.puts("[#{partition}] Processando pedido ##{order.id}")
    end)
    {:noreply, [], partition}
  end
end

Supervision:

children = [
  {OrderProducer, []},
  Supervisor.child_spec({OrderProcessor, :standard}, id: :proc_standard),
  Supervisor.child_spec({OrderProcessor, :express}, id: :proc_express),
  Supervisor.child_spec({OrderProcessor, :international}, id: :proc_international)
]

Supervisor.start_link(children, strategy: :rest_for_one)

Particoes com inteiros e hash automatico

# Particoes 0, 1, 2, 3 com hash automatico via :erlang.phash2
{:producer, state,
 dispatcher: {GenStage.PartitionDispatcher, partitions: 0..3}}

# Inscricao:
{:consumer, state, subscribe_to: [{MyProducer, partition: 0}]}

Atencao: O PartitionDispatcher assume distribuicao uniforme entre particoes. Se os dados sao muito desiguais por longos periodos, particoes ocupadas vao acumular no buffer enquanto particoes ociosas ficam paradas. Isso e aceitavel para picos temporarios, mas problematico para desbalanceamento constante.

A funcao hash pode descartar eventos

hash: fn event ->
  case event.priority do
    :low -> :none  # Evento descartado, nao vai para nenhuma particao
    _ -> {event, event.region}
  end
end

Padrao 5: Rate Limiter (Controle Manual de Demanda)

Quando voce precisa controlar a taxa de processamento (ex: API com rate limit).

A chave e retornar {:manual, state} de handle_subscribe/4 e usar GenStage.ask/3 para controlar quando pedir mais eventos.

defmodule RateLimitedConsumer do
  use GenStage

  def start_link(opts) do
    GenStage.start_link(__MODULE__, opts)
  end

  @impl true
  def init(_opts) do
    # Estado: mapa de producers com seus pending/interval
    {:consumer, %{}}
  end

  @impl true
  def handle_subscribe(:producer, opts, from, producers) do
    # Configura quantos eventos por intervalo
    pending = opts[:max_demand] || 1000
    interval = opts[:interval] || 5000

    producers = Map.put(producers, from, {pending, interval})
    producers = ask_and_schedule(producers, from)

    # :manual = EU controlo quando pedir mais eventos
    {:manual, producers}
  end

  @impl true
  def handle_cancel(_, from, producers) do
    {:noreply, [], Map.delete(producers, from)}
  end

  @impl true
  def handle_events(events, from, producers) do
    # Acumula pending para proxima janela de tempo
    producers =
      Map.update!(producers, from, fn {pending, interval} ->
        {pending + length(events), interval}
      end)

    # Processa os eventos
    Enum.each(events, &IO.inspect/1)

    {:noreply, [], producers}
  end

  @impl true
  def handle_info({:ask, from}, producers) do
    {:noreply, [], ask_and_schedule(producers, from)}
  end

  defp ask_and_schedule(producers, from) do
    case producers do
      %{^from => {pending, interval}} ->
        GenStage.ask(from, pending)
        Process.send_after(self(), {:ask, from}, interval)
        Map.put(producers, from, {0, interval})

      _ ->
        producers
    end
  end
end

Uso

{:ok, producer} = GenStage.start_link(MyProducer, 0)
{:ok, consumer} = GenStage.start_link(RateLimitedConsumer, :ok)

# 10 eventos a cada 2 segundos
GenStage.sync_subscribe(consumer, to: producer, max_demand: 10, interval: 2000)

Padrao 6: ConsumerSupervisor (1 Processo por Evento)

Quando cada evento precisa ser processado por um processo filho separado. O max_demand age como limite de concorrencia (como um pool).

defmodule JobProducer do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def enqueue(job) do
    GenStage.cast(__MODULE__, {:enqueue, job})
  end

  @impl true
  def init(:ok) do
    {:producer, {:queue.new(), 0}}
  end

  @impl true
  def handle_cast({:enqueue, job}, {queue, demand}) do
    queue = :queue.in(job, queue)
    dispatch_jobs(queue, demand, [])
  end

  @impl true
  def handle_demand(incoming_demand, {queue, demand}) do
    dispatch_jobs(queue, incoming_demand + demand, [])
  end

  defp dispatch_jobs(queue, 0, events) do
    {:noreply, Enum.reverse(events), {queue, 0}}
  end

  defp dispatch_jobs(queue, demand, events) do
    case :queue.out(queue) do
      {{:value, job}, queue} ->
        dispatch_jobs(queue, demand - 1, [job | events])
      {:empty, queue} ->
        {:noreply, Enum.reverse(events), {queue, demand}}
    end
  end
end

defmodule JobWorkerSupervisor do
  use ConsumerSupervisor

  def start_link(_opts) do
    ConsumerSupervisor.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  @impl true
  def init(:ok) do
    # Filhos DEVEM ter restart: :temporary ou :transient
    children = [
      %{
        id: JobWorker,
        start: {JobWorker, :start_link, []},
        restart: :temporary
      }
    ]

    opts = [
      strategy: :one_for_one,
      subscribe_to: [{JobProducer, max_demand: 50}]
    ]

    ConsumerSupervisor.init(children, opts)
  end
end

defmodule JobWorker do
  def start_link(job) do
    Task.start_link(fn ->
      IO.puts("Processando job: #{inspect(job)}")
      Process.sleep(Enum.random(100..1000))
      IO.puts("Job concluido: #{inspect(job)}")
    end)
  end
end

Nota: O max_demand no ConsumerSupervisor funciona como limite de concorrencia. Com max_demand: 50, no maximo 50 workers rodam simultaneamente. A media de filhos simultaneos e (max_demand + min_demand) / 2.

Padrao 7: Producer com Buffer Automatico (Eventos Asincronos)

Quando eventos chegam de forma assincrona (webhooks, filas externas, etc.) e nao sob demanda.

defmodule AsyncProducer do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  # API publica - eventos chegam a qualquer momento
  def push(events) when is_list(events) do
    GenStage.cast(__MODULE__, {:push, events})
  end

  @impl true
  def init(:ok) do
    # buffer_size: quantos eventos armazenar quando nao ha demanda
    # buffer_keep: :last descarta os mais antigos quando excede, :first descarta os novos
    {:producer, :ok, buffer_size: 50_000, buffer_keep: :last}
  end

  @impl true
  def handle_demand(_demand, state) do
    # Nao temos eventos sob demanda - eles chegam via push/1
    {:noreply, [], state}
  end

  @impl true
  def handle_cast({:push, events}, state) do
    # Eventos sao automaticamente bufferizados se nao houver demanda
    # Se o buffer exceder buffer_size, um log de erro e emitido
    {:noreply, events, state}
  end
end

Sobre o Buffer Automatico

• :producer: buffer_size padrao = 10_000
• :producer_consumer: buffer_size padrao = :infinity (cuidado com memoria!)
• Quando o buffer excede o tamanho, eventos sao descartados e um log e emitido
• Customize o log implementando format_discarded/2:

@impl true
def format_discarded(discarded_count, current_buffer_size) do
  "Descartados #{discarded_count} eventos. Buffer atual: #{current_buffer_size}"
end

Padrao 8: Inscricao Dinamica (subscribe em runtime)

Quando voce nao sabe em tempo de compilacao quais producers existem.

defmodule DynamicConsumer do
  use GenStage

  def start_link(opts) do
    GenStage.start_link(__MODULE__, opts)
  end

  def subscribe_to_producer(consumer, producer, opts \\ []) do
    # Retorna {:ok, subscription_tag} ou {:error, reason}
    GenStage.sync_subscribe(consumer, [{:to, producer} | opts])
  end

  def unsubscribe(consumer, subscription_tag, reason \\ :normal) do
    GenStage.cancel({subscription_tag, consumer}, reason)
  end

  @impl true
  def init(_opts) do
    # Inicia SEM inscricoes
    {:consumer, %{}}
  end

  @impl true
  def handle_subscribe(:producer, _opts, from, state) do
    IO.puts("Inscrito em: #{inspect(from)}")
    {:automatic, state}
  end

  @impl true
  def handle_cancel({:cancel, _}, from, state) do
    IO.puts("Desinscrito de: #{inspect(from)}")
    {:noreply, [], state}
  end

  @impl true
  def handle_events(events, {producer_pid, _ref}, state) do
    IO.puts("#{length(events)} eventos de #{inspect(producer_pid)}")
    {:noreply, [], state}
  end
end

sync_subscribe vs async_subscribe

# sync_subscribe: bloqueia ate o producer confirmar a inscricao
# Retorna {:ok, tag} ou {:error, reason}
{:ok, tag} = GenStage.sync_subscribe(consumer, to: producer, max_demand: 100)

# async_subscribe: nao bloqueia, inscricao acontece eventualmente
# Retorna {:ok, tag} imediatamente
{:ok, tag} = GenStage.async_subscribe(consumer, to: producer)

Tipos de Cancelamento

GenStage.sync_subscribe(consumer,
  to: producer,
  cancel: :permanent   # (padrao) consumer para se producer morrer
  # cancel: :transient # consumer para so se producer morrer com erro
  # cancel: :temporary # consumer NUNCA para por causa do producer
)

Padrao 9: Usando GenStage com Streams do Elixir

Para integrar GenStage com o ecossistema de Streams (util para testes e scripts).

# Criar um producer a partir de um Enumerable
{:ok, producer} = GenStage.from_enumerable(1..1_000_000)

# Criar um stream a partir de producers (consome como consumer)
GenStage.stream([{producer, max_demand: 100}])
|> Stream.map(&(&1 * 2))
|> Stream.take(10)
|> Enum.to_list()
# => [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]

Util para testes

test "valida transformacao do pipeline" do
  {:ok, producer} = GenStage.from_enumerable([
    %{id: 1, total: 100},
    %{id: 2, total: -10},    # invalido
    %{id: 3, total: 200}
  ])

  {:ok, transformer} = GenStage.start_link(MyTransformer, :ok)
  GenStage.sync_subscribe(transformer, to: producer, max_demand: 10)

  results =
    GenStage.stream([{transformer, max_demand: 10}])
    |> Enum.to_list()

  assert length(results) == 2
end

Padrao 10: Controle de Demanda (accumulate/forward)

Quando voce precisa acumular demanda antes de comecar a processar (ex: esperar que todos os consumers se inscrevam, ou esperar conexao com banco).

defmodule DatabaseProducer do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  # Chamado quando a conexao com o banco estiver pronta
  def start_producing do
    GenStage.demand(__MODULE__, :forward)
  end

  # Consultar o modo atual
  def demand_mode do
    GenStage.demand(__MODULE__)
  end

  @impl true
  def init(:ok) do
    # :accumulate = acumula demanda sem emitir eventos
    {:producer, :not_connected, demand: :accumulate}
  end

  @impl true
  def handle_demand(demand, state) do
    # So e chamado quando o modo e :forward
    events = fetch_from_db(demand)
    {:noreply, events, state}
  end

  defp fetch_from_db(count) do
    Enum.to_list(1..count)
  end
end

Referencia Completa de Callbacks

Obrigatorios

Opcionais

Retornos Possiveis

# De handle_demand, handle_events, handle_cast, handle_info:
{:noreply, events, new_state}
{:noreply, events, new_state, :hibernate}
{:stop, reason, new_state}

# De handle_call (adiciona opcao de reply):
{:reply, reply, events, new_state}
{:reply, reply, events, new_state, :hibernate}
{:noreply, events, new_state}
{:stop, reason, reply, new_state}
{:stop, reason, new_state}

# De handle_subscribe:
{:automatic, new_state}   # demanda automatica (padrao)
{:manual, new_state}      # voce controla via GenStage.ask/3

# De handle_cancel:
{:noreply, events, new_state}

Nota: Em TODOS os callbacks, o campo events DEVE ser [] para consumers. Producers e producer_consumers podem emitir eventos.

Referencia Completa de Dispatchers

DemandDispatcher (padrao)

Envia eventos para o consumer com maior demanda pendente. Ordenacao FIFO.

{:producer, state}  # usa DemandDispatcher por padrao

# Ou explicitamente com opcoes:
{:producer, state,
 dispatcher: {GenStage.DemandDispatcher,
   shuffle_demands_on_first_dispatch: true,  # evita sobrecarregar o 1o consumer
   max_demand: 1000                           # emite warning se consumer pedir mais
 }}

Regra: Todos os consumers de um mesmo producer com DemandDispatcher devem ter o MESMO max_demand. Caso contrario, consumers "gulosos" (com max_demand maior) recebem mais eventos.

BroadcastDispatcher

Envia TODOS os eventos para TODOS os consumers. Aguarda a demanda minima de todos antes de enviar.

{:producer, state, dispatcher: GenStage.BroadcastDispatcher}

# Consumer com filtro:
{:consumer, state, subscribe_to: [
  {producer, selector: fn event -> event.important? end}
]}

Regra: Um mesmo processo NAO pode se inscrever duas vezes. O BroadcastDispatcher rejeita com :already_subscribed.

PartitionDispatcher

Roteia eventos para particoes fixas. Exatamente 1 consumer por particao.

{:producer, state,
 dispatcher: {GenStage.PartitionDispatcher,
   partitions: [:odd, :even],  # ou inteiro: partitions: 4  (gera 0..3)
   hash: fn event ->
     case rem(event, 2) do
       0 -> {event, :even}
       1 -> {event, :odd}
     end
   end
 }}

# Consumer DEVE especificar particao:
{:consumer, state, subscribe_to: [{producer, partition: :even}]}

Regras:

• Se :partitions usa nomes nao-inteiros, voce DEVE fornecer :hash
• A funcao hash pode retornar :none para descartar o evento
• Apenas 1 consumer por particao (erro se tentar 2)

Funcoes da API Publica

Ciclo de Vida

GenStage.start_link(module, args, opts)     # Inicia com link ao processo pai
GenStage.start(module, args, opts)          # Inicia sem link
GenStage.stop(stage, reason \\ :normal, timeout \\ :infinity)

Inscricao

GenStage.sync_subscribe(stage, opts)                        # Bloqueante
GenStage.async_subscribe(stage, opts)                       # Nao-bloqueante
GenStage.sync_resubscribe(stage, tag, reason, to, opts)     # Cancela e reinscreve
GenStage.async_resubscribe(stage, tag, reason, opts)        # Idem, async
GenStage.cancel({tag, stage}, reason)                       # Cancela inscricao

Controle de Demanda

GenStage.ask(from, demand)                  # Pede eventos (so modo :manual)
GenStage.demand(stage)                      # Consulta modo (:forward ou :accumulate)
GenStage.demand(stage, :forward)            # Ativa envio de eventos
GenStage.demand(stage, :accumulate)         # Pausa envio, acumula demanda

Comunicacao (compativel com GenServer)

GenStage.call(stage, request, timeout \\ 5000)
GenStage.cast(stage, request)
GenStage.reply(from, reply)
GenStage.sync_info(stage, msg, timeout \\ 5000)  # Enfileira info (respeita ordem com eventos)
GenStage.async_info(stage, msg)                   # Idem, async

Utilitarios

GenStage.from_enumerable(enumerable, opts \\ [])  # Cria producer de enumerable
GenStage.stream(subscriptions, opts \\ [])          # Cria stream de producers
GenStage.estimate_buffered_count(stage, timeout \\ 5000)  # Tamanho estimado do buffer

Boas Praticas e Armadilhas

FACA

1. Use GenStage para concorrencia e transferencia de dados, nao para organizar codigo ou modelar dominios
2. Combine passos logicos no mesmo stage quando possivel - cada stage adiciona overhead de processo e mensagens
3. Escale com multiplos consumers em vez de mais stages intermediarios
4. Use Task.async_stream/2 para paralelismo simples antes de recorrer ao GenStage
5. Configure max_demand igual em todos os consumers de um mesmo producer com DemandDispatcher
6. Use strategy: :rest_for_one no Supervisor
7. Use :subscribe_to no init ao inves de subscribe manual - garante re-inscricao apos crashes
8. Trate handle_cancel/3 para limpar estado quando um producer cai

NAO FACA

1. Nao crie pipelines longos com um stage por passo logico
2. Nao use max_demand: 1 exceto para debug - mata performance (sem batching)
3. Nao misture consumers com max_demand diferentes no DemandDispatcher
4. Nao esqueça que producer_consumer tem buffer_size: :infinity por padrao
5. Nao ignore o buffer - monitore com GenStage.estimate_buffered_count/2
6. Nao faca trabalho assincrono em handle_events/3 com modo :automatic - a demanda e enviada logo apos o retorno; se precisa de async, use modo :manual

Quando usar GenStage vs Alternativas

Protocolo de Mensagens (Avancado)

GenStage usa um protocolo de mensagens interno para comunicacao entre stages. Util se voce precisa implementar stages customizados ou debugar:

Consumer -> Producer

# Inscricao
{:"$gen_producer", {consumer_pid, subscription_tag}, {:subscribe, current, options}}

# Cancelamento
{:"$gen_producer", {consumer_pid, subscription_tag}, {:cancel, reason}}

# Pedido de demanda
{:"$gen_producer", {consumer_pid, subscription_tag}, {:ask, demand}}

Producer -> Consumer

# Cancelamento (confirmacao ou iniciado pelo producer)
{:"$gen_consumer", {producer_pid, subscription_tag}, {:cancel, reason}}

# Envio de eventos (lista nao-vazia)
{:"$gen_consumer", {producer_pid, subscription_tag}, events}

Ao receber inscricao, o producer DEVE monitorar o consumer. O consumer DEVE monitorar o producer ANTES de enviar a inscricao.

Exemplo Completo: Sistema de Processamento de Pedidos

Exemplo realista com producer assincrono, validacao e persistencia em batch:

# lib/my_app/pipeline/order_producer.ex
defmodule MyApp.Pipeline.OrderProducer do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def new_order(order) do
    GenStage.cast(__MODULE__, {:new_order, order})
  end

  @impl true
  def init(:ok) do
    {:producer, {:queue.new(), 0}, buffer_size: 10_000}
  end

  @impl true
  def handle_cast({:new_order, order}, {queue, demand}) do
    queue = :queue.in(order, queue)
    dispatch(queue, demand, [])
  end

  @impl true
  def handle_demand(incoming, {queue, demand}) do
    dispatch(queue, incoming + demand, [])
  end

  defp dispatch(queue, 0, events),
    do: {:noreply, Enum.reverse(events), {queue, 0}}

  defp dispatch(queue, demand, events) do
    case :queue.out(queue) do
      {{:value, order}, queue} ->
        dispatch(queue, demand - 1, [order | events])
      {:empty, queue} ->
        {:noreply, Enum.reverse(events), {queue, demand}}
    end
  end
end

# lib/my_app/pipeline/order_consumer.ex
defmodule MyApp.Pipeline.OrderConsumer do
  use GenStage

  def start_link(_opts) do
    GenStage.start_link(__MODULE__, :ok)
  end

  @impl true
  def init(:ok) do
    {:consumer, :ok,
     subscribe_to: [{MyApp.Pipeline.OrderProducer, max_demand: 50}]}
  end

  @impl true
  def handle_events(orders, _from, state) do
    orders
    |> Enum.map(&validate/1)
    |> Enum.filter(&match?({:ok, _}, &1))
    |> Enum.map(fn {:ok, order} -> order end)
    |> persist_batch()

    {:noreply, [], state}
  end

  defp validate(order) do
    cond do
      is_nil(order.customer_id) -> {:error, :missing_customer}
      order.total <= 0 -> {:error, :invalid_total}
      true -> {:ok, Map.put(order, :validated_at, DateTime.utc_now())}
    end
  end

  defp persist_batch([]), do: :ok
  defp persist_batch(orders) do
    entries = Enum.map(orders, fn order ->
      %{
        customer_id: order.customer_id,
        total: order.total,
        validated_at: order.validated_at,
        inserted_at: DateTime.utc_now()
      }
    end)

    MyApp.Repo.insert_all("orders", entries)
  end
end

# lib/my_app/pipeline/supervisor.ex
defmodule MyApp.Pipeline.Supervisor do
  use Supervisor

  def start_link(opts) do
    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(_opts) do
    children = [
      MyApp.Pipeline.OrderProducer,
      # 4 consumers processando em paralelo
      Supervisor.child_spec({MyApp.Pipeline.OrderConsumer, []}, id: :consumer_1),
      Supervisor.child_spec({MyApp.Pipeline.OrderConsumer, []}, id: :consumer_2),
      Supervisor.child_spec({MyApp.Pipeline.OrderConsumer, []}, id: :consumer_3),
      Supervisor.child_spec({MyApp.Pipeline.OrderConsumer, []}, id: :consumer_4)
    ]

    Supervisor.init(children, strategy: :rest_for_one)
  end
end

Uso

# Em qualquer parte da aplicacao:
MyApp.Pipeline.OrderProducer.new_order(%{
  customer_id: 42,
  total: 99.90,
  items: ["item_a", "item_b"]
})

Diagrama de Decisao: Qual Dispatcher Usar?

Todos os consumers precisam de todos os eventos?
  |
  |-- SIM --> BroadcastDispatcher
  |            |
  |            +-- Cada consumer quer apenas um subconjunto?
  |                  |-- SIM --> Use :selector na inscricao
  |                  |-- NAO --> Inscricao simples
  |
  |-- NAO --> Eventos devem ir para consumers especificos por chave?
               |
               |-- SIM --> PartitionDispatcher
               |            (defina :hash e :partitions)
               |
               |-- NAO --> DemandDispatcher (padrao)
                            (distribuicao por demanda, tipo round-robin)

Checklist para Implementacao

Ao criar um pipeline GenStage, verifique:

• Cada stage tem use GenStage ou use ConsumerSupervisor
• Producers implementam handle_demand/2
• Consumers/ProducerConsumers implementam handle_events/3
• Consumers retornam [] como lista de eventos
• max_demand esta configurado adequadamente (nao muito alto, nao muito baixo)
• Consumers do mesmo producer com DemandDispatcher tem o mesmo max_demand
• buffer_size esta configurado para producers que recebem eventos de forma assincrona
• buffer_size de producer_consumer esta explicitamente definido (padrao e :infinity)
• Supervision tree usa strategy: :rest_for_one
• Stages estao na ordem correta na supervision tree (producer primeiro)
• subscribe_to esta no init/1 (nao fazendo subscribe manual) para auto-reinscricao
• Se usando BroadcastDispatcher, considerou demand: :accumulate durante setup
• Se fazendo trabalho async, usando modo :manual com GenStage.ask/3
• Voce considerou se Task.async_stream/2 seria suficiente para seu caso