RabbitMQ

Synopsis

Creates a consumer that connects to RabbitMQ servers and consumes messages from specified exchanges and queues. Supports multiple authentication methods, exchange types, TLS encryption, and multiple workers with automatic message acknowledgment.

For details, see Appendix.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: rabbitmq
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <numeric>
    username: <string>
    password: <string>
    authentication: <string>
    exchange:
      name: <string>
      type: <string>
    queue:
      name: <string>
      key: <string>
    reuse: <boolean>
    workers: <numeric>
    tls:
      status: <boolean>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
      ca_name: <string>
      server_name: <string>
      min_tls_version: <string>
      max_tls_version: <string>
      insecure_skip_verify: <boolean>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
id	Y	-	Unique numeric identifier
name	Y	-	Device name
description	N	-	Optional description
type	Y	-	Must be rabbitmq
tags	N	-	Optional tags
pipelines	N	-	Optional pre-processor pipelines
status	N	true	Enable/disable the device

Connection

Field	Required	Default	Description
address	N	"0.0.0.0"	RabbitMQ server address
port	Y	-	RabbitMQ server port
username	N	-	Authentication username
password	N	-	Authentication password
authentication	Y	-	Auth type (plain, amqplain)

Exchange

Field	Required	Default	Description
exchange.name	Y	-	Exchange name
exchange.type	Y	-	Exchange type (direct, fanout, topic, x-custom)

Supported exchange types:

Exchange Type	Description
direct	Exact routing key match
fanout	Broadcast to all bound queues
topic	Pattern-based routing using wildcards (*, #)
x-custom	Custom exchange plugin registered on the broker

Queue

Field	Required	Default	Description
queue.name	Y	-	Queue name
queue.key	Y	-	Routing key pattern

TLS

Field	Required	Default	Description
tls.status	N	false	Enable TLS encryption
tls.cert_name	Y*	cert.pem	Client certificate for mutual TLS
tls.key_name	Y*	key.pem	Client private key for mutual TLS
tls.passphrase	N	-	Passphrase for an encrypted private key
tls.ca_name	N	-	CA bundle used to verify the broker certificate. When unset, the host trust store is used.
tls.server_name	N	-	SNI hostname override for the TLS handshake
tls.min_tls_version	N	tls1.2	Minimum accepted TLS version (tls1.0, tls1.1, tls1.2, tls1.3)
tls.max_tls_version	N	-	Maximum accepted TLS version. When unset, the highest mutually supported version is negotiated.
tls.insecure_skip_verify	N	true	Skip broker certificate verification

* = Required when tls.status is true.

note

TLS material fields (cert_name, key_name, ca_name, client_ca_name) accept any of the following:

• File name — resolved relative to the service root directory. Nested paths such as certs/prod/server.pem are supported.
• Absolute path — honored only if it resolves inside the service root. Any path that escapes the root is refused.
• Inline PEM content — used verbatim when the value contains -----BEGIN.
• Environment variable — ${ENV_VAR}.
• Vault reference — $secret{id=...} or $secret{store=...,ref=...}.

warning

This device defaults tls.insecure_skip_verify to true, unlike every other message queue device. With tls.status: true and nothing further set, the connection is encrypted but the broker certificate is not verified. Set tls.insecure_skip_verify: false to verify it, and supply tls.ca_name when the broker presents a private CA.

Performance

Field	Required	Default	Description
reuse	N	true	Enable multi-worker mode
workers	N	4	Number of worker processes when reuse enabled (capped at the number of available CPU cores)

Examples

Basic

Creating a simple RabbitMQ consumer with a direct exchange...

- id: 1 name: basic_rabbitmq type: rabbitmq properties: address: "rabbitmq.example.com" port: 5672 authentication: "plain" exchange: name: "logs" type: "direct" queue: name: "app_logs" key: "app.*"

Secure

Connecting with authentication and TLS encryption...

- id: 2 name: secure_rabbitmq type: rabbitmq properties: address: "rabbitmq.example.com" port: 5671 username: "consumer" password: "secret" authentication: "plain" exchange: name: "secure_logs" type: "topic" queue: name: "secure_app_logs" key: "secure.app.#" tls: status: true cert_name: "rabbitmq.crt" key_name: "rabbitmq.key"

High-Volume

Optimizing for high throughput with multi-worker mode...

- id: 3 name: performant_rabbitmq type: rabbitmq properties: address: "rabbitmq.example.com" port: 5672 authentication: "plain" exchange: name: "high_volume" type: "direct" queue: name: "high_volume_logs" key: "logs" reuse: true workers: 4

Topic Exchange

Pattern-based routing on a topic exchange...

- id: 4 name: topic_rabbitmq type: rabbitmq properties: address: "rabbitmq.example.com" port: 5672 authentication: "plain" exchange: name: "logs" type: "topic" queue: name: "filtered_logs" key: "app.*.error"

tip

Topic routing keys support * and # wildcards for single and multiple words respectively.

Pipelines

Applying custom processing to messages...

- id: 5 name: pipeline_rabbitmq type: rabbitmq pipelines: - json_parser - field_extractor properties: address: "rabbitmq.example.com" port: 5672 authentication: "plain" exchange: name: "raw_logs" type: "direct" queue: name: "processed_logs" key: "logs"

note

Pipelines are processed sequentially, and can modify or drop messages before ingestion.