# RabbitMQ

Strictly speaking, AMQP (and 0.9.1 version used) is a protocol, not a full-fledged driver, so you can use any servers that support this protocol (on your own, only rabbitmq was tested), such as:

* [RabbitMQ](https://www.rabbitmq.com/),
* [Apache Qpid](http://qpid.apache.org/)
* [Apache ActiveMQ](http://activemq.apache.org/).

However, it is recommended to use RabbitMQ as the main implementation, and reliable performance with other implementations is not guaranteed.

To install and configure the RabbitMQ, use the corresponding [documentation page](https://www.rabbitmq.com/download.html).

{% hint style="info" %}
Every message pushed to the RabbitMQ server uses publisher confirms. A message is considered sent only after the server confirms it. This is a reliable way to ensure delivery to the server.
{% endhint %}

After that, you should configure the connection to the server in the `amqp` section. This configuration section contains exactly one `addr` key with a [connection DSN](https://www.rabbitmq.com/uri-spec.html). The `TLS` configuration sits in the `amqp.tls` section and consists of the following options:

* `key`: path to a key file.
* `cert`: path to a certificate file.
* `root_ca`: path to Root CAs used by the AMQP client to trust and verify the broker/server certificate during TLS dial.
* `client_auth_type`: also known as `mTLS`. Possible values are: `no_client_cert`, `request_client_cert`, `require_any_client_cert`, `verify_client_cert_if_given`, `require_and_verify_client_cert`.

You should also configure `rabbitMQ` with `TLS` support: [link](https://www.rabbitmq.com/ssl.html).

{% code title=".rr.yaml" %}

```yaml
amqp:
  addr: amqp://guest:guest@127.0.0.1:5672

  # AMQPS TLS configuration
  #
  # This section is optional
  tls:
    # Path to the key file
    #
    # This option is required
    key: ""

    # Path to the certificate
    #
    # This option is required
    cert: ""

    # Path to Root CAs used by the AMQP client to trust and verify the broker/server certificate during TLS dial.
    #
    # This option is optional
    root_ca: ""

    # Client auth type (mTLS, peer verification).
    #
    # This option is optional. Default value: no_client_cert. Possible values: no_client_cert, request_client_cert, require_any_client_cert, verify_client_cert_if_given, require_and_verify_client_cert
    client_auth_type: no_client_cert
```

{% endcode %}

Upon establishing a connection to the server, you can create a new queue that utilizes this connection and encompasses the queue settings, including those specific to AMQP.

## Configuration

{% code title=".rr.yaml" %}

```yaml
version: "3"

amqp:
  addr: amqp://guest:guest@127.0.0.1:5672

  # AMQPS TLS configuration
  #
  # This section is optional
  tls:
    # Path to the key file
    #
    # This option is required
    key: ""

    # Path to the certificate
    #
    # This option is required
    cert: ""

    # Path to Root CAs used by the AMQP client to trust and verify the broker/server certificate during TLS dial.
    #
    # This option is optional
    root_ca: ""

    # Client auth type (mTLS, peer verification).
    #
    # This option is optional. Default value: no_client_cert. Possible values: no_client_cert, request_client_cert, require_any_client_cert, verify_client_cert_if_given, require_and_verify_client_cert
    client_auth_type: no_client_cert

jobs:
  pipelines:
    # User defined name of the queue.
    example:
      # Driver name
      #
      # This option is required.
      driver: amqp

      # Driver's configuration
      #
      # Should not be empty
      config:

        # QoS - prefetch.
        #
        # Default: 10
        prefetch: 10

        # Pipeline priority
        #
        # If the job has priority set to 0, it will inherit the pipeline's priority. Default: 10.
        priority: 1

        # Redial timeout (in seconds). How long to try to reconnect to the AMQP server.
        #
        # Default: 60
        redial_timeout: 60

        # Durable queue
        #
        # Default: false
        durable: false

        # Durable exchange (rabbitmq option: https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchanges)
        #
        # Default: false
        exchange_durable: false

        # Auto-delete (exchange is deleted when last queue is unbound from it): https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchanges
        #
        # Default: false
        exchange_auto_delete: false

        # Auto-delete (queue that has had at least one consumer is deleted when last consumer unsubscribes) (rabbitmq option: https://www.rabbitmq.com/queues.html#properties)
        #
        # Default: false
        queue_auto_delete: false

        # Delete queue when stopping the pipeline
        #
        # Default: false
        delete_queue_on_stop: false

        # Queue name
        #
        # Optional for producer-only pipelines. Required for run/resume/pause.
        # Can be omitted for push-only mode.
        queue: test-1-queue

        # Exchange name
        #
        # Optional. Default: amqp.default
        exchange: amqp.default

        # Exchange type
        #
        # Default: direct. Possible values: direct, fanout, topic, headers.
        exchange_type: direct

        # Routing key for the queue
        #
        # Default: empty. Required for push when exchange_type != fanout.
        routing_key: test

        # Declare a queue exclusive at the exchange
        #
        # Default: false
        exclusive: false

        # When multiple is true, this delivery and all prior unacknowledged deliveries
        # on the same channel will be acknowledged.  This is useful for batch processing
        # of deliveries
        #
        # Default: false
        multiple_ack: false

        # The consumer_id is identified by a string that is unique and scoped for all consumers on this channel.
        #
        # Default: "roadrunner" + uuid.
        consumer_id: "roadrunner-uuid"

        # Use rabbitmq mechanism to requeue the job on fail
        #
        # Default: false
        requeue_on_fail: false

        # Queue headers (new in 2.12.2)
        #
        # Default: null
        queue_headers:
          x-queue-mode: lazy
```

{% endcode %}

## Configuration options

**Here is a detailed description of each of the amqp-specific options:**

### Priority

`priority`- job priority. A lower value corresponds to a higher priority. For instance, consider two pipelines: `pipe1` with a priority of `1` and `pipe10` with a priority of `10`. Workers will only take jobs from `pipe10` if all the jobs from `pipe1` have been processed.

### Prefetch

`prefetch` - rabbitMQ QoS prefetch. See also ["prefetch-size"](https://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.qos.prefetch-size). Note that if you use a large number of workers and a small `prefetch` number, some of the workers may not be loaded with messages (jobs) due to the blocking nature of the prefetch. This would result in poor RoadRunner performance and waste of resources.

### Queue

`queue` - AMQP internal (inside the driver) queue name. Optional for producer-only pipelines, required for consumer lifecycle operations (`run`, `resume`, `pause`).

### Exchange

`exchange` - rabbitMQ exchange name. Optional, default: `amqp.default`.

{% hint style="info" %}
See also [AMQP model](https://www.rabbitmq.com/tutorials/amqp-concepts.html#amqp-model) documentation section.
{% endhint %}

{% hint style="info" %}
Producer-only pipeline: `queue` can be empty, `push` works, but `run`, `resume`, and `pause` will fail without a queue.
{% endhint %}

### Exchange type

`exchange_type` - rabbitMQ exchange type. May be one of `direct`, `fanout`, `topic`, `headers`.

### Routing key

`routing_key` - queue's routing key. Required for `push` when `exchange_type != fanout`.

### Exclusive

`exclusive` - applied to the queue, exclusive queues cannot be redeclared. If set to true, and you attempt to declare the same pipeline twice, it will result in an error.

### Multiple ack

`multiple_ack` - this delivery, along with all prior unacknowledged deliveries on the same channel, will be acknowledged. This feature is beneficial for batch processing of deliveries and is applicable only for `Ack`, not for `Nack`.

### Requeue on fail

`requeue_on_fail` - requeue on Nack (by RabbitMQ).

{% hint style="info" %}
Read more about Nack in RabbitMQ official docs: <https://www.rabbitmq.com/confirms.html#consumer-nacks-requeue>
{% endhint %}

### Queue headers

`queue_headers` - used to pass arguments to the `Queue` create method, such as `x-queue-mode: lazy`

### Durable

`durable` - create a durable queue.

Default: `false`

### Delete queue on stop

`delete_queue_on_stop` - delete the queue when the pipeline is stopped.

Default: `false`

### Redial timeout

`redial_timeout` - Redial timeout (in seconds). How long to try to reconnect to the AMQP server.

### Exchange durable

`exchange_durable` - Durable exchange ([rabbitmq option](https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchanges)).

Default: `false`

### Exchange auto delete

`exchange_auto_delete` - Auto-delete (exchange is deleted when last queue is unbound from it): [link](https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchanges).

Default: `false`

### Queue auto delete

`queue_auto_delete` - Auto-delete (queue that has had at least one consumer is deleted when last consumer unsubscribes): [link](https://www.rabbitmq.com/queues.html#properties).

Default: `false`

### Consumer id

`consumer_id` - string that is unique and scoped for all consumers on this channel.