= amqp_0_9
:type: output
:status: stable
:categories: ["Services"]



////
     THIS FILE IS AUTOGENERATED!

     To make changes, edit the corresponding source file under:

     https://github.com/redpanda-data/connect/tree/main/internal/impl/<provider>.

     And:

     https://github.com/redpanda-data/connect/tree/main/cmd/tools/docs_gen/templates/plugin.adoc.tmpl
////

// © 2024 Redpanda Data Inc.


component_type_dropdown::[]


Sends messages to an AMQP (0.91) exchange. AMQP is a messaging protocol used by various message brokers, including RabbitMQ.Connects to an AMQP (0.91) queue. AMQP is a messaging protocol used by various message brokers, including RabbitMQ.


[tabs]
======
Common::
+
--

```yml
# Common config fields, showing default values
output:
  label: ""
  amqp_0_9:
    urls: [] # No default (required)
    exchange: "" # No default (required)
    key: ""
    type: ""
    metadata:
      exclude_prefixes: []
    max_in_flight: 64
```

--
Advanced::
+
--

```yml
# All config fields, showing default values
output:
  label: ""
  amqp_0_9:
    urls: [] # No default (required)
    exchange: "" # No default (required)
    exchange_declare:
      enabled: false
      type: direct
      durable: true
    key: ""
    type: ""
    content_type: application/octet-stream
    content_encoding: ""
    correlation_id: ""
    reply_to: ""
    expiration: ""
    message_id: ""
    user_id: ""
    app_id: ""
    metadata:
      exclude_prefixes: []
    priority: ""
    max_in_flight: 64
    persistent: false
    mandatory: false
    immediate: false
    timeout: ""
    tls:
      enabled: false
      skip_cert_verify: false
      enable_renegotiation: false
      root_cas: ""
      root_cas_file: ""
      client_certs: []
```

--
======

The metadata from each message are delivered as headers.

It's possible for this output type to create the target exchange by setting `exchange_declare.enabled` to `true`, if the exchange already exists then the declaration passively verifies that the settings match.

TLS is automatic when connecting to an `amqps` URL, but custom settings can be enabled in the `tls` section.

The fields 'key', 'exchange' and 'type' can be dynamically set using xref:configuration:interpolation.adoc#bloblang-queries[function interpolations].

== Fields

=== `urls`

A list of URLs to connect to. The first URL to successfully establish a connection will be used until the connection is closed. If an item of the list contains commas it will be expanded into multiple URLs.


*Type*: `array`

Requires version 3.58.0 or newer

```yml
# Examples

urls:
  - amqp://guest:guest@127.0.0.1:5672/

urls:
  - amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/

urls:
  - amqp://127.0.0.1:5672/
  - amqp://127.0.0.2:5672/
```

=== `exchange`

An AMQP exchange to publish to.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`


=== `exchange_declare`

Optionally declare the target exchange (passive).


*Type*: `object`


=== `exchange_declare.enabled`

Whether to declare the exchange.


*Type*: `bool`

*Default*: `false`

=== `exchange_declare.type`

The type of the exchange.


*Type*: `string`

*Default*: `"direct"`

Options:
`direct`
, `fanout`
, `topic`
, `x-custom`
.

=== `exchange_declare.durable`

Whether the exchange should be durable.


*Type*: `bool`

*Default*: `true`

=== `key`

The binding key to set for each message.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `type`

The type property to set for each message.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `content_type`

The content type attribute to set for each message.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `"application/octet-stream"`

=== `content_encoding`

The content encoding attribute to set for each message.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `correlation_id`

Set the correlation ID of each message with a dynamic interpolated expression.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `reply_to`

Carries response queue name - set with a dynamic interpolated expression.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `expiration`

Set the per-message TTL
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `message_id`

Set the message ID of each message with a dynamic interpolated expression.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `user_id`

Set the user ID to the name of the publisher.  If this property is set by a publisher, its value must be equal to the name of the user used to open the connection.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `app_id`

Set the application ID of each message with a dynamic interpolated expression.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

=== `metadata`

Specify criteria for which metadata values are attached to messages as headers.


*Type*: `object`


=== `metadata.exclude_prefixes`

Provide a list of explicit metadata key prefixes to be excluded when adding metadata to sent messages.


*Type*: `array`

*Default*: `[]`

=== `priority`

Set the priority of each message with a dynamic interpolated expression.
This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].


*Type*: `string`

*Default*: `""`

```yml
# Examples

priority: "0"

priority: ${! meta("amqp_priority") }

priority: ${! json("doc.priority") }
```

=== `max_in_flight`

The maximum number of messages to have in flight at a given time. Increase this to improve throughput.


*Type*: `int`

*Default*: `64`

=== `persistent`

Whether message delivery should be persistent (transient by default).


*Type*: `bool`

*Default*: `false`

=== `mandatory`

Whether to set the mandatory flag on published messages. When set if a published message is routed to zero queues it is returned.


*Type*: `bool`

*Default*: `false`

=== `immediate`

Whether to set the immediate flag on published messages. When set if there are no ready consumers of a queue then the message is dropped instead of waiting.


*Type*: `bool`

*Default*: `false`

=== `timeout`

The maximum period to wait before abandoning it and reattempting. If not set, wait indefinitely.


*Type*: `string`

*Default*: `""`

=== `tls`

Custom TLS settings can be used to override system defaults.


*Type*: `object`


=== `tls.enabled`

Whether custom TLS settings are enabled.


*Type*: `bool`

*Default*: `false`

=== `tls.skip_cert_verify`

Whether to skip server side certificate verification.


*Type*: `bool`

*Default*: `false`

=== `tls.enable_renegotiation`

Whether to allow the remote server to repeatedly request renegotiation. Enable this option if you're seeing the error message `local error: tls: no renegotiation`.


*Type*: `bool`

*Default*: `false`
Requires version 3.45.0 or newer

=== `tls.root_cas`

An optional root certificate authority to use. This is a string, representing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.
[CAUTION]
====
This field contains sensitive information that usually shouldn't be added to a config directly, read our xref:configuration:secrets.adoc[secrets page for more info].
====



*Type*: `string`

*Default*: `""`

```yml
# Examples

root_cas: |-
  -----BEGIN CERTIFICATE-----
  ...
  -----END CERTIFICATE-----
```

=== `tls.root_cas_file`

An optional path of a root certificate authority file to use. This is a file, often with a .pem extension, containing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.


*Type*: `string`

*Default*: `""`

```yml
# Examples

root_cas_file: ./root_cas.pem
```

=== `tls.client_certs`

A list of client certificates to use. For each certificate either the fields `cert` and `key`, or `cert_file` and `key_file` should be specified, but not both.


*Type*: `array`

*Default*: `[]`

```yml
# Examples

client_certs:
  - cert: foo
    key: bar

client_certs:
  - cert_file: ./example.pem
    key_file: ./example.key
```

=== `tls.client_certs[].cert`

A plain text certificate to use.


*Type*: `string`

*Default*: `""`

=== `tls.client_certs[].key`

A plain text certificate key to use.
[CAUTION]
====
This field contains sensitive information that usually shouldn't be added to a config directly, read our xref:configuration:secrets.adoc[secrets page for more info].
====



*Type*: `string`

*Default*: `""`

=== `tls.client_certs[].cert_file`

The path of a certificate to use.


*Type*: `string`

*Default*: `""`

=== `tls.client_certs[].key_file`

The path of a certificate key to use.


*Type*: `string`

*Default*: `""`

=== `tls.client_certs[].password`

A plain text password for when the private key is password encrypted in PKCS#1 or PKCS#8 format. The obsolete `pbeWithMD5AndDES-CBC` algorithm is not supported for the PKCS#8 format.

Because the obsolete pbeWithMD5AndDES-CBC algorithm does not authenticate the ciphertext, it is vulnerable to padding oracle attacks that can let an attacker recover the plaintext.
[CAUTION]
====
This field contains sensitive information that usually shouldn't be added to a config directly, read our xref:configuration:secrets.adoc[secrets page for more info].
====



*Type*: `string`

*Default*: `""`

```yml
# Examples

password: foo

password: ${KEY_PASSWORD}
```


