Skip to main content

RabbitMQ pub/sub

Type: pubsub.rabbitmq

Status: stable

Reference: https://docs.dapr.io/reference/components-reference/supported-pubsub/setup-rabbitmq/

Example​

apiVersion: cra.diagrid.io/v1beta1
kind: Component
metadata:
  name: <name>
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
    # The RabbitMQ host address.
    - name: connectionString
      value: "amqp://[username][:password]@host.domain[:port] amqps://[username][:password]@host.domain[:port]"
    # Whether or not the queue consumer should auto-ack messages. (Optional)
    #- name: autoAck
    #  value: "false"
    # The client/connection name. (Optional)
    #- name: clientName
    #  value: "my_client_name"
    # Allows processing multiple messages in parallel (limited by the app-max-concurrency annotation, if configured). Set to single to disable parallel processing. In most situations there's no reason to change this. (Optional)
    #- name: concurrency
    #  value: "parallel"
    # Set the consumer ID to control namespacing. Defaults to the app's ID. (Optional)
    #- name: consumerID
    #  value: "{namespace}"
    # Whether or not the queue should be configured to auto-delete. (Optional)
    #- name: deletedWhenUnused
    #  value: "true"
    # Persistence mode when publishing messages. RabbitMQ treats "2" as persistent, all other numbers as non-persistent. (Optional)
    #- name: deliveryMode
    #  value: "0"
    # Whether or not to use durable queues. (Optional)
    #- name: durable
    #  value: "false"
    # Enable forwarding messages that cannot be handled to a dead-letter topic. (Optional)
    #- name: enableDeadLetter
    #  value: "false"
    # How the component obtains the exchange for a topic. "declare" issues an active AMQP exchange.declare, creating the exchange if it does not exist. "passive" only asserts that the exchange already exists and never creates or modifies it, which lets an external owner - such as the RabbitMQ Cluster Kubernetes Topology Operator or Terraform - be the single source of truth for the topology. This applies to the topic exchange only: the dead-letter exchange is named after the consumer at runtime, so it follows "queueDeclareMode" instead. (Optional)
    #- name: exchangeDeclareMode
    #  value: "declare"
    # Exchange kind of the rabbitmq exchange. "x-consistent-hash" requires the rabbitmq_consistent_hash_exchange plugin to be enabled on the broker; with it, the "routingKey" publish metadata is the partition key and the "routingKey" subscription metadata is the queue's single bucket weight. When "exchangeDeclareMode" is "passive" any exchange kind the broker supports is accepted, because the component does not declare the exchange and RabbitMQ ignores the kind on a passive declare. (Optional)
    #- name: exchangeKind
    #  value: "fanout"
    # The heartbeat used for the connection. (Optional)
    #- name: heartBeat
    #  value: "10s"
    # The maximum number of messages of a queue and its dead letter queue (if dead letter enabled by `enableDeadLetter`). If both `maxLen` and `maxLenBytes` are set then both will apply; whichever limit is hit first will be enforced. (Optional)
    #- name: maxLen
    #  value: "1000"
    # Maximum length in bytes of a queue and its dead letter queue (if dead letter enabled by `enableDeadLetter`). If both `maxLen` and `maxLenBytes` are set then both will apply; whichever limit is hit first will be enforced. (Optional)
    #- name: maxLenBytes
    #  value: "1048576"
    # Number of messages to prefetch. Consider changing this to a non-zero value for production environments. The value of "0" means that all available messages will be pre-fetched. (Optional)
    #- name: prefetchCount
    #  value: "0"
    # Whether to publish AMQP message properties (headers, message ID, etc.) to the metadata. (Optional)
    #- name: publishMessagePropertiesToMetadata
    #  value: "false"
    # If enabled, client waits for publisher confirmation after publishing a message. (Optional)
    #- name: publisherConfirm
    #  value: "false"
    # How the component obtains the queue for a subscription. "declare" creates the queue, its dead-letter objects when "enableDeadLetter" is true, and binds them to their exchanges. "passive" only asserts that the queue already exists, and leaves its bindings and its dead-lettering to the external owner; queue arguments such as "maxLen", "maxPriority" and "queueType", and "enableDeadLetter" itself, are then the owner's responsibility and are ignored by the component. Independent of "exchangeDeclareMode": an operator-managed exchange is commonly paired with queues the component still declares, because consumer queue names are only known at runtime. (Optional)
    #- name: queueDeclareMode
    #  value: "declare"
    # Reconnect wait in Seconds. (Optional)
    #- name: reconnectWaitSeconds
    #  value: "3s"
    # Whether or not to requeue when sending a negative acknowledgement in case of a failure. (Optional)
    #- name: requeueInFailure
    #  value: "false"
    # Default Queue TTL. (Optional)
    #- name: ttlInSeconds
    #  value: "10"

Authentication profiles​

Available authentication profiles:

  • Connection string

  • Authenticate Using RabbitMQ properties

  • Authenticate Using Connection string and External Sasl and TLS

Connection string​

Authenticate using a connection string.

connectionString​

Required - The RabbitMQ host address.

Example value: amqp://[username][:password]@host.domain[:port] amqps://[username][:password]@host.domain[:port]

Authenticate Using RabbitMQ properties​

RabbitMQ authentication by providing hostname, username and password.

hostname (string)​

The RabbitMQ hostname.

Example value: mycompany.com

password (string)​

The RabbitMQ password.

Example value: password

protocol (string)​

The RabbitMQ protocol.

Example value: amqp, amqps

username (string)​

The RabbitMQ username.

Example value: username

Authenticate Using Connection string and External Sasl and TLS​

Authenticate using a connection string along with Exteranl Sasl and TLS.

caCert (string)​

Required - Certificate authority certificate, required for using TLS.

Example value:

-----BEGIN CERTIFICATE-----
<base64-encoded DER>
-----END CERTIFICATE-----

clientCert (string)​

Required - Client certificate, Required for using TLS.

Example value:

-----BEGIN CERTIFICATE-----
<base64-encoded DER>
-----END CERTIFICATE-----

clientKey (string)​

Required - Client key, required for using TLS.

Example value:

-----BEGIN RSA PRIVATE KEY-----
<base64-encoded DER>
-----END RSA PRIVATE KEY-----

connectionString​

Required - The RabbitMQ host address.

Example value: amqp://[username][:password]@host.domain[:port] amqps://[username][:password]@host.domain[:port]

saslExternal (bool)​

With TLS, should the username be taken from an additional field (e.g. CN.) See RabbitMQ Authentication Mechanisms.

Default value: false

Example value: true

RabbitMQ Authentication Mechanisms: https://www.rabbitmq.com/access-control.html#mechanisms

Metadata​

autoAck (bool)​

Whether or not the queue consumer should auto-ack messages.

Default value: false

Example value: true, false

clientName (string)​

The client/connection name.

Example value: my_client_name

concurrency (string)​

Allows processing multiple messages in parallel (limited by the app-max-concurrency annotation, if configured). Set to single to disable parallel processing. In most situations there's no reason to change this.

Default value: parallel

Example value: parallel, single

Allowed values:

  • parallel

  • single

consumerID (string)​

Set the consumer ID to control namespacing. Defaults to the app's ID.

Example value: {namespace}

Documentation: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-overview/#consumer-groups-and-competing-consumers-pattern

deletedWhenUnused (bool)​

Whether or not the queue should be configured to auto-delete.

Default value: true

Example value: true, false

deliveryMode (number)​

Persistence mode when publishing messages. RabbitMQ treats "2" as persistent, all other numbers as non-persistent.

Default value: 0

Example value: 2

durable (bool)​

Whether or not to use durable queues.

Default value: false

Example value: true, false

enableDeadLetter (bool)​

Enable forwarding messages that cannot be handled to a dead-letter topic.

Default value: false

Example value: true, false

exchangeDeclareMode (string)​

How the component obtains the exchange for a topic. "declare" issues an active AMQP exchange.declare, creating the exchange if it does not exist. "passive" only asserts that the exchange already exists and never creates or modifies it, which lets an external owner - such as the RabbitMQ Cluster Kubernetes Topology Operator or Terraform - be the single source of truth for the topology. This applies to the topic exchange only: the dead-letter exchange is named after the consumer at runtime, so it follows "queueDeclareMode" instead.

Default value: declare

Example value: declare,passive

Allowed values:

  • declare

  • passive

exchangeKind (string)​

Exchange kind of the rabbitmq exchange. "x-consistent-hash" requires the rabbitmq_consistent_hash_exchange plugin to be enabled on the broker; with it, the "routingKey" publish metadata is the partition key and the "routingKey" subscription metadata is the queue's single bucket weight. When "exchangeDeclareMode" is "passive" any exchange kind the broker supports is accepted, because the component does not declare the exchange and RabbitMQ ignores the kind on a passive declare.

Default value: fanout

Example value: fanout,topic,direct,headers,x-consistent-hash

Allowed values:

  • fanout

  • topic

  • direct

  • headers

  • x-consistent-hash

heartBeat (duration)​

The heartbeat used for the connection.

Default value: 10s

Example value: 30s

maxLen (number)​

The maximum number of messages of a queue and its dead letter queue (if dead letter enabled by enableDeadLetter). If both maxLen and maxLenBytes are set then both will apply; whichever limit is hit first will be enforced.

Example value: 1000

maxLenBytes (number)​

Maximum length in bytes of a queue and its dead letter queue (if dead letter enabled by enableDeadLetter). If both maxLen and maxLenBytes are set then both will apply; whichever limit is hit first will be enforced.

Example value: 1048576

prefetchCount (number)​

Number of messages to prefetch. Consider changing this to a non-zero value for production environments. The value of "0" means that all available messages will be pre-fetched.

Default value: 0

Example value: 2

publishMessagePropertiesToMetadata (bool)​

Whether to publish AMQP message properties (headers, message ID, etc.) to the metadata.

Default value: false

Example value: true, false

publisherConfirm (bool)​

If enabled, client waits for publisher confirmation after publishing a message.

Default value: false

Example value: true, false

queueDeclareMode (string)​

How the component obtains the queue for a subscription. "declare" creates the queue, its dead-letter objects when "enableDeadLetter" is true, and binds them to their exchanges. "passive" only asserts that the queue already exists, and leaves its bindings and its dead-lettering to the external owner; queue arguments such as "maxLen", "maxPriority" and "queueType", and "enableDeadLetter" itself, are then the owner's responsibility and are ignored by the component. Independent of "exchangeDeclareMode": an operator-managed exchange is commonly paired with queues the component still declares, because consumer queue names are only known at runtime.

Default value: declare

Example value: declare,passive

Allowed values:

  • declare

  • passive

reconnectWaitSeconds (duration)​

Reconnect wait in Seconds.

Default value: 3s

Example value: 10s

requeueInFailure (bool)​

Whether or not to requeue when sending a negative acknowledgement in case of a failure.

Default value: false

Example value: true, false

ttlInSeconds (duration)​

Default Queue TTL.

Example value: 10

Last updated on