Managing Plugins

rabbitmq-plugins list

[  ] rabbitmq_management   3.12.12
[ E] rabbitmq_web_stomp    3.12.12

rabbitmq-plugins enable rabbitmq_management

rabbitmq-plugins disable rabbitmq_management

{application, rabbitmq_management,
 [
  {description, "RabbitMQ Management Plugin"},
  {vsn, "3.12.0"},
  {registered, []},
  {applications, [
   kernel,
   stdlib,
   ranch,
   ssl,
   amqp_client
  ]},
  {included_applications, [rabbit_common]},
  {mod, {rabbit_management_app, []}},
  {env, [{http_port, 15672}]}
 ]}.

rabbitmq-plugins enable rabbitmq_management

http://localhost:15672

{
  "federation": {
    "exchanges": [
      {
        "name": "federated-exchange",
        "upstream": {
          "uri": "amqp://user:pass@remote-server:5672/vhost",
          "exchange": "source-exchange"
        }
      }
    ]
  }
}

{
  "federation": {
    "queues": [
      {
        "name": "federated-queue",
        "upstream": {
          "uri": "amqp://user:pass@remote-server:5672/vhost",
          "queue": "source-queue"
        }
      }
    ]
  }
}

{
  "shovel": {
    "source": {
      "uri": "amqp://source-broker",
      "queue": "source-queue"
    },
    "destination": {
      "uri": "amqp://destination-broker",
      "queue": "destination-queue"
    },
    "ack-mode": "on-confirm"
  }
}

const amqp = require('amqplib');

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

  const exchangeName = 'delayed_exchange';
  const exchangeType = 'x-delayed-message';
  const options = {
    durable: true,
    arguments: {
      'x-delayed-type': 'direct'
    }
  };

  await channel.assertExchange(exchangeName, exchangeType, options);
  console.log('Delayed exchange declared successfully');
  
  await channel.close();
  await connection.close();
}

declareDelayedExchange().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 routingKey = 'delayed.routing.key';
  const message = 'Hello, delayed world!';
  const delayInMs = 5000; // 5 seconds

  const messageOptions = {
    headers: {
      'x-delay': delayInMs
    }
  };

  channel.publish(exchangeName, routingKey, Buffer.from(message), messageOptions);
  console.log(`Message published with delay of ${delayInMs}ms`);

  await channel.close();
  await connection.close();
}

publishDelayedMessage().catch(console.error);

const amqp = require('amqplib');

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

  await channel.assertExchange('consistent-hash-exchange', 'consistent-hash', {
    durable: true,
    autoDelete: false
  });

  console.log('Consistent hash exchange declared');
  await channel.close();
  await connection.close();
}

declareConsistentHashExchange().catch(console.error);

const amqp = require('amqplib');

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

  const exchangeName = 'consistent-hash-exchange';
  const routingKey = 'user-123';

  await channel.publish(exchangeName, routingKey, Buffer.from('Hello World'));

  console.log(`Message published with routing key: ${routingKey}`);
  await channel.close();
  await connection.close();
}

publishToConsistentHashExchange().catch(console.error);

rabbitmq-top -n rabbit@localhost

rabbitmq-plugins enable rabbitmq_tracing

{
  "tracing": {
    "output": {
      "format": "json"
    }
  }
}

// Enable the recent history exchange plugin
// rabbitmq-plugins enable rabbitmq_recent_history_exchange

// Configure exchange with default settings
const config = {
  "recent_history_exchange": {
    "max_age": 3600,
    "max_size": 1000,
    "ttl": 3600
  }
};

// Apply configuration via RabbitMQ management API or config file

// Consume messages from recent history exchange
const amqp = require('amqplib');

async function consumeFromRecentHistory() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  // Bind queue to recent history exchange
  await channel.assertExchange('recent_history_exchange', 'direct');
  await channel.assertQueue('recent_history_queue');
  await channel.bindQueue('recent_history_queue', 'recent_history_exchange', '#');
  
  // Consume messages
  channel.consume('recent_history_queue', (msg) => {
    console.log('Received:', msg.content.toString());
    channel.ack(msg);
  });
}

consumeFromRecentHistory();