Redis Pub/Sub for Real-Time Communication

const redis = require('redis'); // Create publisher client const publisher = redis.createClient({ url: 'redis://localhost:6379' }); // Create subscriber client const subscriber = redis.createClient({ url: 'redis://localhost:6379' }); // Connect both clients await publisher.connect(); await subscriber.connect(); // Subscribe to a channel await subscriber.subscribe('news', (message) => { console.log('Received message:', message); }); // Publish a message await publisher.publish('news', 'Breaking news: Redis is awesome!'); // Output: Received message: Breaking news: Redis is awesome!

// Simple text message await publisher.publish('channel-name', 'Hello World'); // JSON data const data = { userId: 123, action: 'login', timestamp: Date.now() }; await publisher.publish('user-events', JSON.stringify(data)); // Publish returns the number of subscribers who received the message const subscriberCount = await publisher.publish('notifications', 'New message'); console.log(`Message delivered to ${subscriberCount} subscribers`);

// Subscribe to a single channel await subscriber.subscribe('chat', (message) => { console.log('Chat message:', message); }); // Subscribe to multiple channels await subscriber.subscribe('chat', (message, channel) => { console.log(`Message from ${channel}:`, message); }); await subscriber.subscribe('notifications', (message, channel) => { console.log(`Notification from ${channel}:`, message); }); // Unsubscribe from a channel await subscriber.unsubscribe('chat'); // Unsubscribe from all channels await subscriber.unsubscribe();

// Subscribe to all channels starting with "user:" await subscriber.pSubscribe('user:*', (message, channel) => { console.log(`Message from ${channel}:`, message); }); // Matches: user:login, user:logout, user:register, etc. // Subscribe to all chat room channels await subscriber.pSubscribe('chat:room:*', (message, channel) => { const roomId = channel.split(':')[2]; console.log(`Message in room ${roomId}:`, message); }); // Matches: chat:room:1, chat:room:2, chat:room:lobby, etc. // Unsubscribe from patterns await subscriber.pUnsubscribe('user:*');

const express = require('express'); const { Server } = require('socket.io'); const redis = require('redis'); const app = express(); const server = app.listen(3000); const io = new Server(server); // Redis clients const publisher = redis.createClient({ url: 'redis://localhost:6379' }); const subscriber = redis.createClient({ url: 'redis://localhost:6379' }); await publisher.connect(); await subscriber.connect(); // Subscribe to Redis channels await subscriber.subscribe('broadcast', (message) => { const data = JSON.parse(message); // Broadcast to all Socket.io clients io.emit('message', data); }); await subscriber.subscribe('notifications', (message) => { const notification = JSON.parse(message); // Send to specific user io.to(`user-${notification.userId}`).emit('notification', notification); }); // Socket.io connection io.on('connection', (socket) => { socket.on('send_message', (data) => { // Publish to Redis instead of emitting directly publisher.publish('broadcast', JSON.stringify({ userId: socket.userId, message: data.message, timestamp: new Date() })); }); socket.on('join_room', (roomId) => { socket.join(`room-${roomId}`); // Subscribe to room-specific Redis channel subscriber.subscribe(`room:${roomId}`, (message) => { io.to(`room-${roomId}`).emit('room_message', JSON.parse(message)); }); }); });

// Service 1: User Service (users.js) const redis = require('redis'); const publisher = redis.createClient({ url: 'redis://localhost:6379' }); await publisher.connect(); // When user registers async function registerUser(userData) { // Save to database const user = await db.users.create(userData); // Publish event await publisher.publish('user:registered', JSON.stringify({ userId: user.id, email: user.email, name: user.name, timestamp: new Date() })); return user; } // Service 2: Email Service (email.js) const subscriber = redis.createClient({ url: 'redis://localhost:6379' }); await subscriber.connect(); // Listen for user registration events await subscriber.subscribe('user:registered', async (message) => { const user = JSON.parse(message); // Send welcome email await sendWelcomeEmail(user.email, user.name); console.log(`Welcome email sent to ${user.email}`); }); // Service 3: Analytics Service (analytics.js) const analyticsSubscriber = redis.createClient({ url: 'redis://localhost:6379' }); await analyticsSubscriber.connect(); // Listen for all user events await analyticsSubscriber.pSubscribe('user:*', async (message, channel) => { const event = channel.split(':')[1]; // registered, login, logout, etc. const data = JSON.parse(message); // Track event in analytics await trackEvent(event, data); });

// Event publisher utility class EventBus { constructor(redisUrl) { this.publisher = redis.createClient({ url: redisUrl }); } async connect() { await this.publisher.connect(); } async emit(event, data) { const message = JSON.stringify({ event, data, timestamp: new Date(), id: generateUniqueId() }); await this.publisher.publish(event, message); } } // Event subscriber utility class EventListener { constructor(redisUrl) { this.subscriber = redis.createClient({ url: redisUrl }); this.handlers = new Map(); } async connect() { await this.subscriber.connect(); } async on(event, handler) { this.handlers.set(event, handler); await this.subscriber.subscribe(event, async (message) => { const { data } = JSON.parse(message); await handler(data); }); } async onPattern(pattern, handler) { await this.subscriber.pSubscribe(pattern, async (message, channel) => { const { data } = JSON.parse(message); await handler(data, channel); }); } } // Usage const eventBus = new EventBus('redis://localhost:6379'); await eventBus.connect(); const listener = new EventListener('redis://localhost:6379'); await listener.connect(); // Emit events await eventBus.emit('order:created', { orderId: 123, userId: 456 }); await eventBus.emit('order:shipped', { orderId: 123, trackingNumber: 'ABC123' }); // Listen to events await listener.on('order:created', async (data) => { console.log('New order:', data); await processOrder(data.orderId); }); await listener.onPattern('order:*', async (data, channel) => { const action = channel.split(':')[1]; console.log(`Order ${action}:`, data); });

// Messages are lost if: // 1. No subscribers are listening when published // 2. A subscriber disconnects temporarily // 3. Redis server restarts // Check if message was delivered const count = await publisher.publish('channel', 'message'); if (count === 0) { console.log('No subscribers - message was not delivered'); // Consider using Redis Streams or a message queue instead }

// Get active channels const channels = await publisher.pubSubChannels(); console.log('Active channels:', channels); // Get number of subscribers for a channel const numSubs = await publisher.pubSubNumSub('chat'); console.log('Subscribers to chat:', numSubs); // Get number of pattern subscriptions const numPat = await publisher.pubSubNumPat(); console.log('Active pattern subscriptions:', numPat);