Working with Delayed Messages

curl -u guest:guest -H "Content-Type: application/json" \
-X PUT http://localhost:15672/api/exchanges/%2f/delayed_exchange \
-d '{"type":"x-delayed-message","durable":true,"arguments":{"x-delayed-type":"direct"}}'

import com.rabbitmq.client.*;

Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .headers(Map.of("x-delay", 30000))
    .build();

channel.basicPublish("", "queue-name", props, "Hello World".getBytes());

const amqp = require('amqplib');

async function setupDelayedQueue() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Declare the dead letter exchange
  await channel.assertExchange('delayed-messages-exchange', 'direct');

  // Declare the queue with TTL and DLX
  await channel.assertQueue('delayed-queue', {
    arguments: {
      'x-message-ttl': 30000,
      'x-dead-letter-exchange': 'delayed-messages-exchange',
      'x-dead-letter-routing-key': 'delayed-messages-key'
    }
  });

  // Bind the queue to the DLX
  await channel.bindQueue('delayed-queue', 'delayed-messages-exchange', 'delayed-messages-key');

  console.log('Delayed queue configured successfully');
}

setupDelayedQueue().catch(console.error);

const amqp = require('amqplib');

async function publishDelayedMessage() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const exchangeName = 'delayed_exchange';
  const queueName = 'delayed_queue';

  await channel.assertExchange(exchangeName, 'x-delayed-message', { durable: true, arguments: { 'x-delayed-type': 'direct' } });
  await channel.assertQueue(queueName, { durable: true });
  await channel.bindQueue(queueName, exchangeName, 'delayed_key');

  const message = { text: 'Hello, delayed world!' };
  const delay = 5000; // Delay for 5 seconds

  await channel.publish(exchangeName, 'delayed_key', Buffer.from(JSON.stringify(message)), {
    headers: { 'x-delay': delay }
  });

  console.log('Delayed message published');
  await channel.close();
  await connection.close();
}

publishDelayedMessage().catch(console.error);

// Using RabbitMQ Management API to monitor delayed messages
const axios = require('axios');

async function getDelayedMessageStats(vhost = '%2F') {
  const response = await axios.get(
    `http://localhost:15672/api/queues/${vhost}/my-delayed-queue`,
    {
      auth: {
        username: 'guest',
        password: 'guest'
      }
    }
  );

  const queueInfo = response.data;
  console.log('Delayed Messages:', queueInfo.messages_delayed);
  console.log('Total Messages:', queueInfo.messages);
}

getDelayedMessageStats();

# Using rabbitmq-diagnostics to check delayed message status
rabbitmq-diagnostics status

# Output includes:
# - delayed_message_count: 0
# - queue_length: 10

{"delayed_message_queue": {
  "max_delayed_messages": 50000,
  "overflow_strategy": "drop"
}}