Rabbitmq Endpoint
Rabbitmq Endpoint v0.24.0+ Used to create and start the Rabbitmq subscription service, which can subscribe to data from different stream names and then route it to different rule chains for processing.
TIP
- This component is an extension component and requires the introduction of additional extension libraries: rulego-components (opens new window)
# Type
endpoint/rabbitmq
# Startup Configuration
This component allows the reuse of shared connection clients through the server field. See Component Connection Reuse for reference.
| Field | Type | Description | Default |
|---|---|---|---|
| server | string | RabbitMQ server address, example: amqp://guest:guest@127.0.0.1:5672/ | None |
| exchange | string | Exchange name | None |
| exchangeType | string | Exchange type, available values: direct, fanout, topic | None |
| durable | bool | Indicates whether the exchange and queue are durable. If set to true, they will be preserved even if the message server is restarted. | false |
| autoDelete | bool | Indicates whether the queue is automatically deleted. If set to true, the queue will be deleted when the last consumer disconnects. | false |
| queueName | string | Custom queue name. Leave empty to use an auto-generated exclusive temp queue deleted on connection close (legacy behavior). With a fixed name plus durable=true and autoDelete=false, messages accumulate in the queue while the consumer is offline or the broker restarts, and are consumed after the consumer comes back online | None (anonymous queue) |
| exclusive | bool | Whether the queue is restricted to this connection and deleted when it closes. Forced true when queueName is empty; defaults false for a named queue | false |
TIP
To keep undelivered messages across broker restarts and consumer downtime, use a fixed queueName with durable=true and autoDelete=false, and make sure the publisher sends persistent messages (delivery mode 2, e.g. enable persistent on the x/rabbitmqClient node). A queue must be declared with identical arguments every time, otherwise the declaration fails.
# Response
Before responding with exchange.Out.SetBody, you need to specify the responseTopic parameter through exchange.Out.Headers() or exchange.Out.Msg.Metadata, and the component will send data to the specified topic:
exchange.Out.GetMsg().Metadata.PutValue("responseTopic", "device.msg.response")
// or
exchange.Out.Headers().Add("responseTopic", "device.msg.response")
exchange.Out.SetBody([]byte("ok"))
2
3
4
5
Response parameter configuration:
| Field | Type | Required | Description | Default |
|---|---|---|---|---|
| responseTopic/responseKey | string | Yes | Response Key | - |
| responseExchange | string | Yes | Response Exchange | - |
# Multi-replica Deployment
This endpoint declares a private queue per replica bound to the same exchange, so in a multi-replica deployment every message triggers every replica. Once the rule chain config injects a distributed lock, this endpoint elects a single active consumer automatically — only the leader declares the queue and consumes, and a standby takes over when the leader is lost. Single-process deployments are unaffected.
# Example
Below is an example code using the endpoint: