UOR Rabbit

A reusable RabbitMQ utility library for UOR Budget applications that provides both producer and consumer functionality.

Features

• Simple, configurable RabbitMQ connection management
• Message producer for sending messages to queues
• Message consumer for receiving and processing messages
• Automatic reconnection handling
• Event-based communication
• Flexible message format support
• Error handling and logging

Installation

npm install uor-rabbit

Usage

Producer Example

const { RabbitProducer } = require('uor-rabbit');

// Initialize producer with custom options
const producer = new RabbitProducer({
  url: process.env.RABBITMQ_URL || 'amqp://localhost',
  queueName: process.env.RABBITMQ_QUEUE || 'telegram_notifications',
  logger: console // or your custom logger
});

// Connect to RabbitMQ
await producer.initialize();

// Send a message
await producer.sendMessage({
  receivers: ['123456789'],
  body: 'Hello from UOR Budget!',
  data: { timestamp: new Date().toISOString() }
});

// Or send to multiple receivers
await producer.sendMessage({
  receivers: ['123456789', '987654321'],
  body: 'Broadcast message',
  data: { type: 'notification' }
});

// Listen for events
producer.on('messageSent', (message) => {
  console.log('Message sent successfully:', message);
});

producer.on('error', (error) => {
  console.error('RabbitMQ error:', error.message);
});

// Close connection when done
await producer.close();

Consumer Example

const { RabbitConsumer } = require('uor-rabbit');

// Initialize consumer with custom options
const consumer = new RabbitConsumer({
  url: process.env.RABBITMQ_URL || 'amqp://localhost',
  queueName: process.env.RABBITMQ_QUEUE || 'telegram_notifications',
  logger: console // or your custom logger
});

// Connect to RabbitMQ
await consumer.initialize();

// Register message handler
consumer.onMessage(async (receivers, body, data) => {
  console.log(`Processing message for ${receivers.length} receivers`);
  
  for (const userId of receivers) {
    // Process each receiver
    console.log(`Sending message to user ${userId}: ${body}`);
  }
});

// Start consuming messages
await consumer.startConsuming();

// Listen for events
consumer.on('message', (message) => {
  console.log('Received message:', message);
});

consumer.on('error', (error) => {
  console.error('RabbitMQ error:', error.message);
});

// Close connection when done
await consumer.close();

Configuration Options

Both RabbitProducer and RabbitConsumer accept the following options:

| Option | Description | Default | |--------|-------------|---------| | url | RabbitMQ connection URL | process.env.RABBITMQ_URL or 'amqp://localhost' | | queueName | Queue name | process.env.RABBITMQ_QUEUE or 'telegram_notifications' | | queueOptions | Queue options | { durable: true } | | reconnectInterval | Reconnect interval in ms | 5000 | | maxReconnectAttempts | Maximum reconnect attempts | 3 | | logger | Logger instance | console | | messageOptions | Message options (Producer only) | { persistent: true } | | consumeOptions | Consume options (Consumer only) | {} |

Message Format

The library supports the following message format:

{
  "receivers": ["123456789", "987654321"],
  "body": "Your notification message",
  "data": { "key": "value" }
}

The receivers field must be a non-empty array of receiver IDs.

Events

Common Events

• connected: Emitted when connected to RabbitMQ
• error: Emitted on general errors
• connectionError: Emitted on connection errors
• connectionClosed: Emitted when connection is closed unexpectedly
• reconnected: Emitted when successfully reconnected
• reconnectError: Emitted when reconnection fails
• maxReconnectAttemptsReached: Emitted when maximum reconnect attempts are reached
• closed: Emitted when connection is closed
• closeError: Emitted when there's an error closing the connection

Producer Events

• messageSent: Emitted when a message is sent successfully
• sendError: Emitted when there's an error sending a message

Consumer Events

• consumingStarted: Emitted when consuming starts
• consumeError: Emitted when there's an error starting to consume
• message: Emitted when a message is received and validated
• messageError: Emitted when there's an error processing a message
• invalidMessage: Emitted when a message is invalid

License

AGPL-3.0