Redis Publisher

A robust Redis publisher package for Socket.IO room-based messaging with automatic reconnection and error handling. Supports both CommonJS and ES modules.

Features

• Automatic reconnection with configurable retry strategy
• Comprehensive error handling
• Event-based callbacks for connection and publishing events
• Health monitoring and status reporting
• TypeScript support
• Environment variable configuration
• Graceful degradation
• Dual module support (CommonJS and ES modules)
• Singleton service pattern support

Installation

npm install redis-publisher

Usage

Creating a Publish Service

You can create a singleton publish service that can be used throughout your application:

TypeScript Version

// services/publish.service.ts
import { RedisPublisher } from "redis-publisher";

class PublishService {
  private static instance: RedisPublisher;
  private static isInitialized = false;

  private constructor() {}

  public static getInstance(): RedisPublisher {
    if (!PublishService.instance) {
      PublishService.instance = new RedisPublisher({
        host: process.env.REDIS_HOST,
        port: parseInt(process.env.REDIS_PORT || "6379"),
        password: process.env.REDIS_PASSWORD, // optional, can also be set via env
        maxRetries: 5,
        retryDelay: 5000,
        onError: (error) => {
          console.error("Redis error occurred:", error);
        },
        onReconnect: () => {
          console.log("Successfully reconnected to Redis");
        },
      });
      PublishService.isInitialized = true;
    }
    return PublishService.instance;
  }

  public static async initialize(): Promise<void> {
    if (!PublishService.isInitialized) {
      const publisher = PublishService.getInstance();
      await publisher.connect();
    }
  }

  public static isServiceInitialized(): boolean {
    return PublishService.isInitialized;
  }
}

export default PublishService;

CommonJS Version

// services/publish.service.js
const { RedisPublisher } = require("redis-publisher");

class PublishService {
  static #instance = null;
  static #isInitialized = false;

  static getInstance() {
    if (!PublishService.#instance) {
      PublishService.#instance = new RedisPublisher({
        host: process.env.REDIS_HOST,
        port: parseInt(process.env.REDIS_PORT || "6379"),
        password: process.env.REDIS_PASSWORD, // optional, can also be set via env
        maxRetries: 5,
        retryDelay: 5000,
        onError: (error) => {
          console.error("Redis error occurred:", error);
        },
        onReconnect: () => {
          console.log("Successfully reconnected to Redis");
        },
      });
      PublishService.#isInitialized = true;
    }
    return PublishService.#instance;
  }

  static async initialize() {
    if (!PublishService.#isInitialized) {
      const publisher = PublishService.getInstance();
      await publisher.connect();
    }
  }

  static isServiceInitialized() {
    return PublishService.#isInitialized;
  }
}

module.exports = PublishService;

Using the Publish Service

Once you've created the service, you can use it anywhere in your application:

TypeScript Usage

// controllers/chat.controller.ts
import PublishService from "../services/publish.service";

class ChatController {
  async sendMessage(room: string, message: string) {
    const publisher = PublishService.getInstance();

    const success = await publisher.publishEvent(room, "message", {
      text: message,
      timestamp: new Date().toISOString(),
    });

    if (!success) {
      console.log("Failed to send message");
    }
  }
}

// services/notification.service.ts
import PublishService from "../services/publish.service";

class NotificationService {
  async sendNotification(userId: string, message: string) {
    const publisher = PublishService.getInstance();

    const success = await publisher.publishEvent(
      `user_${userId}`,
      "notification",
      {
        userId,
        message,
        timestamp: new Date().toISOString(),
      }
    );

    if (!success) {
      console.log("Failed to send notification");
    }
  }
}

// app.ts
import PublishService from "./services/publish.service";

async function main() {
  // Initialize the publish service
  await PublishService.initialize();

  // Use the service in your application
  const chatController = new ChatController();
  await chatController.sendMessage("general", "Hello everyone!");

  const notificationService = new NotificationService();
  await notificationService.sendNotification(
    "user123",
    "You have a new message"
  );
}

main().catch(console.error);

CommonJS Usage

// controllers/chat.controller.js
const PublishService = require("../services/publish.service");

class ChatController {
  async sendMessage(room, message) {
    const publisher = PublishService.getInstance();

    const success = await publisher.publishEvent(room, "message", {
      text: message,
      timestamp: new Date().toISOString(),
    });

    if (!success) {
      console.log("Failed to send message");
    }
  }
}

// services/notification.service.js
const PublishService = require("../services/publish.service");

class NotificationService {
  async sendNotification(userId, message) {
    const publisher = PublishService.getInstance();

    const success = await publisher.publishEvent(
      `user_${userId}`,
      "notification",
      {
        userId,
        message,
        timestamp: new Date().toISOString(),
      }
    );

    if (!success) {
      console.log("Failed to send notification");
    }
  }
}

// app.js
const PublishService = require("./services/publish.service");

async function main() {
  // Initialize the publish service
  await PublishService.initialize();

  // Use the service in your application
  const chatController = new ChatController();
  await chatController.sendMessage("general", "Hello everyone!");

  const notificationService = new NotificationService();
  await notificationService.sendNotification(
    "user123",
    "You have a new message"
  );
}

main().catch(console.error);

Benefits of Using a Publish Service

1. Single Connection: Maintains a single Redis connection across your entire application
2. Resource Efficiency: Prevents multiple Redis connections from being created
3. Consistent Configuration: Ensures all parts of your application use the same Redis configuration
4. Easy Access: Can be accessed from anywhere in your application
5. Centralized Error Handling: All Redis operations use the same error handling logic
6. Health Monitoring: Easy to monitor the health of your Redis connection from a single point
7. Automatic Reconnection: The service handles reconnection automatically
8. Thread Safety: The singleton pattern ensures thread-safe access to the Redis connection

Basic Usage

ES Modules (TypeScript/ESM)

// Using named imports
import { RedisPublisher } from "redis-publisher";

async function main() {
  // Create a new publisher instance
  const publisher = new RedisPublisher({
    host: "localhost", // optional, defaults to process.env.REDIS_HOST or 'localhost'
    port: 6379, // optional, defaults to process.env.REDIS_PORT or 6379
    maxRetries: 5, // optional, maximum reconnection attempts
    retryDelay: 5000, // optional, delay between reconnection attempts in ms
  });

  // Connect to Redis
  const connected = await publisher.connect();
  if (!connected) {
    console.log(
      "Failed to connect to Redis, but application continues running"
    );
  }

  // Publish a message to a room
  const success = await publisher.publishEvent("chat_room", "calling", {
    text: "Hello from Redis Publisher!",
  });

  if (!success) {
    console.log("Failed to publish message, but application continues running");
  }

  // Check health status
  if (publisher.isHealthy()) {
    console.log("Publisher is healthy");
  } else {
    console.log(
      "Publisher is not healthy, reconnection attempts:",
      publisher.getReconnectAttempts()
    );
  }

  // Get detailed connection status
  const status = publisher.getConnectionStatus();
  console.log("Connection status:", status);

  // Disconnect when done
  await publisher.disconnect();
}

main().catch(console.error);

CommonJs

// Using destructuring
const { RedisPublisher } = require("redis-publisher");

async function main() {
  // Create a new publisher instance
  const publisher = new RedisPublisher({
    host: "localhost", // optional, defaults to process.env.REDIS_HOST or 'localhost'
    port: 6379, // optional, defaults to process.env.REDIS_PORT or 6379
    maxRetries: 5, // optional, maximum reconnection attempts
    retryDelay: 5000, // optional, delay between reconnection attempts in ms
  });

  // Connect to Redis
  const connected = await publisher.connect();
  if (!connected) {
    console.log(
      "Failed to connect to Redis, but application continues running"
    );
  }

  // Publish a message to a room
  const success = await publisher.publishEvent("chat_room", "calling", {
    text: "Hello from Redis Publisher!",
  });

  if (!success) {
    console.log("Failed to publish message, but application continues running");
  }

  // Check health status
  if (publisher.isHealthy()) {
    console.log("Publisher is healthy");
  } else {
    console.log(
      "Publisher is not healthy, reconnection attempts:",
      publisher.getReconnectAttempts()
    );
  }

  // Get detailed connection status
  const status = publisher.getConnectionStatus();
  console.log("Connection status:", status);

  // Disconnect when done
  await publisher.disconnect();
}

main().catch(console.error);

Advanced Usage with Event Handlers

const publisher = new RedisPublisher({
  onError: (error) => {
    console.error("Redis error occurred:", error);
  },
  onReconnect: () => {
    console.log("Successfully reconnected to Redis");
  },
  onPublishError: (room, error) => {
    console.error(`Failed to publish to room ${room}:`, error);
  },
  onPublishSuccess: (room, message) => {
    console.log(`Successfully published to room ${room}:`, message);
  },
});

Using Environment Variables

Configure Redis connection using environment variables:

REDIS_HOST=your-redis-host
REDIS_PORT=6379
REDIS_PASSWORD=yourpassword

If REDIS_PASSWORD is set, the RedisPublisher will authenticate with Redis using this password.

API

Constructor Options

interface RedisPublisherOptions {
  host?: string;
  port?: number;
  url?: string;
  password?: string;
  maxRetries?: number;
  retryDelay?: number;
  onError?: (error: Error) => void;
  onReconnect?: () => void;
  onPublishError?: (room: string, error: Error) => void;
  onPublishSuccess?: (room: string, message: Message) => void;
}

Methods

• connect(): Promise<boolean> - Connect to Redis, returns true if successful
• disconnect(): Promise<void> - Disconnect from Redis
• publishMessage(room: string, message: Message): Promise<boolean> - Publish a custom message
• publishEvent(room: string, event: string, data: MessageData): Promise<boolean> - Publish an event with data
• isHealthy(): boolean - Check if the publisher is in a healthy state
• getReconnectAttempts(): number - Get the number of reconnection attempts
• getConnectionStatus(): { isConnected: boolean, reconnectAttempts: number, maxRetries: number } - Get detailed connection status

Types

interface MessageData {
  text: string;
  timestamp: string;
  [key: string]: any;
}

interface Message {
  event: string;
  data: MessageData;
}

Building

The package supports both CommonJS and ES modules. The build process generates:

• CommonJS version in dist/cjs/
• ES modules version in dist/esm/
• TypeScript declarations in dist/types/

To build the package:

npm run build

This will:

1. Generate TypeScript declarations
2. Build the ES modules version
3. Build the CommonJS version

Error Handling

The package implements several layers of error handling:

1. Connection Errors:

   • Automatic reconnection attempts
   • Configurable retry strategy
   • Custom error handlers

2. Publishing Errors:

   • Returns boolean instead of throwing errors
   • Custom error handlers for publishing failures
   • Automatic reconnection on failure

3. Graceful Degradation:

   • Application continues running even if Redis is unavailable
   • Health monitoring to track connection status
   • Detailed status reporting

License

ISC