### prepare_for_start/2 example implementation

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-producer.md

Example implementation of prepare_for_start/2 that starts a DynamicSupervisor and enables rate limiting. It returns the child specs and updated options. A bad return (non-{list, opts} tuple) raises an ArgumentError.

```elixir
defmodule MyProducer do
  @behaviour Broadway.Producer

  # other GenStage callbacks...

  @impl true
  def prepare_for_start(_module, broadway_options) do
    children = [
      {DynamicSupervisor, strategy: :one_for_one, name: MyApp.DynamicSupervisor}
    ]

    updated_options =
      put_in(broadway_options, [:producer, :rate_limiting],
        interval: 1000,
        allowed_messages: 10
      )

    {children, updated_options}
  end
end
```

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

### Install gcloud with Homebrew

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/google-cloud-pubsub.md

Installs the Google Cloud SDK on macOS using Homebrew.

```bash
$ brew install --cask google-cloud-sdk
```

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

### Combining with ack_immediately/1

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-noop-acknowledger.md

Example handle_message/3 that runs a side effect and then calls ack_immediately/1, which installs the no-op acknowledger.

```elixir
def handle_message(_processor, message, _context) do
  MySideEffect.run(message.data)
  Broadway.Message.ack_immediately(message)  # acknowledger becomes {NoopAcknowledger, nil, nil}
end
```

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

### Error handling example for start_link/2

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Example showing how to handle an ArgumentError raised by start_link/2 when invalid options are given. It logs the error message.

```elixir
# {:error, %ArgumentError{message: "invalid configuration given to Broadway.start_link/2 for key [:producer], ..."}}
case Broadway.start_link(MyBroadway, name: MyBroadway, producer: []) do
  {:ok, pid} -> {:ok, pid}
  {:error, %ArgumentError{} = err} -> Logger.error("invalid config: #{Exception.message(err)}")
end
```

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

### Expected log output from :brod supervisor

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/apache-kafka.md

Example of an info log message from the :brod supervisor when the pipeline starts successfully.

```text
15:14:04.356 [info]  [supervisor: {:local, :brod_sup}, started: [pid: #PID<0.251.0>, id: :test_client, mfargs: {:brod_client, :start_link, [[localhost: 9092], :test_client, []]}, restart_type: {:permanent, 10}, shutdown: 5000, child_type: :worker]]
```

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

### Partition-aware handling of batch info

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-batch-info.md

Example showing how to handle batches with and without a partition, using pattern matching on the BatchInfo struct.

```elixir
# Partition-aware handling: `:partition` is present only when :partition_by is configured
def handle_batch(:s3, messages, %Broadway.BatchInfo{partition: nil}, _context) do
  upload(messages)
end

def handle_batch(:s3, messages, %Broadway.BatchInfo{partition: partition}, _context) do
  upload(messages, shard: partition)
end
```

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

### prepare_for_start/2

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-producer.md

Invoked once by Broadway during start_link/2, before the producer supervisor boots. It allows manipulating topology options and introducing child specs that start before the producers' supervisor.

```APIDOC
## [CALLBACK] prepare_for_start/2

### Description
Invoked once by Broadway during `Broadway.start_link/2`, inside the Broadway main process and before the producer supervisor boots. Its purpose is to manipulate topology options and introduce child specs started **before** the producers' supervisor in Broadway's supervision tree. Since the tree is a `rest_for_one` supervisor, a crash in these children brings down the whole pipeline before restart.

### Callback Signature
```elixir
@callback prepare_for_start(module :: atom, options :: keyword) ::
            {[child_spec], updated_options :: keyword}
          when child_spec: :supervisor.child_spec() | {module, any} | module
```

### Parameters
- **module** (atom) - Required - The Broadway module passed to `start_link/2`.
- **options** (keyword) - Required - All topology options passed to `start_link/2`.

### Returns
- **{child_specs, updated_options}** (tuple) - The child specs to start under Broadway's supervision tree, and any updated options used instead of those passed to `start_link/2` (re-validated against `Broadway.Options.definition()`).

### Errors
- Returning a non-`{list, opts}` tuple raises `ArgumentError` with message `"expected <mfa> to return {child_specs, options}, got: ..."`.

### Example
```elixir
defmodule MyProducer do
  @behaviour Broadway.Producer

  # other GenStage callbacks...

  @impl true
  def prepare_for_start(_module, broadway_options) do
    children = [
      {DynamicSupervisor, strategy: :one_for_one, name: MyApp.DynamicSupervisor}
    ]

    updated_options =
      put_in(broadway_options, [:producer, :rate_limiting],
        interval: 1000,
        allowed_messages: 10
      )

    {children, updated_options}
  end
end
```
```

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

### Basic Broadway pipeline example

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Example of a Broadway pipeline module using start_link/2 with producer, processors, and batchers. It defines handle_message/3 and handle_batch/4 callbacks.

```elixir
defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [module: {MyProducer, []}, concurrency: 1],
      processors: [default: [concurrency: 10]],
      batchers: [s3: [concurrency: 1, batch_size: 100, batch_timeout: 1000]]
    )
  end

  @impl true
  def handle_message(_processor, message, _context), do: message

  @impl true
  def handle_batch(_batcher, messages, _batch_info, _context), do: messages
end
```

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

### Start iex session

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/apache-kafka.md

Command to start an interactive Elixir session to test the Broadway pipeline.

```shell
$ iex -S mix
```

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

### Add Broadway to supervision tree

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/amazon-sqs.md

Shows how to start the Broadway pipeline by adding it as a child to a supervisor. The pipeline starts when the application starts, and should be listed after its dependencies.

```elixir
children = [
  {MyBroadway, []}
]

Supervisor.start_link(children, strategy: :one_for_one)
```

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

### Tuning Broadway pipeline configuration with concurrency, max_demand, and batch_size

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/amazon-sqs.md

Example configuration for a Broadway pipeline that tunes concurrency for producers, processors, and batchers, along with max_demand and batch_size. Use this as a starting point to adjust values based on your workload and server limitations. The example includes placeholder ellipses for producer options and callbacks.

```elixir
defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        ...
        concurrency: 10,
      ],
      processors: [
        default: [
          concurrency: 100,
          max_demand: 1,
        ]
      ],
      batchers: [
        default: [
          batch_size: 10,
          concurrency: 10,
        ]
      ]
    )
  end

  ...callbacks...
end
```

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

### start_link/2

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Starts a Broadway pipeline linked to the caller. Validates options and raises ArgumentError on invalid configuration.

```APIDOC
## start_link/2

### Description
Starts a Broadway pipeline linked to the caller. `module` is the module implementing the `Broadway` behaviour, `opts` the pipeline blueprint. Validates options with `NimbleOptions.validate/2`, raising `ArgumentError` on failure. Also raises `ArgumentError` when any batcher has a tuple `:batch_size` without an integer `:max_demand`. Carries top-level `:hibernate_after`/`:spawn_opt` into the producer and top-level `:partition_by`/`:hibernate_after`/`:spawn_opt` into each processor and batcher, and starts the topology.

### Method
N/A (Elixir function)

### Endpoint
N/A

### Parameters
#### Function Parameters
- **module** (`module()`) - Required - Module implementing the `Broadway` behaviour
- **opts** (`keyword()`) - Required - Pipeline options; `:name`, `:producer`, and `:processors` are required

### Returns
`t:on_start/0` — `{:ok, pid}` on success; `:ignore`; or `{:error, {:already_started, pid} | term}`.

### Raises
- `ArgumentError` - Invalid option values (formatted `NimbleOptions.ValidationError`)
- `ArgumentError` - Batcher with tuple `:batch_size` and no integer `:max_demand`

### Example
```elixir
defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [module: {MyProducer, []}, concurrency: 1],
      processors: [default: [concurrency: 10]],
      batchers: [s3: [concurrency: 1, batch_size: 100, batch_timeout: 1000]]
    )
  end

  @impl true
  def handle_message(_processor, message, _context), do: message

  @impl true
  def handle_batch(_batcher, messages, _batch_info, _context), do: messages
end
```

Error handling — a bad option raises at boot:

```elixir
# {:error, %ArgumentError{message: "invalid configuration given to Broadway.start_link/2 for key [:producer], ..."}}
case Broadway.start_link(MyBroadway, name: MyBroadway, producer: []) do
  {:ok, pid} -> {:ok, pid}
  {:error, %ArgumentError{} = err} -> Logger.error("invalid config: #{Exception.message(err)}")
end
```
```

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

### get_rate_limiting/1 example

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Reads the rate-limiting configuration and matches the success tuple.

```elixir
{:ok, %{allowed_messages: 2000, interval: 1000}} = Broadway.get_rate_limiting(broadway)
```

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

### prepare_for_start/2 callback specification

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-producer.md

Defines the callback signature for prepare_for_start/2. It must return a tuple of child specs and updated options. Used by Broadway to allow producers to modify topology options and add child specs before the producer supervisor starts.

```elixir
@callback prepare_for_start(module :: atom, options :: keyword) ::
            {[child_spec], updated_options :: keyword}
          when child_spec: :supervisor.child_spec() | {module, any} | module
```

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

### Broadway.start_link/2

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Starts a Broadway pipeline as part of a supervision tree. This is the main entry point for creating a pipeline.

```APIDOC
## [start_link] [Broadway.start_link/2]

### Description
Starts a Broadway pipeline with the given name and options. This function is typically called from a supervisor to start the pipeline.

### Method
start_link

### Endpoint
Broadway.start_link/2

### Parameters
#### Arguments
- **name** (atom() | {:via, module(), term()}) - Required - The name to register the pipeline process under.
- **options** (keyword()) - Required - Options for the pipeline, including producers, processors, and batchers.

### Request Example
```elixir
Broadway.start_link(MyPipeline, name: MyPipeline)
```

### Response
#### Success Response
- **{:ok, pid()}** - The PID of the pipeline process.
- **:ignore** - If the pipeline is intentionally not started.
- **{:error, {:already_started, pid()}}** - If a pipeline with the same name is already running.

#### Response Example
```elixir
{:ok, #PID<0.123.0>}
```
```

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

### Broadway.on_start/0 type declaration

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

Defines the return type of `Broadway.start_link/2`. The `:ok` variant indicates successful start, `:ignore` indicates the pipeline declined to start, and `{:error, ...}` covers failures including `{:already_started, pid()}` when a pipeline with the same name is already running.

```elixir
@type on_start() :: {:ok, pid()} | :ignore | {:error, {:already_started, pid()} | term()}
```

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

### test_batch/3 bulk mode example

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Pushes a batch in bulk mode and receives the ack. No ordering is guaranteed; a bulk batch may ack in separate messages.

```elixir
ref = Broadway.test_batch(MyBroadway, Enum.to_list(1..1000), batch_mode: :bulk)

receive do
  {:ack, ^ref, successful, failed} -> {successful, failed}
end
```

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

### Acknowledger with configure/3 and usage

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-acknowledger.md

A module that implements both ack/3 and configure/3, followed by a usage example of Broadway.Message.configure_ack/2.

```elixir
defmodule ConfiguredAcknowledger do
  @behaviour Broadway.Acknowledger

  @impl true
  def ack(_ack_ref, _successful, _failed), do: :ok

  @impl true
  def configure(ack_ref, ack_data, options) do
    {:ok, Map.merge(ack_data, Map.new(options))}
  end
end

```

```elixir
# used from a pipeline callback:
message = Broadway.Message.configure_ack(message, requeue: false)

```

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

### Broadway.start_link/2 full pipeline configuration

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

Use this snippet to configure a complete Broadway pipeline with an SQS producer, custom partitioning, and batchers for S3. The example sets concurrency, rate limiting, and batch sizes. It references MyApp.Repo and custom functions partition/1 and custom_splitter/2.

```elixir
Broadway.start_link(__MODULE__,
  name: __MODULE__,
  context: %{repo: MyApp.Repo},
  producer: [
    module: {BroadwaySQS.Producer, queue_url: "https://.../my_queue"},
    concurrency: 4,
    rate_limiting: [allowed_messages: 2000, interval: 1000]
  ],
  processors: [
    default: [concurrency: 50, max_demand: 20, partition_by: &partition/1]
  ],
  batchers: [
    default: [concurrency: 5, batch_size: 100, batch_timeout: 500],
    s3: [concurrency: 2, batch_size: {0, &custom_splitter/2}]
  ],
  partition_by: &partition/1
)
```

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

### update_rate_limiting/2 success and error examples

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Updates rate limiting with allowed_messages and interval. The error example handles the case when rate limiting is not enabled.

```elixir
:ok = Broadway.update_rate_limiting(broadway, allowed_messages: 100)
:ok = Broadway.update_rate_limiting(broadway, allowed_messages: 50, interval: 500, reset: true)

# error example — pipeline without rate limiting
case Broadway.update_rate_limiting(broadway, allowed_messages: 10) do
  :ok -> :ok
  {:error, :rate_limiting_not_enabled} -> Logger.warn("rate limiting is off")
end
```

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

### SQS integration example with Broadway

Source: https://github.com/elixir-broadway/broadway/blob/main/README.md

Defines a Broadway module consuming Amazon SQS events. Requires `broadway_sqs` dependency and configured SQS credentials. Configures a producer, 50 concurrent processors, and an S3 batcher with batch size 10 and timeout 1000 ms. Overrides `handle_message/3` to process data and route to the `:s3` batcher, and `handle_batch/4` to send batches to S3.

```elixir
defmodule MyBroadway do
  use Broadway

  alias Broadway.Message

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwaySQS.Producer, queue_url: "https://us-east-2.queue.amazonaws.com/100000000001/my_queue"}
      ],
      processors: [
        default: [concurrency: 50]
      ],
      batchers: [
        s3: [concurrency: 5, batch_size: 10, batch_timeout: 1000]
      ]
    )
  end

  def handle_message(_processor_name, message, _context) do
    message
    |> Message.update_data(&process_data/1)
    |> Message.put_batcher(:s3)
  end

  def handle_batch(:s3, messages, _batch_info, _context) do
    # Send batch of messages to S3
  end

  defp process_data(data) do
    # Do some calculations, generate a JSON representation, process images.
  end
end
```

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

### Declare a RabbitMQ queue with rabbitmqadmin

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/rabbitmq.md

Use this command to create a durable queue named 'my_queue' in RabbitMQ. Requires RabbitMQ installed with the Management plugin enabled.

```shell
$ rabbitmqadmin declare queue name=my_queue durable=true
```

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

### Implement Broadway callbacks (handle_message/3 and handle_batch/4)

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/amazon-sqs.md

Defines a MyBroadway module using Broadway. handle_message/3 squares the message data, and handle_batch/4 prints the batch data. This is a minimal example for processing numeric messages.

```elixir
defmodule MyBroadway do
  use Broadway

  alias Broadway.Message

  ...start_link...

  @impl true
  def handle_message(_, %Message{data: data} = message, _) do
    message
    |> Message.update_data(fn data -> data * data end)
  end

  @impl true
  def handle_batch(_, messages, _, _) do
    list = messages |> Enum.map(fn e -> e.data end)
    IO.inspect(list, label: "Got batch of finished jobs from processors, sending ACKs to SQS as a batch.")
    messages
  end
end
```

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

### Set config_storage to :ets in config/config.exs

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

Use this in `config/config.exs` to select the `:ets` backend for storing pipeline topology configs. Recommended when pipelines are started dynamically to avoid persistent-term GC pauses.

```elixir
# config/config.exs
config :broadway, config_storage: :ets
```

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

### Stop a Broadway pipeline with default and custom options

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Examples of calling Broadway.stop/3. The first stops with default reason and timeout. The second stops with reason :shutdown and a 10-second timeout.

```elixir
:ok = Broadway.stop(MyBroadway)
:ok = Broadway.stop(MyBroadway, :shutdown, 10_000)
```

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

### Default child_spec/1 implementation

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Default child_spec/1 injected by use Broadway. It sets the id to the module, start to {__MODULE__, :start_link, [arg]}, and shutdown to :infinity for graceful drain on shutdown.

```elixir
def child_spec(arg) do
  default = %{
    id: unquote(module),
    start: {__MODULE__, :start_link, [arg]},
    shutdown: :infinity
  }
  Supervisor.child_spec(default, unquote(Macro.escape(opts)))
end
```

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

### Create Google Cloud project

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/google-cloud-pubsub.md

Creates a new Google Cloud project named test-pubsub.

```bash
$ gcloud projects create test-pubsub
```

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

### Create service account

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/google-cloud-pubsub.md

Creates a service account for the project and shows the expected output.

```bash
$ gcloud iam service-accounts create test-account --project test-pubsub
Created service account [test-account].
```

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

### Error handling for prepare_for_start/2 return

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-producer.md

Shows how Broadway.Topology.prepare_for_start/2 validates the return value from the producer's prepare_for_start/2. If the return is not a {child_specs, options} tuple, it raises an ArgumentError with a specific message.

```elixir
# inside Broadway.Topology.prepare_for_start/2 (lib/broadway/topology.ex:165-170)
case producer_mod.prepare_for_start(module, opts) do
  {child_specs, opts} when is_list(child_specs) ->
    {child_specs, NimbleOptions.validate!(opts, Broadway.Options.definition())}

  other ->
    raise ArgumentError,
          "expected #{Exception.format_mfa(producer_mod, :prepare_for_start, 2)} " <>
            "to return {child_specs, options}, got: #{inspect(other)}"
end
```

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

### Create Pub/Sub subscription

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/google-cloud-pubsub.md

Creates a Pub/Sub subscription for the test-topic and shows the expected output.

```bash
$ gcloud pubsub subscriptions create test-subscription --project test-pubsub --topic test-topic
Created subscription [projects/test-pubsub/subscriptions/test-subscription].
```

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

### Enable Pub/Sub API

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/google-cloud-pubsub.md

Enables the Pub/Sub API for the project and shows the expected output.

```bash
$ gcloud services enable pubsub --project test-pubsub
Operation "operations/xxx" finished successfully.
```

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

### Create a Kafka topic

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/apache-kafka.md

Command to create a new Kafka topic named 'test' with 3 partitions using the kafka-topics CLI.

```bash
$ kafka-topics --create --zookeeper localhost:2181 --partitions 3 --topic test
```

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

### Create Pub/Sub topic

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/google-cloud-pubsub.md

Creates a Pub/Sub topic and shows the expected output. May fail if the project's organization policy is still provisioning; wait and retry.

```bash
$ gcloud pubsub topics create test-topic --project test-pubsub
Created topic [projects/test-pubsub/topics/test-topic].
```

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

### Add Broadway dependency in mix.exs

Source: https://github.com/elixir-broadway/broadway/blob/main/README.md

Add `:broadway` to the list of dependencies in `mix.exs` to install the library. Use version `~> 1.0`.

```elixir
def deps do
  [
    {:broadway, "~> 1.0"}
  ]
end
```

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

### Acknowledge a batch of messages immediately

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-message.md

Example of calling ack_immediately/1 on a list of messages to acknowledge a whole batch as soon as it is received.

```elixir
# ack a whole batch as soon as it is received
messages = Broadway.Message.ack_immediately(messages)
```

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

### Using NoopAcknowledger in a producer

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-noop-acknowledger.md

Example producer module that creates Broadway.Message structs with the no-op acknowledger, so messages require no acknowledgement.

```elixir
defmodule MyProducer do
  use GenStage
  @behaviour Broadway.Producer

  @impl true
  def init(_opts) do
    state = %{data: [1, 2, 3]}
    {:producer, state}
  end

  @impl true
  def handle_demand(demand, %{data: data} = state) do
    {to_send, rest} = Enum.split(data, demand)

    events =
      Enum.map(to_send, fn item ->
        %Broadway.Message{
          data: item,
          acknowledger: Broadway.NoopAcknowledger.init()
        }
      end)

    {:noreply, events, %{state | data: rest}}
  end
end
```

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

### Expected output showing generated batches

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/apache-kafka.md

Example of the output from the Broadway pipeline, showing batches of processed messages with doubled values.

```text
Got batch: [
  {"2", 4},
  {"5", 10},
  {"8", 16},
  {"11", 22},
  {"14", 28},
  ...
]
Got batch: [
  {"3", 6},
  {"6", 12},
  {"9", 18},
  {"12", 24},
  {"15", 30},
  ...
]
```

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

### Config for test and dev environments

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-dummy-producer.md

Shows configuration files for test and dev environments. In test, DummyProducer is used with empty options; in dev, a real producer (BroadwaySQS.Producer) is configured with a queue URL.

```elixir
# config/test.exs
config :my_app,
  producer_module: Broadway.DummyProducer,
  producer_options: []

# config/dev.exs
config :my_app,
  producer_module: BroadwaySQS.Producer,
  producer_options: [queue_url: "..."]
```

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

### Acknowledge a single message immediately

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-message.md

Example of calling ack_immediately/1 on a single message. The message is acknowledged immediately without waiting for the rest of the pipeline.

```elixir
message = Broadway.Message.ack_immediately(message)
```

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

### Create a new Elixir project with supervision tree

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/amazon-sqs.md

Run this command to generate a new Elixir application with a supervision tree, which is required for Broadway.

```shell
$ mix new my_app --sup
```

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

### Reading batch info in a pipeline

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-batch-info.md

Example of a Broadway pipeline that uses the batch_info argument in handle_batch/4 to log batch metadata and branch on the trigger.

```elixir
defmodule MyBroadway do
  use Broadway

  @impl true
  def handle_batch(_batcher, messages, batch_info, _context) do
    Logger.info("batch of #{batch_info.size} messages for key #{inspect(batch_info.batch_key)}, " <>
                "triggered by #{batch_info.trigger}")

    case batch_info.trigger do
      :timeout -> compact_and_send(messages)
      _ -> send_batch(messages)
    end
  end
end
```

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

### Configure RabbitMQ connection with credentials

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/rabbitmq.md

Shows how to provide connection credentials (username and password) for an existing broker that requires authorization. The connection option is passed to the producer module.

```elixir
...
producer: [
  module: {BroadwayRabbitMQ.Producer,
    queue: "my_queue",
    connection: [
      username: "user",
      password: "password",
    ]
    ...
  }
]
...
```

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

### Minimal Broadway pipeline configuration for SQS

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/amazon-sqs.md

Defines the start_link/1 function that delegates to Broadway.start_link/2, setting up the producer for an SQS queue, a default processor, and a default batcher with a batch size of 10 and a timeout of 2000 ms. Requires AWS credentials to be set in the environment.

```elixir
defmodule MyBroadway do
  use Broadway

  alias Broadway.Message

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwaySQS.Producer,
                 queue_url: "https://us-east-2.queue.amazonaws.com/100000000001/my_queue"}
      ],
      processors: [
        default: []
      ],
      batchers: [
        default: [
          batch_size: 10,
          batch_timeout: 2000
        ]
      ]
    )
  end

  ...callbacks...
end
```

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

### GenStage producer example (Counter)

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/custom-producers.md

A regular GenStage producer that emits plain events (integers) and does not generate Broadway.Message structs. It must be wrapped with a transformer to be used with Broadway.

```elixir
defmodule Counter do
  use GenStage

  def start_link(number) do
    GenStage.start_link(Counter, number)
  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_link/2 function specification

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Specification for the start_link/2 function, showing the expected types and default implementation.

```elixir
@spec start_link(module(), keyword()) :: on_start()
def start_link(module, opts)
```

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

### c:prepare_messages/2

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway.md

Callback invoked before handle_message/3 to preload shared data. Must return all received messages.

```APIDOC
## c:prepare_messages/2

### Description
Invoked before `handle_message/3` with a list of messages, as a best-effort place to preload shared data (e.g., a single DB query). Must return **all** messages it received; otherwise the processor raises `"expected all messages to be returned from prepared_messages/2"`.

### Callback Signature
```elixir
@callback prepare_messages(messages :: [Message.t()], context :: term) :: [Message.t()]
```

### Source
`lib/broadway.ex:947`

### Behavior
- Messages failed here with `Broadway.Message.failed/2` still reach `handle_message/3`.

```

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

### prepare_for_draining/1

Source: https://github.com/elixir-broadway/broadway/blob/main/_autodocs/api-reference/broadway-producer.md

Invoked by the terminator right before Broadway starts draining in-flight messages during shutdown, once per producer stage. Producers that must ask the data provider to stop sending should implement this.

```APIDOC
## [CALLBACK] prepare_for_draining/1

### Description
Invoked by the terminator right before Broadway starts draining in-flight messages during shutdown — once per producer stage. Producers that must ask the data provider to stop sending (e.g. RabbitMQ prefetch) should implement this.

### Callback Signature
```elixir
@callback prepare_for_draining(state :: any) ::
            {:noreply, [event], new_state}
            | {:noreply, [event], new_state, :hibernate}
            | {:stop, reason :: term, new_state}
          when new_state: term, event: term
```

### Parameters
- **state** (any) - Required - The current producer state.

### Returns
- **{:noreply, [event], new_state}** (tuple) - Standard GenStage return, forwarded via `handle_no_reply/2`.
- **{:noreply, [event], new_state, :hibernate}** (tuple) - Standard GenStage return with hibernation, forwarded via `handle_no_reply/2`.
- **{:stop, reason, new_state}** (tuple) - Standard GenStage return to stop the producer, forwarded via `handle_no_reply/2`.
```

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

### Create service account key

Source: https://github.com/elixir-broadway/broadway/blob/main/guides/examples/google-cloud-pubsub.md

Generates a credentials.json file for the service account. The IAM account pattern is <account>@<project>.iam.gserviceaccount.com.

```bash
$ gcloud iam service-accounts keys create credentials.json --iam-account=test-account@test-pubsub.iam.gserviceaccount.com
created key [xxx] of type [json] as [key] for [test-account@test-pubsub.iam.gserviceaccount.com]
```