GenStage Documentation

A ConsumerSupervisor is a specialized supervisor designed to be used as a consumer in a GenStage pipeline. Instead of managing a static set of children, it starts a new child process for every event that flows in from a producer.

Key Mechanics

• Event-to-Process Mapping: Each event received from a producer is appended to the arguments of the child specification, triggering the start of a new process.
• Demand Management: The supervisor manages demand automatically. It asks the producer for :max_demand events. As child processes terminate, the supervisor accumulates demand and requests more events once :min_demand is reached. This effectively creates a dynamic pool where the concurrency is bounded by these demand settings.
• Subscription: It can be attached to a producer by returning :subscribe_to from the init/1 callback or by using GenStage.sync_subscribe/3 or GenStage.async_subscribe/2.

Important Constraints

• Restart Policy: You cannot use :restart, :permanent for children. Because children are spawned per event, :permanent is not supported. You must explicitly set the :restart option to either :temporary (never restarted) or :transient (restarted only on abnormal exits).
• Process Linking: The child's start_link function must return {:ok, pid} and the process must be linked back to the supervisor (e.g., using Task.start_link/1).

defmodule Consumer do
  use ConsumerSupervisor

  def start_link(arg) do
    ConsumerSupervisor.start_link(__MODULE__, arg)
  end

  def init(_arg) do
    # Note: You must explicitly set :restart to :temporary or :transient
    children = [%{id: Printer, start: {Printer, :start_link, []}, restart: :transient}]
    opts = [strategy: :one_for_one, subscribe_to: [{Producer, max_demand: 50}]]
    ConsumerSupervisor.init(children, opts)
  end
end

defmodule Printer do
  def start_link(event) do
    # Must return {:ok, pid} and be linked to the supervisor
    Task.start_link(fn ->
      IO.inspect({self(), event})
    end)
  end
end

GenStage is a framework for building data-exchange pipelines using stages. A stage can play one of three roles:

1. Producer (or Source): Only produces and sends data.
2. Consumer (or Sink): Only receives and consumes data.
3. Producer-Consumer: Acts as both, receiving data from an upstream producer and sending transformed data to a downstream consumer.

Communication is driven by demand. Consumers send demand upstream to producers. A producer will never emit more events than the consumer has requested, providing a built-in back-pressure mechanism. This allows you to build pipelines that scale based on runtime needs like concurrency and data transfer rather than just code organization.

defmodule A do
  use GenStage

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

  def handle_demand(demand, counter) when demand > 0 do
    events = Enum.to_list(counter..counter+demand-1)
    {:noreply, events, counter + demand}
  end
end

To create a custom consumer supervisor, use the ConsumerSupervisor module. This provides the necessary boilerplate and implements the GenStage behaviour.

Implementation Steps

1. use ConsumerSupervisor in your module.
2. Implement start_link/1 to call ConsumerSupervisor.start_link/3.
3. Implement init/1. The init/1 callback must return {:ok, children, opts} or :ignore.

Child Specification Requirements

When defining children in init/1, ensure the :restart option is set to :temporary or :transient. Using :permanent will result in an error.

Example

defmodule MyConsumer do
  use ConsumerSupervisor

  def start_link(arg) do
    ConsumerSupervisor.start_link(__MODULE__, arg)
  end

  def init(_arg) do
    # Define the template for children to be spawned per event
    children = [%{id: Worker, start: {Worker, :start_link, []}, restart: :transient}]
    # Configure subscription and strategy
    opts = [strategy: :one_for_one, subscribe_to: [{Producer, max_demand: 20, min_demand: 10}]]
    ConsumerSupervisor.init(children, opts)
  end
end

defmodule Consumer do
        use ConsumerSupervisor

def start_link(arg) do
          ConsumerSupervisor.start_link(__MODULE__, arg)
        end

def init(_arg) do
          # Note: By default the restart for a child is set to :permanent
          # which is not supported in ConsumerSupervisor. You need to explicitly
          # set the :restart option either to :temporary or :transient.
          children = [%{id: Printer, start: {Printer, :start_link, []}, restart: :transient}]
          opts = [strategy: :one_for_one, subscribe_to: [{Producer, max_demand: 50}]]
          ConsumerSupervisor.init(children, opts)
        end
      end

Then in the `Printer` module:

defmodule Printer do
        def start_link(event) do
          # Note: this function must return the format of `{:ok, pid}` and like
          # all children started by a Supervisor, the process must be linked
          # back to the supervisor (if you use `Task.start_link/1` then both
          # requirements are met automatically)
          Task.start_link(fn ->
            IO.inspect({self(), event})
          end)
        end
end