# RabbitMQ

> Publish and read RabbitMQ messages, create, purge and delete queues and exchanges, bind them together, and check the broker's health, through the management API of your own broker or CloudAMQP.

App · Databases · 20 actions. Web page: https://docs.lodol.com/docs/integrations/rabbitmq

## Actions

### Bind exchange to exchange

Route messages from one exchange on to another. Returns the binding, including the properties key needed to delete it.

**Inputs**

- Source exchange (`exchange`, text, required): The exchange messages come from.
- Destination exchange (`destination_exchange`, text, required): The exchange to pass them on to.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Routing key (`routing_key`, text): The routing key the binding matches. For a topic exchange use * for one word and # for any number, such as orders.*. Leave empty for none (fanout exchanges ignore it).
- Arguments (`arguments`, object): Optional binding arguments. For a headers exchange, the headers to match plus x-match set to all or any.

**Outputs**

- `arguments` (object): Binding arguments
- `destination` (text): Destination queue or exchange
- `destination_type` (text): queue or exchange
- `properties_key` (text): The binding's id, used to delete it
- `routing_key` (text): Routing key
- `source` (text): Source exchange (empty for the default exchange)
- `vhost` (text): Virtual host

**Example output**

```json
{
  "arguments": {},
  "destination": "orders.audit",
  "destination_type": "exchange",
  "properties_key": "orders.created",
  "routing_key": "orders.created",
  "source": "orders.events",
  "vhost": "/"
}
```

### Bind queue to exchange

Route messages from an exchange to a queue. Returns the binding, including the properties key needed to delete it.

**Inputs**

- Exchange (`exchange`, text, required): The exchange messages come from.
- Queue (`queue`, text, required): The queue to deliver them to.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Routing key (`routing_key`, text): The routing key the binding matches. For a topic exchange use * for one word and # for any number, such as orders.*. Leave empty for none (fanout exchanges ignore it).
- Arguments (`arguments`, object): Optional binding arguments. For a headers exchange, the headers to match plus x-match set to all or any.

**Outputs**

- `arguments` (object): Binding arguments
- `destination` (text): Destination queue or exchange
- `destination_type` (text): queue or exchange
- `properties_key` (text): The binding's id, used to delete it
- `routing_key` (text): Routing key
- `source` (text): Source exchange (empty for the default exchange)
- `vhost` (text): Virtual host

**Example output**

```json
{
  "arguments": {},
  "destination": "orders",
  "destination_type": "queue",
  "properties_key": "orders.created",
  "routing_key": "orders.created",
  "source": "orders.events",
  "vhost": "/"
}
```

### Check alarms

Whether a memory or disk alarm is in effect anywhere in the cluster. While one is, RabbitMQ blocks publishers. Returns healthy true or false rather than failing the step.

**Inputs**: None.

**Outputs**

- `alarms` (list)
- `healthy` (true/false): True when no alarm is in effect
- `reason` (text): Why the check failed, if it did
- `status` (text): ok or failed

**Example output**

```json
{
  "alarms": [],
  "healthy": true,
  "reason": null,
  "status": "ok"
}
```

### Create exchange

Create an exchange. If an identical one already exists nothing changes; one with different settings is an error. Returns the exchange.

**Inputs**

- Exchange name (`exchange`, text, required): Up to 255 characters. Names starting with amq. are reserved.
- Type (`exchange_type`, text, required): How the exchange routes messages to queues. One of: `direct` (Direct - route by exact routing key), `fanout` (Fanout - copy to every bound queue), `topic` (Topic - route by routing key pattern), `headers` (Headers - route by message headers). Default: `direct`.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Durable (`durable`, true/false): Keep the exchange when the broker restarts. Default: `true`.
- Delete when the last binding is removed (`auto_delete`, true/false): Delete the exchange once it has had a binding and the last one is removed. Default: `false`.
- Internal (`internal`, true/false): Only other exchanges can route to it; clients cannot publish to it directly. Default: `false`.
- Alternate exchange (`alternate_exchange`, text): Where to send messages this exchange cannot route, instead of dropping them.
- Other arguments (`arguments`, object): Any other exchange arguments.

**Outputs**

- `arguments` (object): Exchange arguments
- `auto_delete` (true/false): Deleted when its last binding is removed
- `created` (true/false): False when an identical exchange already existed
- `durable` (true/false): Survives a broker restart
- `internal` (true/false): Only other exchanges can publish to it
- `name` (text): Exchange name (empty for the default exchange)
- `policy` (text): Policy applied, if any
- `type` (text): direct, fanout, topic or headers
- `vhost` (text): Virtual host

**Example output**

```json
{
  "arguments": {},
  "auto_delete": false,
  "created": true,
  "durable": true,
  "internal": false,
  "name": "orders.events",
  "type": "topic",
  "vhost": "/"
}
```

### Create queue

Create a durable queue. If an identical queue already exists nothing changes; one with different settings is an error. Returns the queue.

**Inputs**

- Queue name (`queue`, text, required): Up to 255 characters, such as orders.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Queue type (`queue_type`, text): Leave unset to use the virtual host's default. Quorum queues are replicated across nodes; streams keep messages after they are read. One of: `classic`, `quorum` (Quorum (replicated)), `stream`.
- Delete when the last consumer leaves (`auto_delete`, true/false): Delete the queue once it has had a consumer and the last one disconnects. Classic queues only. Default: `false`.
- Message time to live (seconds) (`message_ttl_seconds`, number): Discard messages that have waited this long. Leave empty to keep them until read. At least 0.001.
- Delete when unused for (seconds) (`expires_seconds`, number): Delete the queue after it has had no consumers and no reads for this long. Good for temporary queues. At least 0.001.
- Maximum messages (`max_length`, number): Once the queue holds this many, the oldest are dropped (or dead-lettered). Leave empty for no limit. At least 0.
- Dead-letter exchange (`dead_letter_exchange`, text): Exchange that receives messages that expire, are dropped for length or are rejected without requeue.
- Dead-letter routing key (`dead_letter_routing_key`, text): Routing key for dead-lettered messages. Leave empty to keep each message's own.
- Other arguments (`arguments`, object): Any other queue arguments, such as x-max-priority. The fields above take precedence.

**Outputs**

- `arguments` (object): Queue arguments such as x-queue-type
- `auto_delete` (true/false): Deleted when its last consumer leaves
- `consumers` (number): Number of consumers
- `created` (true/false): False when an identical queue already existed
- `durable` (true/false): Survives a broker restart
- `effective_policy_definition` (object): The policy settings in effect
- `exclusive` (true/false): Owned by one connection
- `messages` (number): Messages in the queue (ready + unacknowledged)
- `messages_ready` (number): Messages ready for delivery
- `messages_unacknowledged` (number): Messages delivered but not acknowledged
- `name` (text): Queue name
- `node` (text): Node hosting the queue
- `policy` (text): Policy applied, if any
- `state` (text): running, idle, ...
- `type` (text): classic, quorum or stream
- `vhost` (text): Virtual host

**Example output**

```json
{
  "arguments": {
    "x-queue-type": "classic"
  },
  "auto_delete": false,
  "consumers": 1,
  "created": true,
  "durable": true,
  "effective_policy_definition": {},
  "exclusive": false,
  "messages": 0,
  "messages_ready": 3,
  "messages_unacknowledged": 0,
  "name": "orders",
  "node": "rabbit@rabbit-1",
  "policy": null,
  "state": "running",
  "type": "classic",
  "vhost": "/"
}
```

### Delete binding

Remove one binding from an exchange to a queue or exchange, identified by its properties key.

**Inputs**

- Source exchange (`exchange`, text, required): The exchange the binding routes from.
- Bound to a (`destination_type`, text): Whether the binding delivers to a queue or exchange. One of: `queue`, `exchange`. Default: `queue`.
- Destination (`destination`, text, required): The queue or exchange the binding delivers to.
- Properties key (`properties_key`, text): The binding's properties_key, from a Bind step or List queue bindings. Leave empty for a binding with no routing key and no arguments.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).

**Outputs**

- `deleted` (true/false): Always true
- `destination` (text): Destination
- `destination_type` (text): queue or exchange
- `properties_key` (text): The binding's id
- `source` (text): Source exchange
- `vhost` (text): Virtual host

**Example output**

```json
{
  "deleted": true,
  "destination": "orders",
  "destination_type": "queue",
  "properties_key": "orders.created",
  "source": "orders.events",
  "vhost": "/"
}
```

### Delete exchange

Delete an exchange and its bindings. Optionally only if nothing is bound to it.

**Inputs**

- Exchange (`exchange`, text, required): The exchange's name, exactly as it appears in List exchanges.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Only if unused (`if_unused`, true/false): Fail instead of deleting an exchange with bindings. Default: `false`.

**Outputs**

- `deleted` (true/false): Always true
- `exchange` (text): Exchange
- `vhost` (text): Vhost

**Example output**

```json
{
  "deleted": true,
  "exchange": "orders.events",
  "vhost": "/"
}
```

### Delete queue

Delete a queue and any messages in it. Optionally only if it is empty or has no consumers.

**Inputs**

- Queue (`queue`, text, required): The queue's name, exactly as it appears in List queues.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Only if empty (`if_empty`, true/false): Fail instead of deleting a queue with messages. Default: `false`.
- Only if it has no consumers (`if_unused`, true/false): Fail instead of deleting a queue being consumed. Default: `false`.

**Outputs**

- `deleted` (true/false): Always true
- `queue` (text): Queue
- `vhost` (text): Vhost

**Example output**

```json
{
  "deleted": true,
  "queue": "orders",
  "vhost": "/"
}
```

### Get current user

The RabbitMQ user this connection signs in as, with its tags. Useful to check the connection works.

**Inputs**: None.

**Outputs**

- `is_internal_user` (true/false): Whether RabbitMQ's own user database holds it
- `name` (text): Username
- `tags` (list)

**Example output**

```json
{
  "is_internal_user": true,
  "name": "lodol",
  "tags": [
    "management"
  ]
}
```

### Get exchange

One exchange's type, settings and arguments.

**Inputs**

- Exchange (`exchange`, text): The exchange's name. Leave empty for the default exchange.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).

**Outputs**

- `arguments` (object): Exchange arguments
- `auto_delete` (true/false): Deleted when its last binding is removed
- `durable` (true/false): Survives a broker restart
- `internal` (true/false): Only other exchanges can publish to it
- `name` (text): Exchange name (empty for the default exchange)
- `policy` (text): Policy applied, if any
- `type` (text): direct, fanout, topic or headers
- `vhost` (text): Virtual host

**Example output**

```json
{
  "arguments": {},
  "auto_delete": false,
  "durable": true,
  "internal": false,
  "name": "orders.events",
  "type": "topic",
  "vhost": "/"
}
```

### Get messages

Read messages from the front of a queue - either peeking and putting them back, or taking them off the queue. Meant for occasional reads, not as a consumer: a peeked message comes back marked redelivered and may change position.

**Inputs**

- Queue (`queue`, text, required): The queue's name, exactly as it appears in List queues.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- How many (`count`, number): Up to 100. Defaults to 1. From 1 to 100. Default: `1`.
- Mode (`ackmode`, text): Whether the messages stay on the queue. One of: `ack_requeue_true` (Peek - read and put back), `ack_requeue_false` (Take - read and remove from the queue), `reject_requeue_true` (Reject and put back), `reject_requeue_false` (Reject and remove (dead-letters it if the queue has a dead-letter exchange)). Default: `ack_requeue_true`.
- Payload format (`encoding`, text): How each message body is returned. One of: `auto` (Text when possible, otherwise base64), `base64` (Always base64). Default: `auto`.
- Cut bodies at (bytes) (`truncate`, number): Shorten longer bodies to this many bytes. At least 1.

**Outputs**

- `count` (number): How many came back
- `messages` (list)

**Example output**

```json
{
  "count": 1,
  "messages": [
    {
      "exchange": "orders.events",
      "message_count": 2,
      "payload": "{\"order_id\": 1042}",
      "payload_bytes": 18,
      "payload_encoding": "string",
      "payload_json": {
        "order_id": 1042
      },
      "properties": {
        "content_type": "application/json",
        "delivery_mode": 2
      },
      "redelivered": false,
      "routing_key": "orders.created"
    }
  ]
}
```

### Get overview

The broker at a glance: cluster name, RabbitMQ and Erlang versions, the exchange types it supports, and totals of queued messages, queues, exchanges, connections and consumers.

**Inputs**: None.

**Outputs**

- `cluster_name` (text): Cluster name
- `default_queue_type` (text): Default queue type
- `erlang_version` (text): Erlang version
- `exchange_types` (list)
- `management_version` (text): Management plugin version
- `message_stats` (object): Message rates
- `object_totals` (object)
- `product_name` (text): Product name
- `product_version` (text): Product version
- `queue_totals` (object)
- `rabbitmq_version` (text): RabbitMQ version

**Example output**

```json
{
  "cluster_name": "rabbit@rabbit-1",
  "default_queue_type": "classic",
  "erlang_version": "27.3",
  "exchange_types": [
    {
      "description": "",
      "enabled": true,
      "name": "direct"
    }
  ],
  "management_version": "4.3.6",
  "object_totals": {
    "channels": 2,
    "connections": 2,
    "consumers": 1,
    "exchanges": 8,
    "queues": 1
  },
  "product_name": "RabbitMQ",
  "product_version": "4.3.6",
  "queue_totals": {
    "messages": 3,
    "messages_ready": 3,
    "messages_unacknowledged": 0
  },
  "rabbitmq_version": "4.3.6"
}
```

### Get queue

One queue in full: type, settings, how many messages are ready and unacknowledged, its consumers and the policy in effect.

**Inputs**

- Queue (`queue`, text, required): The queue's name, exactly as it appears in List queues.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).

**Outputs**

- `arguments` (object): Queue arguments such as x-queue-type
- `auto_delete` (true/false): Deleted when its last consumer leaves
- `consumers` (number): Number of consumers
- `durable` (true/false): Survives a broker restart
- `effective_policy_definition` (object): The policy settings in effect
- `exclusive` (true/false): Owned by one connection
- `messages` (number): Messages in the queue (ready + unacknowledged)
- `messages_ready` (number): Messages ready for delivery
- `messages_unacknowledged` (number): Messages delivered but not acknowledged
- `name` (text): Queue name
- `node` (text): Node hosting the queue
- `policy` (text): Policy applied, if any
- `state` (text): running, idle, ...
- `type` (text): classic, quorum or stream
- `vhost` (text): Virtual host

**Example output**

```json
{
  "arguments": {
    "x-queue-type": "classic"
  },
  "auto_delete": false,
  "consumers": 1,
  "durable": true,
  "effective_policy_definition": {},
  "exclusive": false,
  "messages": 3,
  "messages_ready": 3,
  "messages_unacknowledged": 0,
  "name": "orders",
  "node": "rabbit@rabbit-1",
  "policy": null,
  "state": "running",
  "type": "classic",
  "vhost": "/"
}
```

### List consumers

The consumers subscribed in a virtual host, optionally only one queue's, with their prefetch and acknowledgement settings.

**Inputs**

- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Queue (`queue`, text): Only this queue's consumers. Leave empty for all.

**Outputs**

- `count` (number): How many came back
- `items` (list): The consumers

**Example output**

```json
{
  "count": 1,
  "items": [
    {
      "ack_required": true,
      "active": true,
      "activity_status": "up",
      "arguments": {},
      "consumer_tag": "amq.ctag-3fJ1xQ",
      "exclusive": false,
      "prefetch_count": 10,
      "queue": {
        "name": "orders",
        "vhost": "/"
      }
    }
  ]
}
```

### List exchanges

The exchanges in a virtual host, sorted by name, with each one's type and settings. The default exchange has an empty name.

**Inputs**

- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Name contains (`name_filter`, text): Only names containing this text. Leave empty for all of them.
- Treat the name as a regular expression (`use_regex`, true/false): Match 'Name contains' as a regular expression instead of text. Default: `false`.
- Page (`page`, number): Which page of results, starting at 1. From 1 to 1000000. Default: `1`.
- Results per page (`page_size`, number): How many to return, 1-500. Defaults to 100. From 1 to 500. Default: `100`.

**Outputs**

- `count` (number): How many came back on this page
- `filtered_count` (number): How many match the name filter
- `has_more` (true/false): Whether a later page exists
- `items` (list): The matching exchanges
- `page` (number): This page's number
- `page_count` (number): How many pages match
- `page_size` (number): Page size asked for
- `total_count` (number): How many exist in the virtual host

**Example output**

```json
{
  "count": 1,
  "filtered_count": 1,
  "has_more": false,
  "items": [
    {
      "arguments": {},
      "auto_delete": false,
      "durable": true,
      "internal": false,
      "name": "orders.events",
      "type": "topic",
      "vhost": "/"
    }
  ],
  "page": 1,
  "page_count": 1,
  "page_size": 100,
  "total_count": 1
}
```

### List queue bindings

Every binding that delivers to a queue: the source exchange, routing key and arguments, and the properties key needed to delete it.

**Inputs**

- Queue (`queue`, text, required): The queue's name, exactly as it appears in List queues.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).

**Outputs**

- `count` (number): How many came back
- `items` (list): The queue's bindings

**Example output**

```json
{
  "count": 1,
  "items": [
    {
      "arguments": {},
      "destination": "orders",
      "destination_type": "queue",
      "properties_key": "orders.created",
      "routing_key": "orders.created",
      "source": "orders.events",
      "vhost": "/"
    }
  ]
}
```

### List queues

The queues in a virtual host, sorted by name, with each one's type, state and message and consumer counts. Filter by part of the name.

**Inputs**

- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Name contains (`name_filter`, text): Only names containing this text. Leave empty for all of them.
- Treat the name as a regular expression (`use_regex`, true/false): Match 'Name contains' as a regular expression instead of text. Default: `false`.
- Page (`page`, number): Which page of results, starting at 1. From 1 to 1000000. Default: `1`.
- Results per page (`page_size`, number): How many to return, 1-500. Defaults to 100. From 1 to 500. Default: `100`.

**Outputs**

- `count` (number): How many came back on this page
- `filtered_count` (number): How many match the name filter
- `has_more` (true/false): Whether a later page exists
- `items` (list): The matching queues
- `page` (number): This page's number
- `page_count` (number): How many pages match
- `page_size` (number): Page size asked for
- `total_count` (number): How many exist in the virtual host

**Example output**

```json
{
  "count": 1,
  "filtered_count": 1,
  "has_more": false,
  "items": [
    {
      "arguments": {
        "x-queue-type": "classic"
      },
      "auto_delete": false,
      "consumers": 1,
      "durable": true,
      "effective_policy_definition": {},
      "exclusive": false,
      "messages": 3,
      "messages_ready": 3,
      "messages_unacknowledged": 0,
      "name": "orders",
      "node": "rabbit@rabbit-1",
      "policy": null,
      "state": "running",
      "type": "classic",
      "vhost": "/"
    }
  ],
  "page": 1,
  "page_count": 1,
  "page_size": 100,
  "total_count": 1
}
```

### List virtual hosts

The virtual hosts this user has permissions on, with their descriptions, default queue type and message counts.

**Inputs**: None.

**Outputs**

- `count` (number): How many came back
- `items` (list): The virtual hosts

**Example output**

```json
{
  "count": 1,
  "items": [
    {
      "default_queue_type": "classic",
      "description": "Default virtual host",
      "messages": 3,
      "name": "/",
      "tags": [],
      "tracing": false
    }
  ]
}
```

### Publish message

Publish a message to an exchange, or straight to a queue by leaving the exchange empty and putting the queue's name in Routing key. Returns whether any queue received it.

**Inputs**

- Exchange (`exchange`, text): The exchange to publish to. Leave empty to deliver straight to the queue named in Routing key.
- Routing key (`routing_key`, text): Decides which bound queues receive the message - or, with no exchange, the queue's name.
- Message body (`payload`, text, required): The message. An object from an earlier step is sent as JSON.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).
- Body format (`payload_encoding`, text): Choose Base64 to send binary data encoded as base64. One of: `string` (Text), `base64` (Base64 (binary data)). Default: `string`.
- Content type (`content_type`, text): Such as application/json or text/plain.
- Persistent (`persistent`, true/false): Write the message to disk so it survives a broker restart (in a durable queue). Default: `true`.
- Headers (`headers`, object): Message headers, also used by headers exchanges.
- Priority (`priority`, number): 0-255. Only matters on a queue created with x-max-priority. From 0 to 255.
- Correlation ID (`correlation_id`, text): Ties a reply to its request.
- Reply to (`reply_to`, text): The queue a reply should go to.
- Expires after (seconds) (`expiration_seconds`, number): Discard the message if unread after this long. At least 0.001.
- Message ID (`message_id`, text): Your own identifier for the message.
- Timestamp (`timestamp`, date and time): The message's timestamp property.
- Message type (`message_type`, text): The message's type property, such as order.created.
- App ID (`app_id`, text): The publishing application's name.
- Fail if no queue receives it (`fail_if_unroutable`, true/false): Fail the step instead of succeeding when the message matched no binding and was dropped. Default: `false`.

**Outputs**

- `exchange` (text): Exchange published to (empty for the default)
- `routed` (true/false): Whether at least one queue received the message
- `routing_key` (text): Routing key used
- `vhost` (text): Virtual host

**Example output**

```json
{
  "exchange": "orders.events",
  "routed": true,
  "routing_key": "orders.created",
  "vhost": "/"
}
```

### Purge queue

Delete every ready message in a queue, keeping the queue. This cannot be undone.

**Inputs**

- Queue (`queue`, text, required): The queue's name, exactly as it appears in List queues.
- Virtual host (`vhost`, text): The virtual host to work in. Leave empty to use the one set on the connection ('/' if none was).

**Outputs**

- `messages_before` (number): Approximate messages in the queue before
- `purged` (true/false): Always true
- `queue` (text): Queue name
- `vhost` (text): Virtual host

**Example output**

```json
{
  "messages_before": 3,
  "purged": true,
  "queue": "orders",
  "vhost": "/"
}
```