Broadway RabbitMQ

https://hexdocs.pm/broadway_rabbitmq
Admin
A RabbitMQ producer for Broadway that automatically acknowledges/rejects messages and handles...
Tokens:5,735
Snippets:77
Trust Score:9
Update:3 months ago
Show doc for...
Tokens:
Raw
### Custom Channel Pool Implementation

Source: https://hexdocs.pm/broadway_rabbitmq/BroadwayRabbitMQ.ChannelPool.html

Example implementation of a custom channel pool using the `BroadwayRabbitMQ.ChannelPool` behavior. This module defines `checkout_channel/1` and `checkin_channel/2` callbacks.

```elixir
defmodule MyPool do
  @behaviour BroadwayRabbitMQ.ChannelPool

  @impl true
  def checkout_channel(name) do
    conn = %AMQP.Connection{pid: Process.whereis(name)}

    case AMQP.Channel.open(conn) do
      {:ok, channel} -> {:ok, channel}
      {:error, reason} -> {:error, %RuntimeError{message: inspect(reason)}}
    end
  end

  @impl true
  def checkin_channel(_name, channel) do
    case AMQP.Channel.close(channel) do
      :ok -> :ok
      {:error, reason} -> {:error, %RuntimeError{message: inspect(reason)}}
    end
  end
end
```

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

### Start Broadway RabbitMQ Pipeline

Source: https://hexdocs.pm/broadway_rabbitmq/index.html

Configure the Broadway pipeline with RabbitMQ producer settings, including queue name, connection details, and failure handling. Ensure the AMQP connection is configured directly within the Broadway pipeline as global AMQP connections are not supported.

```elixir
@processor_concurrency 50
@max_demand 2

Broadway.start_link(MyBroadway,
  name: MyBroadway,
  producer: [
    module:
      {BroadwayRabbitMQ.Producer,
       queue: "my_queue",
       connection: [
         username: "user",
         password: "password",
         host: "192.168.0.10"
       ],
       qos: [
         # See "Back-pressure and `:prefetch_count`" section
         prefetch_count: @processor_concurrency * @max_demand
       ],
       on_failure: :reject_and_requeue},
    # See "Producer concurrency" section
    concurrency: 1
  ],
  processors: [
    default: [
      concurrency: @processor_concurrency,
      # See "Max demand" section
      max_demand: @max_demand
    ]
  ]
)

```

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

### Declare Queues and Bind to Exchanges

Source: https://hexdocs.pm/broadway_rabbitmq/index.html

Starts a Broadway producer with a RabbitMQ connection, specifying queue name, declaration, and bindings to exchanges. This is typically done when setting up the pipeline.

```elixir
Broadway.start_link(MyBroadway,
  name: MyBroadway,
  producer: [
    module:
      {BroadwayRabbitMQ.Producer,
      queue: "my_queue",
      declare: [],
      bindings: [{"my-exchange", []}]},
    concurrency: 1
  ],
  processors: [
    default: []
  ]
)
```

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

### Producer Connection with Custom Pool

Source: https://hexdocs.pm/broadway_rabbitmq/BroadwayRabbitMQ.ChannelPool.html

Pass `{:custom_pool, module, args}` as the `:connection` option to `BroadwayRabbitMQ.Producer` to use a custom channel pool. `module` must implement the `BroadwayRabbitMQ.ChannelPool` behavior.

```elixir
connection: {:custom_pool, MyPool, _amqp_connection = :big_pool}
```

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

### Configure Broadway RabbitMQ Producer

Source: https://hexdocs.pm/broadway_rabbitmq/BroadwayRabbitMQ.Producer.html

Configure the Broadway RabbitMQ producer with connection details, queue name, quality of service (prefetch count), and failure handling. Set producer concurrency to 1 for reduced internal queueing, and processor concurrency and max demand as needed.

```elixir
@processor_concurrency 50
@max_demand 2

Broadway.start_link(MyBroadway,
  name: MyBroadway,
  producer: [
    module:
      {BroadwayRabbitMQ.Producer,
       queue: "my_queue",
       connection: [
         username: "user",
         password: "password",
         host: "192.168.0.10"
       ],
       qos: [
         # See "Back-pressure and `:prefetch_count`" section
         prefetch_count: @processor_concurrency * @max_demand
       ],
       on_failure: :reject_and_requeue},
    # See "Producer concurrency" section
    concurrency: 1
  ],
  processors: [
    default: [
      concurrency: @processor_concurrency,
      # See "Max demand" section
      max_demand: @max_demand
    ]
  ]
)
```

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

### Configure Dead-letter Exchange in Broadway RabbitMQ

Source: https://hexdocs.pm/broadway_rabbitmq/index.html

This Elixir code defines a Broadway pipeline that uses a dead-letter exchange. Messages that fail processing in `handle_message` are automatically routed to the configured dead-letter exchange. Ensure the `declare_rabbitmq_topology` function is called to set up the necessary exchanges and queues.

```elixir
defmodule MyPipeline do
  use Broadway

  @queue "my_queue"
  @exchange "my_exchange"
  @queue_dlx "my_queue.dlx"
  @exchange_dlx "my_exchange.dlx"

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          BroadwayRabbitMQ.Producer,
          on_failure: :reject,
          after_connect: &declare_rabbitmq_topology/1,
          queue: @queue,
          declare: [
            durable: true,
            arguments: [
              {"x-dead-letter-exchange", :longstr, @exchange_dlx},
              {"x-dead-letter-routing-key", :longstr, @queue_dlx}
            ]
          ],
          bindings: [{@exchange, []}],
        },
        concurrency: 2
      ],
      processors: [default: [concurrency: 4]]
    )
  end

  defp declare_rabbitmq_topology(amqp_channel) do
    with :ok <- AMQP.Exchange.declare(amqp_channel, @exchange, :fanout, durable: true),
         :ok <- AMQP.Exchange.declare(amqp_channel, @exchange_dlx, :fanout, durable: true),
         {:ok, _} <- AMQP.Queue.declare(amqp_channel, @queue_dlx, durable: true),
         :ok <- AMQP.Queue.bind(amqp_channel, @queue_dlx, @exchange_dlx) do
      :ok
    end
  end

  @impl true
  def handle_message(_processor, message, _context) do
    # Raising errors or returning a "failed" message here sends the message to the
    # dead-letter queue.
  end
end
```

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

### Ack Message Immediately Before Processing

Source: https://hexdocs.pm/broadway_rabbitmq/index.html

Acknowledges a message immediately before processing it, ensuring it's not redelivered if the consumer loses connection. This is useful for 'at most once' processing semantics.

```elixir
def handle_message(_, message, _context) do
  Broadway.Message.ack_immediately(message)
  process_message(message)
  message
end
```
Tokens:
Raw