AMQP Nodes

AMQP 1.0 is the open, vendor-neutral standard for enterprise messaging, spoken natively by ActiveMQ Artemis, Azure Service Bus, Solace, IBM MQ, and RabbitMQ 4.x. MaestroHub provides five messaging node types for AMQP — Publish, Publish Batch, Request, Receive, and Browse. For starting pipelines on incoming messages, see the AMQP Trigger Node.

All five nodes share the same configuration shape: select the AMQP connection profile, select a function of the matching type, and map any ((parameter)) inputs the function declares.

Configuration Quick Reference

Field	What you choose	Details
Parameters	Connection, Function, Function Parameters, Timeout Override	Select the AMQP connection profile, the function, configure function parameters via expressions, and optionally override timeout.
Settings	Description, Timeout (seconds), Retry on Timeout, Retry on Fail, On Error	Node description, maximum execution time, retry behavior, and error handling strategy. All execution settings default to pipeline-level values.

RabbitMQ address format

On RabbitMQ 4.x every function's Address must use the v2 grammar — /queues/<name> or /exchanges/<exchange>/<routing-key> — and the queue must already exist. Other brokers accept plain queue/topic names. See the AMQP connector guide.

AMQP Publish Node

Send a single message to a queue or topic. The node waits for the broker's disposition — success means the broker settled the transfer, not just that bytes left the socket. Publish is Store & Forward eligible: with pipeline durability on, messages buffer locally through a broker outage and drain automatically afterwards.

Supported Function Types:

Function Name	Purpose	Common Use Cases
Publish	Send one message with optional message properties (content type, subject, correlation ID) and application properties	Line events to an enterprise bus, commands to a device queue, pipeline output to an ERP integration queue

How It Works

When the pipeline executes, the Publish node:

1. Connects via the selected AMQP connection profile (already established by the connector runtime)
2. Resolves any ((parameter)) placeholders in address, payload, and properties from upstream node outputs and trigger data
3. Sends the message and waits for the broker to settle the transfer (accept)
4. Returns the publish result — or, on a durable pipeline during an outage, reports the message as buffered for later drain

Publish Result

{
  "result": {
    "address": "orders",
    "messageSize": 42
  },
  "_metadata": { "success": true, "functionId": "<function-id>", "durationMs": 12, "timestamp": "2026-09-07T10:30:00Z" }
}

For detailed function configuration options, see the AMQP Function Builder documentation.

AMQP Publish Batch Node

Send an array of messages over a single sender link in one operation. Each element becomes one AMQP message and is individually settled by the broker; sending stops at the first rejection and the result reports how many were sent. Also Store & Forward eligible.

Supported Function Types:

Function Name	Purpose	Common Use Cases
Publish Batch	Send a JSON array of message bodies as individual messages	Bulk-forward buffered telemetry, ship a batch of transformed records

Publish Batch Result

{
  "result": {
    "address": "telemetry",
    "sent": 3
  },
  "_metadata": { "success": true, "functionId": "<function-id>", "durationMs": 12, "timestamp": "2026-09-07T10:30:00Z" }
}

AMQP Request Node

Send a request and synchronously await the correlated reply — RPC-style over the bus. The node creates a dynamic reply queue for the call, stamps the request's reply-to and message ID, and blocks until a reply with a matching correlation ID arrives or the timeout fires.

Supported Function Types:

Function Name	Purpose	Common Use Cases
Request	Send an AMQP request and wait for the correlated reply	RPC-style service calls, synchronous command/response with an ERP adapter

Request Result

{
  "result": {
    "address": "rpc.orders",
    "reply": "{\"status\":\"ok\",\"orderId\":\"ORD-12345\"}",
    "replySize": 38
  },
  "_metadata": { "success": true, "functionId": "<function-id>", "durationMs": 12, "timestamp": "2026-09-07T10:30:00Z" }
}

The reply field holds the responder's payload as a string. Use a downstream Code or Set node to JSON-parse it for typed access.

note

Dynamic reply queues require RabbitMQ 4.1+; Artemis, Service Bus, and the other brokers support them out of the box. The responder must copy the request's message ID into the reply's correlation ID.

AMQP Receive Node

Pull up to N messages from a queue in one shot, consuming them (each message is accepted and removed). The pull returns as soon as the queue stops delivering — an empty queue completes near-immediately with count: 0, which is success, not an error. This is the building block for scheduled queue-drain pipelines: a Schedule Trigger fires every few minutes, the Receive node pulls whatever is pending, downstream nodes process the batch.

Supported Function Types:

Function Name	Purpose	Common Use Cases
Receive	One-shot pull of up to Max Messages, with optional long-polling for the first message	Scheduled queue draining, fetching pending commands for sequential processing

How It Works

1. Opens a receiver with credit equal to the function's Max Messages
2. If Wait Time (seconds) is set, blocks up to that long for the first message (covers an empty queue and slow brokers)
3. Collects messages until Max Messages is reached or the Idle Cutoff window passes with no delivery ("queue is dry")
4. Accepts every collected message (removing them) and returns the batch

Receive Result

{
  "result": {
    "address": "work-items",
    "count": 2,
    "messages": [
      { "payload": "{\"job\":17}", "contentType": "application/json" },
      { "payload": "{\"job\":18}", "contentType": "application/json" }
    ]
  },
  "_metadata": { "success": true, "functionId": "<function-id>", "durationMs": 12, "timestamp": "2026-09-07T10:30:00Z" }
}

Each entry carries the message payload plus any properties the sender stamped (messageId, correlationId, contentType, subject, replyTo, applicationProperties). Note Max Messages is a cap, not a target — the node never waits for the queue to fill up to it.

A message sent by an AMQP Request node arrives with replyTo set to the address its sender is waiting on. To answer it, publish to that replyTo with the request's messageId as the Correlation ID.

AMQP Browse Node

Peek up to N messages without consuming them — every peeked message is released back to the broker and stays on the queue for real consumers. Result shape matches Receive, plus a "browsed": true marker.

Supported Function Types:

Function Name	Purpose	Common Use Cases
Browse	Non-destructive read of up to Max Messages	Dead-letter queue monitoring, verifying a producer before wiring a consumer

Browse Result

{
  "result": {
    "address": "orders.dlq",
    "count": 4,
    "browsed": true,
    "messages": [ { "payload": "…" } ]
  },
  "_metadata": { "success": true, "functionId": "<function-id>", "durationMs": 12, "timestamp": "2026-09-07T10:30:00Z" }
}

note

While the browse is in flight the peeked messages are temporarily held by the browsing receiver and invisible to competing consumers; their redelivery order afterwards is broker-dependent. See the connector guide for the full caveat.

Output

Every AMQP node delivers its data under result, and execution facts (success, functionId, durationMs, timestamp) under _metadata:

Node	Expression	Description
Publish	$node["Name"].result.address, $node["Name"].result.messageSize	The address, the payload size in bytes, and true — the broker settled the transfer
Publish Batch	$node["Name"].result.sent	How many messages the broker settled, and true
Request	$node["Name"].result.reply, $node["Name"].result.replySize	The responder's payload as text (parse it downstream if it is JSON), and its size
Receive, Browse	$node["Name"].result.count	How many messages were pulled — 0 on an empty queue is success
	$node["Name"].result.messages	One object per message: payload always; messageId, correlationId, contentType, subject, replyTo, applicationProperties when the sender stamped them. $node["Name"].result.messages[0].payload reads the first
Browse	$node["Name"].result.browsed	true — the messages were released back to the queue
all nodes	$node["Name"]._metadata.method, $node["Name"]._metadata.connectionId, $node["Name"]._metadata.protocol	The call's own facts: the operation, the connection it ran over, and amqp

Choosing the Right Node

You want to…	Node
Emit one event/command from the pipeline	Publish
Emit many records in one bounded operation	Publish Batch
Call a service over the bus and use its answer downstream	Request
Process pending queue items on a schedule (consuming them)	Receive
Look at a queue's content without touching it	Browse
Start a pipeline whenever a message arrives	AMQP Trigger