### Full Example: EventBroadcaster and EventListeners

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Demonstrates a complete GenStage application with an EventBroadcaster producer and multiple EventListener consumers using the BroadcastDispatcher. Includes setup, notification, and subscription with a selector.

```elixir
defmodule EventBroadcaster do
  use GenStage

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

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

  def notify(server, event) do
    GenStage.cast(server, {:notify, event})
  end

  def handle_cast({:notify, event}, state) do
    {:noreply, [event], state}
  end

  def handle_demand(_demand, state) do
    {:noreply, [], state}
  end
end

defmodule EventListener do
  use GenStage

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

  def init(name) do
    {:consumer, name, subscribe_to: [EventBroadcaster]}
  end

  def handle_events(events, _from, name) do
    IO.inspect({:listener, name, :received, events})
    {:noreply, [], name}
  end
end

# Usage: broadcast to multiple listeners
{:ok, bc} = EventBroadcaster.start_link(name: :bc)
{:ok, _} = EventListener.start_link(:listener1)
{:ok, _} = EventListener.start_link(:listener2)
{:ok, _} = EventListener.start_link(:listener3)

EventBroadcaster.notify(:bc, {:alert, :system_down})

# Or with selector
GenStage.sync_subscribe(high_priority_listener,
  to: :bc,
  selector: fn {:alert, severity} -> severity == :critical end
)
```

--------------------------------

### Start ConsumerSupervisor with Callback Module

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

This function starts a consumer supervisor where the child specifications and options are defined within a callback module's init/1 function. The module must implement the ConsumerSupervisor.init/1 callback.

```elixir
defmodule MyConsumerSupervisor do
  use ConsumerSupervisor

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

  def init(_arg) do
    children = [
      %{id: WorkerTask, start: {Worker, :start_link, []}, restart: :transient}
    ]
    
    opts = [
      strategy: :one_for_one,
      subscribe_to: [{Producer, max_demand: 50, min_demand: 25}]
    ]
    
    ConsumerSupervisor.init(children, opts)
  end
end

{:ok, sup} = MyConsumerSupervisor.start_link(:ok)
```

--------------------------------

### Starting Integer Partitioned Producer and Consumers

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Code to start the PartitionedProducer and four PartitionedConsumer instances, each subscribed to a different integer partition.

```elixir
# Start producer and 4 consumers (one per partition)
{:ok, _} = PartitionedProducer.start_link(name: :producer)
for p <- 0..3 do
  {:ok, _} = PartitionedConsumer.start_link(p)
end
```

--------------------------------

### Starting Named Partitioned Producer and Consumers

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Code to start the KeyPartitionedProducer and two KeyPartitionedConsumer instances, one for the :odd partition and one for the :even partition.

```elixir
# Start producer and consumers for odd/even partitions
{:ok, _} = KeyPartitionedProducer.start_link(name: :producer)
{:ok, _} = KeyPartitionedConsumer.start_link(:odd)
{:ok, _} = KeyPartitionedConsumer.start_link(:even)
```

--------------------------------

### GenStage Back-Pressure Examples: Calculating Average Concurrency

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/backpressure-guide.md

Examples demonstrating how to set `max_demand` and `min_demand` to control the average number of events in flight, influencing system concurrency and buffering.

```elixir
# Low concurrency (10 events in flight on average)
max_demand: 20, min_demand: 0    # avg = 10
```

```elixir
# Medium concurrency (50 events in flight)
max_demand: 100, min_demand: 0   # avg = 50
```

```elixir
# High concurrency (500 events in flight)
max_demand: 1000, min_demand: 0  # avg = 500
```

```elixir
# Tight buffering (95 events in flight on average)
max_demand: 100, min_demand: 90  # avg = 95
```

--------------------------------

### ConsumerSupervisor.start_link(module, args, options \ [])

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Starts a consumer supervisor using a callback module that defines the supervisor structure via its `init/1` callback.

```APIDOC
## ConsumerSupervisor.start_link(module, args, options \ [])

### Description
Starts a consumer supervisor where the supervisor structure is defined in a callback module. The module's `init/1` callback must return `{:ok, children, opts}`.

### Method
`ConsumerSupervisor.start_link/2` or `ConsumerSupervisor.start_link/3`

### Parameters
#### Path Parameters
- `module` (atom) - Required - Callback module implementing `ConsumerSupervisor.init/1`
- `args` (term) - Required - Arguments passed to `module.init(args)`
- `options` (keyword) - Optional - Startup options (:name, :debug, etc.) (default: [])

#### Query Parameters
- `:strategy` (atom) - Required - Must be `:one_for_one`
- `:max_restarts` (integer) - Optional - Maximum restarts in time frame (default: 3)
- `:max_seconds` (integer) - Optional - Time frame for restart limit in seconds
- `:subscribe_to` (list) - Optional - Producers to subscribe to: `[producer]` or `[{producer, opts}]` (default: [])
- `:name` (atom/tuple) - Optional - Supervisor registration name

### Response
#### Success Response
- `{:ok, pid}` - On successful startup.
- `{:error, reason}` - On failure.

### Request Example
```elixir
defmodule MyConsumerSupervisor do
  use ConsumerSupervisor

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

  def init(_arg) do
    children = [
      %{id: WorkerTask, start: {Worker, :start_link, []}, restart: :transient}
    ]
    
    opts = [
      strategy: :one_for_one,
      subscribe_to: [{Producer, max_demand: 50, min_demand: 25}]
    ]
    
    ConsumerSupervisor.init(children, opts)
  end
end

{:ok, sup} = MyConsumerSupervisor.start_link(:ok)
```
```

--------------------------------

### GenStage DemandDispatcher Example

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

A complete example demonstrating a producer and multiple consumers using GenStage.DemandDispatcher with load balancing enabled via `shuffle_demands_on_first_dispatch`. The producer handles demand by generating event batches, and consumers process these events.

```elixir
defmodule MyProducer do
  use GenStage

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

  def init(counter) do
    # Use demand dispatcher with shuffling to balance load
    {:producer, counter, 
      dispatcher: {GenStage.DemandDispatcher, shuffle_demands_on_first_dispatch: true}
    }
  end

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

defmodule MyConsumer do
  use GenStage

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

  def consumer(n), do: :"consumer_#{n}"

  def init(number) do
    {:consumer, number, subscribe_to: [{MyProducer, max_demand: 100}]}
  end

  def handle_events(events, _from, number) do
    IO.inspect({:consumer, number, :received, length(events), :events})
    {:noreply, [], number}
  end
end

# Usage
{:ok, _} = MyProducer.start_link(0)
{:ok, _} = MyConsumer.start_link(1)
{:ok, _} = MyConsumer.start_link(2)
{:ok, _} = MyConsumer.start_link(3)
```

--------------------------------

### Start ConsumerSupervisor with Child Specification

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Use this function to start a consumer supervisor with a predefined child template. Ensure the children list contains exactly one specification and the strategy is :one_for_one. The subscribe_to option is used to specify producers.

```elixir
children = [
  %{id: WorkerTask, start: {Worker, :start_link, []}, restart: :transient}
]

{:ok, sup} = ConsumerSupervisor.start_link(children, 
  strategy: :one_for_one,
  subscribe_to: [{Producer, max_demand: 50}]
)
```

--------------------------------

### ConsumerSupervisor.start_link(children, options)

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Starts a consumer supervisor with a simple child specification. The children list must contain exactly one child template specification.

```APIDOC
## ConsumerSupervisor.start_link(children, options)

### Description
Starts a consumer supervisor from a simple child template. The template must be a valid supervisor child specification with exactly one entry.

### Method
`ConsumerSupervisor.start_link/2`

### Parameters
#### Path Parameters
- `children` (list) - Required - List with exactly one child template specification
- `options` (keyword) - Required - Supervisor options

#### Query Parameters
- `:strategy` (atom) - Required - Must be `:one_for_one`
- `:max_restarts` (integer) - Optional - Maximum restarts in time frame (default: 3)
- `:max_seconds` (integer) - Optional - Time frame for restart limit in seconds
- `:subscribe_to` (list) - Optional - Producers to subscribe to: `[producer]` or `[{producer, opts}]` (default: [])
- `:name` (atom/tuple) - Optional - Supervisor registration name

### Response
#### Success Response
- `{:ok, pid}` - On successful startup.
- `{:error, reason}` - On failure.

### Request Example
```elixir
children = [
  %{id: WorkerTask, start: {Worker, :start_link, []}, restart: :transient}
]

{:ok, sup} = ConsumerSupervisor.start_link(children, 
  strategy: :one_for_one,
  subscribe_to: [{Producer, max_demand: 50}]
)
```
```

--------------------------------

### Producer-Consumer Initialization with Buffering and Subscription

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/configuration.md

Example of initializing a producer-consumer stage with custom buffer size, dispatcher, and subscription configuration including demand options.

```elixir
defmodule MyTransformer do
  use GenStage

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

  def init(multiplier) do
    {:producer_consumer, multiplier,
      # Producer side
      buffer_size: 1000,
      dispatcher: GenStage.DemandDispatcher,
      
      # Consumer side
      subscribe_to: [
        {UpstreamProducer, max_demand: 100, min_demand: 50}
      ]
    }
  end

  def handle_events(events, _from, multiplier) do
    transformed = Enum.map(events, &(&1 * multiplier))
    {:noreply, transformed, multiplier}
  end

  def handle_demand(_demand, state) do
    {:noreply, [], state}
  end
end
```

--------------------------------

### start/3

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Starts a GenStage process without linking to the current process. Useful for starting stages outside of supervision trees.

```APIDOC
## start/3

### Description
Starts a GenStage process without linking to the current process.

### Signature
```elixir
@spec start(module, term, GenServer.options()) :: GenServer.on_start()
```

### Parameters
#### Path Parameters
- **module** (atom) - Required - Callback module implementing GenStage behaviour
- **args** (term) - Required - Arguments passed to the module's init/1 callback
- **options** (keyword) - Optional - GenServer options including :name, :debug

### Return Type
`{:ok, pid}` on success, `{:error, reason}` on failure.

### Example
```elixir
{:ok, pid} = GenStage.start(MyProducer, 0)
```
```

--------------------------------

### Start GenStage Producer with Link

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Starts a GenStage process linked to the current process. The stage will call the init/1 callback from the given module with the provided args. The function blocks until init/1 returns.

```elixir
defmodule MyProducer do
  use GenStage

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

  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

# Start the producer
{:ok, pid} = MyProducer.start_link(0)
```

--------------------------------

### Custom Dispatcher Implementation Example

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

An example of a custom dispatcher module implementing the GenStage.Dispatcher behaviour. This includes basic implementations for each callback.

```elixir
defmodule MyCustomDispatcher do
  @behaviour GenStage.Dispatcher

  def init(opts) do
    {:ok, {[], %{}}}  # {consumers, state}
  end

  def subscribe(opts, {pid, ref}, {consumers, state}) do
    {:ok, 0, {[{pid, ref} | consumers], state}}
  end

  def ask(demand, {pid, ref}, {consumers, state}) do
    {:ok, demand, {consumers, state}}
  end

  def dispatch(events, _length, {consumers, state}) do
    for {pid, ref} <- consumers do
      Process.send(pid, {:"$gen_consumer", {self(), ref}, events}, [:noconnect])
    end
    {:ok, [], {consumers, state}}
  end

  def cancel({_pid, ref}, {consumers, state}) do
    consumers = Enum.reject(consumers, fn {_, r} -> r == ref end)
    {:ok, 0, {consumers, state}}
  end

  def info(msg, state) do
    send(self(), msg)
    {:ok, state}
  end
end
```

```elixir
# Usage
def init(state) do
  {:producer, state, dispatcher: MyCustomDispatcher}
end
```

--------------------------------

### Subscribe to Producers with Options

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/configuration.md

Example of initializing a consumer stage to subscribe to producers, specifying subscription options like max_demand and min_demand for each.

```elixir
def init(:ok) do
  {:consumer, :ok, subscribe_to: [
    {Producer1, max_demand: 100},
    {Producer2, max_demand: 50, min_demand: 25}
  ]}
end
```

--------------------------------

### start_child

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Starts a child process with additional arguments. This is useful for testing or special cases where a child needs to be started outside the normal event-driven flow. The provided arguments are appended to the child template's arguments.

```APIDOC
## start_child(supervisor, args)

### Description
Starts a child process with additional arguments. Useful for testing or special cases. The arguments in `args` are appended to the existing arguments in the child specification.

### Signature
`@spec start_child(Supervisor.supervisor(), [term]) :: Supervisor.on_start_child()`

### Parameters
#### Path Parameters
- supervisor (supervisor pid) - Required - The consumer supervisor
- args (list) - Required - Arguments to append to child template arguments

### Return Type
`{:ok, pid}` or `{:ok, pid, info}` on success, `:ignore` or `{:error, reason}` on failure.

### Example
```elixir
# If template is {Worker, :start_link, []}
# This becomes: Worker.start_link(extra_data)
{:ok, pid} = ConsumerSupervisor.start_child(supervisor, [extra_data])
```
```

--------------------------------

### Partitioned Producer-Consumer Configuration

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/configuration.md

Set up a GenStage as both a producer and consumer, subscribing to a specific partition of a master producer. This example also configures producer-side options like buffer_size.

```elixir
defmodule PartitionedStage do
  use GenStage

  def init(partition_id) do
    {:producer_consumer, %{partition: partition_id},
      # Consumer side
      subscribe_to: [{MasterProducer,
        partition: partition_id,
        max_demand: 100
      }],
      
      # Producer side
      dispatcher: GenStage.DemandDispatcher,
      buffer_size: 5000
    }
  end

  def handle_events(events, _from, state) do
    # Transform events
    transformed = Enum.map(events, &transform/1)
    {:noreply, transformed, state}
  end

  def handle_demand(demand, state) do
    {:noreply, [], state}
  end
end
```

--------------------------------

### Start GenStage Producer and Consumer

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/INDEX.md

Initiates a GenStage producer or a consumer that subscribes to a producer. For consumers, specify the producer to subscribe to using the `subscribe_to` option.

```elixir
# Producer
{:ok, pid} = GenStage.start_link(MyProducer, args)

# Consumer subscribing to producer
{:ok, pid} = GenStage.start_link(MyConsumer, args, subscribe_to: [MyProducer])
```

--------------------------------

### Subscribe to a Specific Partition (Named)

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Example of a consumer subscribing to a specific named partition (e.g., :odd) from a partition dispatcher.

```elixir
GenStage.sync_subscribe(consumer, to: producer, partition: :odd)
```

--------------------------------

### Producer Initialization with Broadcast Dispatcher

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Configure a GenStage producer to use the BroadcastDispatcher. This is the basic setup for broadcasting events.

```elixir
def init(state) do
  {:producer, state, dispatcher: GenStage.BroadcastDispatcher}
end
```

--------------------------------

### Start a Child Process with Arguments

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Use `start_child` to initiate a child process with additional arguments appended to its specification. This is useful for testing or specific scenarios outside the regular event flow.

```elixir
# If template is {Worker, :start_link, []}
# This becomes: Worker.start_link(extra_data)
{:ok, pid} = ConsumerSupervisor.start_child(supervisor, [extra_data])
```

--------------------------------

### Subscribe to a Specific Partition (Integer)

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Example of a consumer subscribing to a specific integer partition (e.g., partition 0) from a partition dispatcher.

```elixir
GenStage.sync_subscribe(consumer, to: producer, partition: 0)
```

--------------------------------

### Subscribe to Multiple Producers

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/configuration.md

Example of initializing a consumer stage to subscribe to multiple producers by their atom names.

```elixir
def init(:ok) do
  {:consumer, :ok, subscribe_to: [Producer1, Producer2]}
end
```

--------------------------------

### Subscribe to Single Producer

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/configuration.md

Example of initializing a consumer stage to subscribe to a single producer using its atom name.

```elixir
def init(:ok) do
  {:consumer, :ok, subscribe_to: [MyProducer]}
end
```

--------------------------------

### start_link/3

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Starts a GenStage process linked to the current process. The stage will call the init/1 callback from the given module with the provided args. The function blocks until init/1 returns.

```APIDOC
## start_link/3

### Description
Starts a GenStage process linked to the current process.

### Signature
```elixir
@spec start_link(module, term, GenServer.options()) :: GenServer.on_start()
```

### Parameters
#### Path Parameters
- **module** (atom) - Required - Callback module implementing GenStage behaviour
- **args** (term) - Required - Arguments passed to the module's init/1 callback
- **options** (keyword) - Optional - GenServer options including :name, :debug

### Return Type
`{:ok, pid}` on success, `{:error, reason}` on failure, or `:ignore`.

### Example
```elixir
defmodule MyProducer do
  use GenStage

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

  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

# Start the producer
{:ok, pid} = MyProducer.start_link(0)
```
```

--------------------------------

### Default Hash Function Example

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Illustrates the default hash function used by GenStage.PartitionDispatcher, which hashes events based on their value and the total number of partitions.

```elixir
fn event -> {event, :erlang.phash2(event, Enum.count(partitions))} end
```

--------------------------------

### Subscribe to Mixed Producers

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/configuration.md

Example demonstrating a mixed subscription list for a consumer stage, including direct producer names and producers with options, and remote producers.

```elixir
def init(:ok) do
  {:consumer, :ok, subscribe_to: [
    Producer1,
    {Producer2, max_demand: 100},
    {:"producer3@node1", max_demand: 50}
  ]}
end
```

--------------------------------

### Handle Events with Manual Demand Control

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/callbacks.md

This example shows how to manage event demand manually. After processing events, it uses `GenStage.ask/2` to request more events from the producer if the state indicates manual demand control.

```elixir
def handle_subscribe(:producer, opts, from, state) do
  max = Keyword.get(opts, :max_demand, 1000)
  {:manual, Map.put(state, :producer, {from, max})}
end

def handle_events(events, {_pid, ref}, state) do
  Enum.each(events, &process_event/1)
  
  # For manual demand, ask for more
  case state do
    %{producer: {from, max}} when ref == elem(from, 1) ->
      GenStage.ask(from, length(events))
    _ ->
      :ok
  end
  
  {:noreply, [], state}
end
```

--------------------------------

### Initialize GenStage with Producer or Consumer

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/callbacks.md

Implement the `init/1` callback to define the initial state and type of the GenStage. It can start as a producer, consumer, or producer-consumer, with optional configurations. Returns `:ignore` to exit silently or `{:stop, reason}` to exit with an error.

```elixir
def init({producer_arg, consumer_opts}) do
  if System.get_env("MODE") == "producer" do
    {:producer, producer_arg, buffer_size: 1000}
  else
    {:consumer, :ok, subscribe_to: [MyProducer]}
  end
end
```

--------------------------------

### init/1

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/callbacks.md

Invoked when the stage is started. It determines the stage type and initial state. Must return a valid stage type tuple, :ignore, or {:stop, reason}.

```APIDOC
## init/1

### Description
Invoked when the stage is started. Must determine and return the stage type. Returns `:ignore` to silently exit without error, or `{:stop, reason}` to exit with an error.

### Signature
```elixir
@callback init(args :: term) ::
            {:producer, state}
            | {:producer, state, [producer_option]}
            | {:producer_consumer, state}
            | {:producer_consumer, state, [producer_consumer_option]}
            | {:consumer, state}
            | {:consumer, state, [consumer_option]}
            | :ignore
            | {:stop, reason :: any}
          when state: any
```

### Parameters

#### args
* **Type:** `term`
* **Description:** Arguments passed to start_link/3 or start/3

### Return Values

* `{:producer, state}`: Start as producer with given state
* `{:producer, state, opts}`: Producer with options
* `{:consumer, state}`: Start as consumer
* `{:consumer, state, opts}`: Consumer with options
* `{:producer_consumer, state}`: Start as both producer and consumer
* `{:producer_consumer, state, opts}`: Producer-consumer with options
* `:ignore`: Don't start the stage
* `{:stop, reason}`: Start fails with reason

### Example
```elixir
def init({producer_arg, consumer_opts}) do
  if System.get_env("MODE") == "producer" do
    {:producer, producer_arg, buffer_size: 1000}
  else
    {:consumer, :ok, subscribe_to: [MyProducer]}
  end
end
```
```

--------------------------------

### GenStage ask/3 Example

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Demonstrates how to ask a producer for a specific number of events when GenStage is in manual demand mode. This is typically used within a consumer's handle_info callback after receiving a request for events.

```elixir
def handle_subscribe(:producer, opts, from, state) do
  {:manual, Map.put(state, :producer, from)}
end

def handle_info({:get_events, count}, %{producer: from} = state) do
  GenStage.ask(from, count)
  {:noreply, [], state}
end
```

--------------------------------

### Multi-Stage Transformation Pipeline in Elixir

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/patterns-and-examples.md

Demonstrates a GenStage pipeline with multiple transformation stages (doubling, incrementing) between a producer and a consumer. All stages must be started in order.

```Elixir
defmodule Producer do
  use GenStage

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

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

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

defmodule Doubler do
  use GenStage

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

  def init(:ok) do
    {:producer_consumer, :ok, subscribe_to: [Producer]}
  end

  def handle_demand(_demand, state) do
    {:noreply, [], state}
  end

  def handle_events(events, _from, state) do
    doubled = Enum.map(events, &(&1 * 2))
    {:noreply, doubled, state}
  end
end

defmodule Incrementer do
  use GenStage

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

  def init(:ok) do
    {:producer_consumer, :ok, subscribe_to: [Doubler]}
  end

  def handle_demand(_demand, state) do
    {:noreply, [], state}
  end

  def handle_events(events, _from, state) do
    incremented = Enum.map(events, &(&1 + 1))
    {:noreply, incremented, state}
  end
end

defmodule Consumer do
  use GenStage

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

  def init(:ok) do
    {:consumer, :ok, subscribe_to: [Incrementer]}
  end

  def handle_events(events, _from, state) do
    # Original: 0, 1, 2, ...
    # Doubled: 0, 2, 4, ...
    # Incremented: 1, 3, 5, ...
    Enum.each(events, &IO.inspect/1)
    {:noreply, [], state}
  end
end

# Start pipeline
{:ok, _} = Producer.start_link()
{:ok, _} = Doubler.start_link()
{:ok, _} = Incrementer.start_link()
{:ok, _} = Consumer.start_link()
```

--------------------------------

### Complete GenStage Pipeline with ConsumerSupervisor

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

This snippet defines an EventWorker, an EventProducer, and an EventConsumerSupervisor to create a complete GenStage pipeline. It shows how to start the producer and supervisor, and how events flow through the system with managed concurrency.

```elixir
# Define a worker that processes events
defmodule EventWorker do
  def start_link(event) do
    Task.start_link(fn ->
      IO.inspect({:processing, event})
      Process.sleep(100)  # Simulate work
      IO.inspect({:completed, event})
    end)
  end
end

# Define the producer
defmodule EventProducer do
  use GenStage

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

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

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

# Define the consumer supervisor
defmodule EventConsumerSupervisor do
  use ConsumerSupervisor

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

  @impl true
  def init(:ok) do
    children = [
      %{ 
        id: EventWorker,
        start: {EventWorker, :start_link, []},
        restart: :transient
      }
    ]

    opts = [
      strategy: :one_for_one,
      max_restarts: 5,
      max_seconds: 60,
      subscribe_to: [{EventProducer, max_demand: 20, min_demand: 10}]
    ]

    ConsumerSupervisor.init(children, opts)
  end
end

# Start the pipeline
{:ok, _producer} = EventProducer.start_link(name: EventProducer)
{:ok, _supervisor} = EventConsumerSupervisor.start_link(name: EventConsumerSupervisor)

# Events will now flow through the pipeline, with up to 20 concurrent workers

```

--------------------------------

### Create Producer from Enumerable

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Starts a producer stage from an enumerable or stream. The enumerable is consumed in batches matching the demand. When the enumerable finishes, the stage exits with :normal.

```elixir
 {:ok, producer} = GenStage.from_enumerable(1..100)
```

```elixir
 stream = Stream.unfold(0, fn n -> {n, n+1} end)
 {:ok, producer} = GenStage.from_enumerable(stream, max_demand: 50)
```

--------------------------------

### Basic Producer-Consumer Pipeline in Elixir

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/patterns-and-examples.md

Implements a simple GenStage pipeline where a producer emits events and a consumer processes them. Ensure Counter and Printer modules are defined and started.

```Elixir
defmodule Counter do
  use GenStage

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

  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

defmodule Printer do
  use GenStage

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

  def init(:ok) do
    {:consumer, :ok, subscribe_to: [Counter]}
  end

  def handle_events(events, _from, state) do
    Enum.each(events, &IO.inspect/1)
    {:noreply, [], state}
  end
end

# Start pipeline
{:ok, _} = Counter.start_link(0)
{:ok, _} = Printer.start_link()
```

--------------------------------

### Slow Consumer with Max/Min Demand

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/configuration.md

Configure a consumer to process events slowly by setting specific max and min demand values. This example shows how to limit the number of events processed concurrently and when to request more.

```elixir
defmodule SlowConsumer do
  use GenStage

  def init(:ok) do
    {:consumer, :ok,
      subscribe_to: [{FastProducer,
        max_demand: 10,
        min_demand: 5
      }]
    }
  end

  def handle_events(events, _from, state) do
    Enum.each(events, fn event ->
      # Slow processing
      Process.sleep(100)
      process_event(event)
    end)
    {:noreply, [], state}
  end
end
```

--------------------------------

### Fan-In Pattern: Many Producers to One Consumer in Elixir

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/patterns-and-examples.md

Demonstrates a fan-in pattern where multiple producers send events to a single consumer. The consumer subscribes to a list of producer names. Producers are started in a loop.

```Elixir
defmodule Producer do
  def start_link(id) do
    GenStage.start_link(__MODULE__, id, name: :
```

--------------------------------

### Start GenStage Producer without Link

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Starts a GenStage process without linking to the calling process. Useful for starting stages outside of supervision trees.

```elixir
{:ok, pid} = GenStage.start(MyProducer, 0)
```

--------------------------------

### Initialize with Named Partitions

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Configure the producer to use GenStage.PartitionDispatcher with a list of named partitions.

```elixir
def init(state) do
  {:producer, state, dispatcher: {GenStage.PartitionDispatcher, partitions: [:odd, :even]}}
end
```

--------------------------------

### Initialize with Integer Partitions

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Configure the producer to use GenStage.PartitionDispatcher with a specified number of integer partitions.

```elixir
def init(state) do
  {:producer, state, dispatcher: {GenStage.PartitionDispatcher, partitions: 4}}
end
```

--------------------------------

### Implement Supervisor init/1 Callback

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Implement this callback to define the supervisor's children and options. It must return a list with exactly one child specification and supervisor options like `:strategy`, `:max_restarts`, `:max_seconds`, and `:subscribe_to`.

```elixir
@impl true
def init(opts) do
  children = [
    %{id: TaskWorker, start: {Task, :start_link, []}, restart: :transient}
  ]
  
  supervisor_opts = [
    strategy: :one_for_one,
    max_restarts: 10,
    max_seconds: 60,
    subscribe_to: [{MyProducer, max_demand: 100, min_demand: 50}]
  ]
  
  {:ok, children, supervisor_opts}
end
```

--------------------------------

### Producer Init with Partition Dispatcher

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/types.md

Initializes a producer stage using a partition dispatcher with a specified range of partitions. Used in the options of the `init/1` return tuple.

```elixir
def init(state) do
  {:producer, state,
    dispatcher: {GenStage.PartitionDispatcher, partitions: 0..3}
  }
end
```

--------------------------------

### Handle Subscription with Manual Demand

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/callbacks.md

Implement this callback to manage subscription initialization and enable manual demand control. Consumers can return `:manual` to explicitly control demand via `ask/3`.

```elixir
def handle_subscribe(:producer, opts, from, state) do
  # Store subscription for later manual demand
  max = Keyword.get(opts, :max_demand, 1000)
  subscription = {from, max}
  {:manual, Map.put(state, :subscription, subscription)}
end

def handle_info({:request_more, count}, %{subscription: from} = state) do
  GenStage.ask(from, count)
  {:noreply, [], state}
end
```

--------------------------------

### Get Producer Demand Mode

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Retrieves the current demand forwarding mode of a producer. The mode can be either `:forward` or `:accumulate`.

```elixir
mode = GenStage.demand(producer)
```

--------------------------------

### Initialize with DemandDispatcher

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/backpressure-guide.md

Use DemandDispatcher for scenarios with varying consumer speeds and load balancing needs. It sends events to the consumer with the highest demand.

```elixir
def init(state) do
  {:producer, state, dispatcher: GenStage.DemandDispatcher}
end
```

--------------------------------

### Initialize with PartitionDispatcher

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/backpressure-guide.md

Use PartitionDispatcher for partitioned processing where each partition has independent workloads. Each partition manages its own demand.

```elixir
def init(state) do
  {:producer, state,
    dispatcher: {GenStage.PartitionDispatcher, partitions: 0..2}
  }
end
```

--------------------------------

### GenStage.from_enumerable/2

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Starts a producer stage from an enumerable or stream. The enumerable is consumed in batches matching the demand. When the enumerable finishes, the stage exits with :normal.

```APIDOC
## GenStage.from_enumerable(enumerable, opts \\ [])

### Description
Starts a producer stage from an enumerable or stream. The enumerable is consumed in batches matching the demand. When the enumerable finishes, the stage exits with :normal.

### Signature
```elixir
@spec from_enumerable(Enumerable.t(), keyword) :: GenServer.on_start()
```

### Parameters
#### Path Parameters
- **enumerable** (Enumerable) - Required - The enumerable or stream to produce from
- **opts** (keyword) - Optional - Configuration options

#### Options
- **:link** (boolean) - Optional - Default: `true` - Whether to link to the current process
- **:dispatcher** (atom or tuple) - Optional - Default: `GenStage.DemandDispatcher` - Dispatcher module and options
- **:demand** (atom) - Optional - Default: `:forward` - Demand mode (:forward or :accumulate)
- **:on_cancel** (atom) - Optional - Default: `—` - What to do when all consumers cancel (:stop)
- **:stacktrace** (list) - Optional - Default: `—` - Stacktrace for debugging

### Return Type
`{:ok, pid}` on success, `{:error, reason}` on failure.

### Example
```elixir
# From an enumerable
{:ok, producer} = GenStage.from_enumerable(1..100)

# From a stream with custom max_demand
stream = Stream.unfold(0, fn n -> {n, n+1} end)
{:ok, producer} = GenStage.from_enumerable(stream, max_demand: 50)
```
```

--------------------------------

### Producer-Consumer Init with Mixed Options

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/types.md

Initializes a producer-consumer stage with buffer, dispatcher, and subscription configurations. Used in the options of the `init/1` return tuple.

```elixir
def init(state) do
  {:producer_consumer, state,
    buffer_size: :infinity,
    dispatcher: GenStage.DemandDispatcher,
    subscribe_to: [Producer1, {Producer2, max_demand: 100}]
  }
end
```

--------------------------------

### Check Buffered Count with GenStage

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/backpressure-guide.md

Use GenStage.estimate_buffered_count/1 to get an estimate of the number of events currently buffered by the producer. This is useful for monitoring and debugging.

```elixir
buffered = GenStage.estimate_buffered_count(producer)
IO.puts("Producer has #{buffered} events buffered")
```

--------------------------------

### ConsumerSupervisor Callback: init/1

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Callback invoked to initialize the supervisor. Must return a list with exactly one child specification and supervisor options.

```APIDOC
## `c:init/1`

### Description
Callback invoked to initialize the supervisor. Must return a list with exactly one child specification and supervisor options. The options may include `:strategy`, `:max_restarts`, `:max_seconds`, and `:subscribe_to`.

### Signature
```elixir
@callback init(args :: term) ::
            {:ok, [:supervisor.child_spec()], options :: keyword()}
            | :ignore
```

### Parameters
#### Path Parameters
- **args** (term) - Required - Arguments from `start_link/3`

### Return Type
`{:ok, children, options}` or `:ignore`.

### Example
```elixir
@impl true
def init(opts) do
  children = [
    %{ 
      id: TaskWorker,
      start: {Task, :start_link, []},
      restart: :transient
    }
  ]
  
  supervisor_opts = [
    strategy: :one_for_one,
    max_restarts: 10,
    max_seconds: 60,
    subscribe_to: [{MyProducer, max_demand: 100, min_demand: 50}]
  ]
  
  {:ok, children, supervisor_opts}
end
```
```

--------------------------------

### Producer Init with Buffer and Dispatcher Options

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/types.md

Initializes a producer stage with custom buffer size, event keeping strategy, and a specific dispatcher module. Used in the options of the `init/1` return tuple.

```elixir
def init(state) do
  {:producer, state, 
    buffer_size: 50_000,
    buffer_keep: :first,
    dispatcher: GenStage.BroadcastDispatcher
  }
end
```

--------------------------------

### Get Demand Mode

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Retrieves the current demand forwarding mode of a producer stage. This mode dictates whether demand is immediately processed or accumulated.

```APIDOC
## `demand(stage)`

### Description
Returns the current demand mode for a producer.

### Method
`demand/1`

### Parameters
#### Path Parameters
- **stage** (stage term) - Required - The producer stage

### Return Type
`:forward` | `:accumulate`

### Description
Returns the current demand forwarding mode. `:forward` means demand is immediately forwarded to `handle_demand/2`. `:accumulate` means demand is accumulated until switched to `:forward`.

### Request Example
```elixir
mode = GenStage.demand(producer)
```
```

--------------------------------

### Consumer Init with Multiple Subscriptions

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/types.md

Initializes a consumer stage to subscribe to multiple producers, including options for specific producers. Used in the options of the `init/1` return tuple.

```elixir
def init(state) do
  {:consumer, state, 
    subscribe_to: [
      Producer1,
      {Producer2, max_demand: 100},
      {Producer3, max_demand: 50, min_demand: 25}
    ]
  }
end
```

--------------------------------

### Initialize Consumer Supervisor with Helper

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Use this helper function within your callback module's `init/1` to validate child specifications and format the initialization tuple.

```elixir
defmodule MyConsumerSupervisor do
  use ConsumerSupervisor

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

  def init(_arg) do
    children = [
      worker(Worker, [], restart: :transient)
    ]
    
    opts = [strategy: :one_for_one, subscribe_to: [MyProducer]]
    
    ConsumerSupervisor.init(children, opts)
  end
end
```

--------------------------------

### Test GenStage Producer Stream

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/patterns-and-examples.md

Use GenStage.stream/1 to test if a producer emits the correct sequence of events. Ensure the producer is started and then subscribe to its stream for verification.

```elixir
defmodule ProducerTest do
  use ExUnit.Case

  test "producer emits correct sequence" do
    {:ok, producer} = GenStage.start_link(MyProducer, 0)
    
    stream = GenStage.stream([producer])
    
    result = stream
    |> Stream.take(10)
    |> Enum.to_list()
    
    assert result == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  end
end
```

--------------------------------

### Reduce Demand for Memory Constraints

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/backpressure-guide.md

If events are large, reduce the `max_demand` to limit memory buffering. This example sets the buffer to 100MB assuming 1MB events.

```elixir
# Each event is 1MB
# Default: max_demand = 1000
# Memory = 1000 * 1MB = 1GB

# Reduce to 100MB buffering
GenStage.sync_subscribe(consumer, to: producer, max_demand: 100)
```

--------------------------------

### GenStage cancel/3 Example

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/genstage-api-reference.md

Shows how to cancel a subscription with a producer. This function is used to signal to the producer that the consumer no longer wishes to receive events, providing a reason for the cancellation.

```elixir
GenStage.cancel({producer_pid, ref}, :shutdown)
```

--------------------------------

### Get Consumer Supervisor Child Information

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/consumer-supervisor.md

Retrieves a list of tuples containing information about each child process within the supervisor. Useful for debugging and monitoring active or restarting children.

```elixir
children = ConsumerSupervisor.which_children(supervisor)
# => [
#      {:undefined, #PID<0.123.0>, :worker, [MyWorker]},
#      {:undefined, #PID<0.124.0>, :worker, [MyWorker]},
#      {:undefined, :restarting, :worker, [MyWorker]}
#    ]
```

--------------------------------

### Configure GenStage to use DemandDispatcher

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/dispatchers.md

Demonstrates how to configure a GenStage producer to use the DemandDispatcher, either by default or explicitly. Options like `shuffle_demands_on_first_dispatch` can be passed.

```elixir
def init(state) do
  {:producer, state}  # Uses DemandDispatcher by default
end
```

```elixir
def init(state) do
  {:producer, state, dispatcher: GenStage.DemandDispatcher}
end
```

```elixir
def init(state) do
  {:producer, state, dispatcher: {GenStage.DemandDispatcher, shuffle_demands_on_first_dispatch: true}}
end
```

--------------------------------

### Consumer: Manual Demand for Asynchronous Processing

Source: https://github.com/elixir-lang/gen_stage/blob/main/_autodocs/backpressure-guide.md

Example of a consumer using manual demand to manage asynchronous event processing. It controls the flow of events based on inflight tasks and capacity.

```elixir
defmodule AsyncConsumer do
  use GenStage

  def init(:ok) do
    {:consumer, %{inflight: 0, max_concurrent: 10, subscription: nil}}
  end

  def handle_subscribe(:producer, opts, from, state) do
    # Manual to control async work
    max = Keyword.get(opts, :max_demand, 100)
    {:manual, %{state | subscription: from, max_concurrent: max}}
  end

  def handle_events([event], _from, state) do
    # Start async work
    Task.start(fn ->
      process_event(event)
      GenStage.cast(self(), :task_complete)
    end)
    
    # Ask for next event only when we have capacity
    state = %{state | inflight: state.inflight + 1}
    
    if state.inflight < state.max_concurrent do
      GenStage.ask(state.subscription, 1)
    end
    
    {:noreply, [], state}
  end

  def handle_cast(:task_complete, state) do
    state = %{state | inflight: state.inflight - 1}
    
    # Ask for more work
    GenStage.ask(state.subscription, 1)
    
    {:noreply, [], state}
  end
end
```