# rabbitmq

***

**1. Simple Message Publishing and Consumption**

```js
// Publish message
connection.createChannel((err, channel) => {
  channel.assertQueue('my-queue');
  channel.sendToQueue('my-queue', Buffer.from('Hello World!'));
});

// Consume message
connection.createChannel((err, channel) => {
  channel.assertQueue('my-queue');
  channel.consume('my-queue', (msg) => {
    console.log(msg.content.toString());
  });
});
```

**2. Message Prioritization**

```js
// Publish message with priority
channel.sendToQueue('my-queue', Buffer.from('High Priority Message'), {
  priority: 10,
});

// Consume message by priority
channel.consume('my-queue', (msg) => {
  console.log(msg.content.toString(), `Priority: ${msg.properties.priority}`);
}, {
  noAck: true,
});
```

**3. Message Expiration**

```js
// Publish message with expiration
channel.sendToQueue('my-queue', Buffer.from('Message Expires in 5s'), {
  expiration: 5000,
});

// Consume message before expiration
channel.consume('my-queue', (msg) => {
  console.log(msg.content.toString());
});
```

**4. Dead Letter Exchange**

```js
// Declare dead letter exchange
channel.assertExchange('my-dead-letter-exchange', 'direct');
channel.bindQueue('my-dead-letter-queue', 'my-dead-letter-exchange', 'my-dl-routing-key');

// Declare original exchange and bind to dead letter exchange
channel.assertExchange('my-exchange', 'topic');
channel.bindExchange('my-exchange', 'my-dead-letter-exchange', 'my-dl-routing-key');

// Publish message to original exchange
channel.publish('my-exchange', 'my-routing-key', Buffer.from('Message to Dead Letter Exchange'));
```

**5. Message Retries**

```js
// Define retry policy with max 5 attempts and delay between attempts
channel.publish('my-exchange', 'my-routing-key', Buffer.from('Message with Retries'), {
  retry: {
    max: 5,
    delay: 100,
  },
});
```

**6. Message Deduplication**

```js
// Define deduplication strategy
channel.prefetch(1);
channel.consume('my-queue', (msg) => {
  channel.nack(msg, false, true);
});
```

**7. Publish Confirms**

```js
// Enable publish confirms
channel.confirmChannel();

// Publish message
channel.sendToQueue('my-queue', Buffer.from('Message with Publish Confirm'));
```

**8. Transactional Publish**

```js
// Start transaction
channel.tx(async (err, tx) => {
  if (err) throw err;

  try {
    // Publish message in transaction
    await tx.publish('my-exchange', 'my-routing-key', Buffer.from('Message in Transaction'));

    // Commit transaction
    await tx.commit();
  } catch (err) {
    // Rollback transaction on error
    await tx.rollback();
  }
});
```

**9. Consumer Prefetching**

```js
// Set prefetch count
channel.prefetch(10);

// Consume messages
channel.consume('my-queue', (msg) => {
  // Process message
});
```

**10. Consumer Acknowledgements**

```js
// Consume message and acknowledge it
channel.consume('my-queue', (msg) => {
  channel.ack(msg);
});
```

**11. Consumer Nacks**

```js
// Consume message and nack it
channel.consume('my-queue', (msg) => {
  channel.nack(msg);
});
```

**12. Exchange Types (Direct)**

```js
// Declare direct exchange
channel.assertExchange('my-direct-exchange', 'direct');

// Bind queue to exchange
channel.bindQueue('my-direct-queue', 'my-direct-exchange', 'my-direct-routing-key');

// Publish message to exchange
channel.publish('my-direct-exchange', 'my-direct-routing-key', Buffer.from('Message to Direct Exchange'));
```

**13. Exchange Types (Fanout)**

```js
// Declare fanout exchange
channel.assertExchange('my-fanout-exchange', 'fanout');

// Bind queue to exchange
channel.bindQueue('my-fanout-queue', 'my-fanout-exchange', '');

// Publish message to exchange
channel.publish('my-fanout-exchange', '', Buffer.from('Message to Fanout Exchange'));
```

**14. Exchange Types (Topic)**

```js
// Declare topic exchange
channel.assertExchange('my-topic-exchange', 'topic');

// Bind queue to exchange
channel.bindQueue('my-topic-queue', 'my-topic-exchange', 'my-topic-routing-key.*');

// Publish message to exchange
channel.publish('my-topic-exchange', 'my-topic-routing-key.test', Buffer.from('Message to Topic Exchange'));
```

**15. Message Persistence**

```js
// Set queue durability
channel.assertQueue('my-durable-queue', {
  durable: true,
});
```

**16. Message Persistence Confirmation**

```js
// Enable publisher confirms
channel.confirmChannel();

// Publish message and wait for confirmation
channel.waitForConfirms(true);
channel.sendToQueue('my-durable-queue', Buffer.from('Message with Persistence Confirmation'));
```

**17. Consumer Recovery**

```js
// Use channel reconnect strategy
channel.addSetup((ch) => {
  ch.on('error', (err) => {
    if (err.code === 'ECONNRESET') {
      ch.setup();
    }
  });
});
```

**18. Queue Auto-Deletion**

```js
// Declare auto-delete queue
channel.assertQueue('my-auto-delete-queue', {
  autoDelete: true,
});
```

**19. Queue Expiration**

```js
// Declare queue with expiration
channel.assertQueue('my-expired-queue', {
  expires: 3600000, // 1 hour in milliseconds
});
```

**20. Exclusive Queue**

```js
// Declare exclusive queue
channel.assertQueue('my-exclusive-queue', {
  exclusive: true,
});
```

**21. Queue Message Limit**

```js
// Declare queue with message limit
channel.assertQueue('my-message-limit-queue', {
  messageLimit: 100,
});
```

**22. Queue Size Limit**

```js
// Declare queue with size limit
channel.assertQueue('my-size-limit-queue', {
  maxLength: 102400, // 100 KB
});
```

**23. Prefetch Size Limit**

```js
// Set prefetch size limit
channel.prefetch(null, 102400); // 100 KB
```

**24. Message TTL**

```js
// Publish message with TTL
channel.sendToQueue('my-ttl-queue', Buffer.from('Message with TTL'), {
  ttl: 30000, // 30 seconds
});
```

**25. RabbitMQ Management Plugin**

```js
// Use RabbitMQ Management Plugin
const management = require('rabbit-management');

// Connect to management API
const api = management('amqp://localhost:15672', 'guest', 'guest');

// List all queues
api.listQueues((err, queues) => {
  if (err) throw err;

  queues.forEach((queue) => console.log(queue.name));
});
```