RabbitMQ

Publish messages to RabbitMQ queues and exchanges

Overview

The RabbitMQ tool connects to an AMQP 0-9-1 broker (RabbitMQ) and publishes messages. It supports two modes: (1) Queue mode sends messages directly to a named queue using channel.sendToQueue(); (2) Exchange mode publishes to a named exchange with a routing key using channel.publish(). Exchange types supported: direct, topic, headers, fanout. Message content can be the entire input JSON (sendInputData=true), a custom string, or binary data from an upstream node. When binaryData is enabled, the tool reads binary content from the specified binary property and sends the raw bytes as the message body. AMQP headers and publish arguments are configurable via fixedCollection parameters. Queue/exchange options include durable, autoDelete, exclusive, assertQueue/assertExchange. Uses the amqplib npm package.

Category: Development
Tool Name: rabbitmq
Version: 1

Appearance: Icon: si-rabbitmq | Color: #FF6600

Node Type

Action (Binary) — handles file/binary data operations

Input / Output

Direction	Port(s)
Input	Input
Output	Output, Error

Credentials

This tool requires RabbitMQ credentials. See the Credentials Guide for setup instructions.

Operations

Operation	Value	Description
Send a Message to RabbitMQ	sendMessage	Publish a message to a queue or exchange
Delete From Queue	deleteMessage	This operation is a pass-through — RabbitMQ ack/nack requires a consumer channel which this tool does not implement.

Parameters

Delete From Queue takes no parameters of its own — see All Operations.

Send a Message to RabbitMQ (sendMessage)

Parameter	Type	Required	Default	Description
Mode	options	No	queue	Whether to send to a queue directly or publish to an exchange.
				Options: queue (publish data to a queue), exchange (publish data to an exchange)
Queue / Topic	string	Yes	—	Name of the queue to publish to. Supports expressions like {{ $json.queue }}. (shown when Mode is queue)
Exchange	string	Yes	—	Name of the exchange to publish to. Supports expressions like {{ $json.exchange }}. (shown when Mode is exchange)
Type	options	No	fanout	Type of exchange. (shown when Mode is exchange)
				Options: direct (exact routing-key match), topic (routing-key patterns), headers (routes on header attributes), fanout (broadcasts to every bound queue)
Routing Key	string	No	—	The routing key for the message. Supports expressions like {{ $json.routingKey }}. (shown when Mode is exchange)
Send Input Data	boolean	No	true	Whether to send the entire input JSON as the message body.
Message	string	No	—	The message body to send. Used when Send Input Data is false and Binary Data is false. Supports expressions like {{ $json.text }}. (shown when Send Input Data is false)
Binary Data	boolean	No	false	Whether to send binary data from the input item as the message content. (shown when Send Input Data is false)
Binary Property	string	No	data	Name of the binary property to read message content from. Names are case-sensitive — see the upstream node’s Binary Data panel for the exact names to use. (shown when Binary Data is true and Send Input Data is false)
Options	collection	No	{}	Additional AMQP queue, exchange and message options.
— Alternate Exchange	string	No	—	An exchange to send messages to if this exchange cannot route them to any queues. (shown when Mode is exchange)
— Arguments	fixedCollection	No	{}	Additional amqplib publish arguments (e.g., expiration, messageId, correlationId).
— — Key	string	No	—	The argument name.
— — Value	string	No	—	The argument value.
— Assert Exchange	boolean	No	true	Whether to assert the exchange exists before publishing. If false, checkExchange is used instead. (shown when Mode is exchange)
— Assert Queue	boolean	No	true	Whether to assert the queue exists before sending. If false, checkQueue is used instead. (shown when Mode is queue)
— Auto Delete Queue	boolean	No	false	Whether the queue will be deleted when the number of consumers drops to zero.
— Durable	boolean	No	true	Whether the queue/exchange will survive broker restarts.
— Exclusive	boolean	No	false	Whether to scope the queue to the connection. (shown when Mode is queue)
— Headers	fixedCollection	No	{}	AMQP headers to include with the message.
— — Key	string	No	—	The header name.
— — Value	string	No	—	The header value.

All Operations

Parameter	Type	Required	Default	Description
Max Concurrency	number	No	10	Maximum number of items to process concurrently.

Output Data

One output item per input item — neither operation fans out. Binary data is forwarded unchanged in both cases.

• sendMessage — the output item JSON replaces the input item JSON with a fixed confirmation object. Upstream fields do not pass through, so capture anything you still need before this node.

{
  "success": true
}

• deleteMessage — a pure pass-through. The input item JSON is emitted unchanged and no broker call is made.

The publish waits for the broker to acknowledge each message before the item is written, so an item on the Output port means the broker accepted it. Messages are published as persistent, and the message’s content type is set from how you supplied the body: application/json for Send Input Data, the binary property’s own MIME type for Binary Data, and text/plain for a custom Message.

Usage Examples

• Send a JSON message to a RabbitMQ queue
• Publish a message to a topic exchange with a routing key
• Send binary file content as a RabbitMQ message
• Publish workflow data to a fanout exchange

Example Configuration

Send each incoming item to a durable work queue:

{
  "type": "rabbitmq",
  "parameters": {
    "operation": "sendMessage",
    "mode": "queue",
    "queue": "work-queue",
    "sendInputData": true,
    "options": {
      "durable": true,
      "assertQueue": true
    }
  }
}

Route log events through a topic exchange, choosing the routing key per item:

{
  "type": "rabbitmq",
  "parameters": {
    "operation": "sendMessage",
    "mode": "exchange",
    "exchange": "logs",
    "exchangeType": "topic",
    "routingKey": "app.{{ $json.severity }}",
    "sendInputData": true,
    "options": {
      "durable": true,
      "assertExchange": true
    }
  }
}

Broadcast a fixed announcement to every bound queue:

{
  "type": "rabbitmq",
  "parameters": {
    "operation": "sendMessage",
    "mode": "exchange",
    "exchange": "notifications",
    "exchangeType": "fanout",
    "routingKey": "",
    "sendInputData": false,
    "message": "System maintenance scheduled",
    "options": {
      "durable": false,
      "assertExchange": true
    }
  }
}

Publish a file from an upstream node, with headers and a TTL argument:

{
  "type": "rabbitmq",
  "parameters": {
    "operation": "sendMessage",
    "mode": "queue",
    "queue": "file-uploads",
    "sendInputData": false,
    "binaryData": true,
    "binaryPropertyName": "data",
    "options": {
      "durable": true,
      "assertQueue": true,
      "headers": {
        "header": [
          { "key": "source", "value": "workflow" }
        ]
      },
      "arguments": {
        "argument": [
          { "key": "expiration", "value": "60000" }
        ]
      }
    }
  }
}

Error Handling

Mode	Behavior
stop	Halts workflow on first error
continue	Skips failed items, passes successful ones through
errorPort	Routes failed items to Error output port

Tips

Publish messages (JSON, text, or binary) to RabbitMQ queues or exchanges via AMQP 0-9-1.