Agent skill

gen-stage

Use when working with GenStage for building concurrent data pipelines with back-pressure in Elixir. Covers producers, consumers, dispatchers (Demand, Broadcast, Partition), rate limiting, ConsumerSupervisor, and supervision patterns.

View SKILL.md on GitHub Repository

Stars 163

Forks 31

Install this agent skill to your Project

npx add-skill https://github.com/majiayu000/claude-skill-registry/tree/main/skills/other/other/gen-stage

SKILL.md

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

elixir

# 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

:producer (source) - Apenas gera eventos. Implementa handle_demand/2.
:consumer (sink) - Apenas consome eventos. Implementa handle_events/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.

elixir

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

elixir

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.

elixir

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).

elixir

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)

elixir

# 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.

elixir

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.

elixir

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:

elixir

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

elixir

# 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

elixir

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.

elixir

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

elixir

{: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).

elixir

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.

elixir

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:

elixir

@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.

elixir

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

elixir

# 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

elixir

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).

elixir

# 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

elixir

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).

elixir

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

Callback	Tipo	Assinatura	Descricao
`init/1`	Todos	`init(args)`	Retorna `{:producer, state, opts}`, `{:consumer, state, opts}` ou `{:producer_consumer, state, opts}`. Tambem aceita `:ignore` ou `{:stop, reason}`
`handle_demand/2`	Producer	`handle_demand(demand, state)`	Recebe demanda (sempre > 0). Retorna `{:noreply, events, new_state}`
`handle_events/3`	Consumer/PC	`handle_events(events, from, state)`	Recebe lista de eventos. `from` e `{producer_pid, subscription_tag}`

Opcionais

Callback	Assinatura	Descricao
`handle_subscribe/4`	`handle_subscribe(:producer\|:consumer, opts, from, state)`	Controla modo da inscricao. Retorna `{:automatic, state}` ou `{:manual, state}`
`handle_cancel/3`	`handle_cancel(reason, from, state)`	Reage a cancelamento. `reason` e `{:cancel, _}`, `{:down, _}`, etc.
`handle_call/3`	`handle_call(request, from, state)`	Chamadas sincronas. Pode retornar `{:reply, reply, events, state}`
`handle_cast/2`	`handle_cast(request, state)`	Mensagens assincronas
`handle_info/2`	`handle_info(msg, state)`	Mensagens genericas
`terminate/2`	`terminate(reason, state)`	Limpeza ao parar
`format_status/2`	`format_status(reason, [pdict, state])`	Formata status do processo para debug
`format_discarded/2`	`format_discarded(discarded, buffered)`	Customiza mensagem de log quando buffer excede

Retornos Possiveis

elixir

# 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.

elixir

{: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.

elixir

{: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.

elixir

{: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

elixir

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

elixir

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

elixir

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)

elixir

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

elixir

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

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

NAO FACA

Nao crie pipelines longos com um stage por passo logico
Nao use max_demand: 1 exceto para debug - mata performance (sem batching)
Nao misture consumers com max_demand diferentes no DemandDispatcher
Nao esqueça que producer_consumer tem buffer_size: :infinity por padrao
Nao ignore o buffer - monitore com GenStage.estimate_buffered_count/2
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

Cenario	Solucao
Paralelismo simples sobre dados em memoria	`Task.async_stream/2`
Pipeline com back-pressure customizado	GenStage
Pipeline com janelas/reducoes/particionamento	Flow (construido sobre GenStage)
Ingestao de dados (Kafka, SQS, RabbitMQ)	Broadway (construido sobre GenStage)
Pub/sub simples	`Phoenix.PubSub` ou `Registry`

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

elixir

# 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

elixir

# 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:

elixir

# 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

elixir

# 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

Maintainer

majiayu000 Core maintainer

Source details

Full Name: majiayu000/claude-skill-registry
Branch: main
Path in repo: skills/other/other/gen-stage
License: MIT License

Featured Tools

Join Our Newsletter

Maintain .agent state files. Use at session start, after meaningful steps, and before concluding: read/update constitution/memory/focus/issues/baseline consistently.

163 31

Explore

Didn't find tool you were looking for?

Search AI Tools

Install this agent skill to your Project

SKILL.md

Skill GenStage - Guia Completo para Desenvolvedores Elixir

O que e GenStage?

Instalacao

Conceitos Fundamentais

Os 3 Tipos de Stage

Como funciona o Back-Pressure

Padrao 1: Pipeline Basico (Producer -> Consumer)

Supervision Tree

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

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

Consumer com filtro (selector)

Padrao 4: PartitionDispatcher (Roteamento por Chave)

Particoes com inteiros e hash automatico

A funcao hash pode descartar eventos

Padrao 5: Rate Limiter (Controle Manual de Demanda)

Uso

Padrao 6: ConsumerSupervisor (1 Processo por Evento)

Padrao 7: Producer com Buffer Automatico (Eventos Asincronos)

Sobre o Buffer Automatico

Padrao 8: Inscricao Dinamica (subscribe em runtime)

sync_subscribe vs async_subscribe

Tipos de Cancelamento

Padrao 9: Usando GenStage com Streams do Elixir

Util para testes

Padrao 10: Controle de Demanda (accumulate/forward)

Referencia Completa de Callbacks

Obrigatorios

Opcionais

Retornos Possiveis

Referencia Completa de Dispatchers

DemandDispatcher (padrao)

BroadcastDispatcher

PartitionDispatcher

Funcoes da API Publica

Ciclo de Vida

Inscricao

Controle de Demanda

Comunicacao (compativel com GenServer)

Utilitarios

Boas Praticas e Armadilhas

FACA

NAO FACA

Quando usar GenStage vs Alternativas

Protocolo de Mensagens (Avancado)

Consumer -> Producer

Producer -> Consumer

Exemplo Completo: Sistema de Processamento de Pedidos

Uso

Diagrama de Decisao: Qual Dispatcher Usar?

Checklist para Implementacao

Recommended Agent Skills

agent-ops-spec

agent-ops-state

agent-ops-spec

agent-ops-testing

agent-ops-testing

agent-ops-state