A simple Node.js script to move messages between AMQP queues using the RabbitMQ Management API.

README.md

AMQP Message Mover

A simple Node.js script to move messages between AMQP queues using the RabbitMQ Management API.

Description

This script allows you to move messages from one queue to another in a RabbitMQ instance. It fetches messages from the origin queue and publishes them to the destination queue using the RabbitMQ HTTP API.

Prerequisites

• Node.js (version 14 or higher)
• Access to RabbitMQ Management API
• Valid authentication token for the RabbitMQ instance

Environment Variables

• AMQP_TOKEN - Base64 encoded authentication token for RabbitMQ Management API

Usage

Running with npx from GitHub Gist

npx https://gist.github.com/filiafobico/ecc52e8bb539c8a9302fb4c67b6b3863 \
  --host=https://your-rabbitmq-host.com \
  --origin=source-queue-name \
  --destination=target-queue-name \
  --vhost=your-virtual-host \
  --count=10

Running locally

# Set your authentication token
export AMQP_TOKEN="your_base64_encoded_token"

# Run the script
node index.mjs \
  --host=https://your-rabbitmq-host.com \
  --origin=source-queue-name \
  --destination=target-queue-name \
  --vhost=your-virtual-host \
  --count=10

Parameters

Parameter	Required	Description	Default
--host	Yes	RabbitMQ Management API base URL	-
--origin	Yes	Source queue name	-
--destination	Yes	Destination queue name	-
--vhost	Yes	Virtual host name	-
--count	No	Number of messages to move	1

Authentication Token

The AMQP_TOKEN should be a Base64 encoded string of your RabbitMQ username and password:

echo -n "username:password" | base64

Example

# Set token
export AMQP_TOKEN="dXNlcm5hbWU6cGFzc3dvcmQ="

# Move 5 messages from dlq to main queue
npx https://gist.github.com/filiafobico/ecc52e8bb539c8a9302fb4c67b6b3863 \
  --host=https://rabbitmq.example.com \
  --origin=my-service.dlq \
  --destination=my-service.main \
  --vhost=production \
  --count=5

Notes

• Messages are acknowledged and removed from the source queue after being moved
• The script preserves message properties during the move operation
• All messages are logged to the console during processing
• If any required parameter is missing, the script will exit with an error message

Error Handling

The script includes basic error handling and will:

• Validate all required parameters before execution
• Log errors for failed API calls
• Exit with code 1 if required parameters are missing

index.mjs

#!/usr/bin/env node

const token = process.env.AMQP_TOKEN;

const host = process.argv.find(arg => arg.startsWith('--host='))?.split('=')[1]
const origin = process.argv.find(arg => arg.startsWith('--origin='))?.split('=')[1]
const destination = process.argv.find(arg => arg.startsWith('--destination='))?.split('=')[1]
const vhost = process.argv.find(arg => arg.startsWith('--vhost='))?.split('=')[1]
const count = +process.argv.find(arg => arg.startsWith('--count='))?.split('=')[1] || 1

if (!token || !host || !origin || !destination || !vhost) {
  console.error('Missing required parameters. Usage: node amqpMove.mjs --host=HOST --origin=ORIGIN_QUEUE --destination=DESTINATION_QUEUE --vhost=VHOST [--count=NUMBER_OF_MESSAGES]');
  process.exit(1);
}

const myHeaders = new Headers();
myHeaders.append("Content-Type", "application/json");
myHeaders.append("Authorization", `Basic ${token}`);

console.log(`Moving messages from ${origin} to ${destination}`);

const messages = await fetch(`${host}/api/queues/${vhost}/${origin}/get`, {
  method: "POST",
  headers: myHeaders,
  body: JSON.stringify({
    count,
    "ackmode": "ack_requeue_false",
    "encoding": "auto"
  }),
  redirect: "follow"
})
  .then((response) => response.json())
  .catch((error) => console.error(error));

for (const message of messages) {
  console.log(message);

  await fetch(`${host}/api/exchanges/${vhost}/amq.default/publish`, {
    method: "POST",
    headers: myHeaders,
    body: JSON.stringify({
      "properties": message.properties,
      "routing_key": destination,
      "payload": message.payload,
      "payload_encoding": "string"
    }),
    redirect: "follow"
  })
    .then((response) => response.text())
    .then((result) => console.log(result))
    .catch((error) => console.error(error));
}

package.json

{
  "name": "amqp-message-mover",
  "description": "",
  "version": "0.0.1",
  "bin": "./index.mjs"
}