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.

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>

Configuration

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*	-	TLS certificate file name
tls.key_name	Y*	-	TLS private key file name

* = Conditionally required when tls.status is true.

note

The TLS certificate and key files must be placed in the service root directory.

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.