Menjembatani ke Antrean Pesan

Connect WebSockets to Queues

What is a Message Queue?

Why Bridge WebSockets?

Popular Message Brokers

The Bridging Architecture

WS Server as Producer

class MockMessageQueue {
  constructor() {
    this.messages = [];
  }
  publish(queueName, message) {
    console.log(`[MQ] Publishing to '${queueName}': ${message}`);
    this.messages.push({ queueName, message });
  }
}

const mq = new MockMessageQueue();

function onWebSocketMessage(clientMessage) {
  console.log(`[WS] Received from client: ${clientMessage}`);
  // A real server would connect to RabbitMQ/Kafka here
  mq.publish('client_updates', clientMessage);
}

// Simulate a message from a WebSocket client
onWebSocketMessage("User clicked button X");
onWebSocketMessage("User typed 'hello'");

Independent Backend Consumers

class MockMessageQueue {
  constructor() {
    this.listeners = {}; // { queueName: [callback1, callback2] }
  }
  publish(queueName, message) {
    if (this.listeners[queueName]) {
      this.listeners[queueName].forEach(callback => callback(message));
    }
  }
  subscribe(queueName, callback) {
    if (!this.listeners[queueName]) {
      this.listeners[queueName] = [];
    }
    this.listeners[queueName].push(callback);
    console.log(`[MQ] Subscribed to '${queueName}'`);
  }
}

const mq = new MockMessageQueue();

function backendServiceLogic(message) {
  console.log(`[Backend] Processing message: ${message}`);
  // Perform database operations, API calls, etc.
}

// Simulate a separate backend service subscribing
mq.subscribe('client_updates', backendServiceLogic);

// Simulate messages arriving in the queue (from a WS server, for example)
mq.publish('client_updates', "New user registered");
mq.publish('client_updates', "Product added to cart");

Server-to-Client Broadcasts

class MockMessageQueue {
  constructor() {
    this.listeners = {};
    this.messages = {}; // To store published messages for consumers
  }
  publish(queueName, message) {
    if (!this.messages[queueName]) {
      this.messages[queueName] = [];
    }
    this.messages[queueName].push(message);
    if (this.listeners[queueName]) {
      this.listeners[queueName].forEach(callback => callback(message));
    }
  }
  subscribe(queueName, callback) {
    if (!this.listeners[queueName]) {
      this.listeners[queueName] = [];
    }
    this.listeners[queueName].push(callback);
  }
}

const mq = new MockMessageQueue();

// --- WebSocket Server Component ---
const connectedClients = []; // Simulate connected WebSocket clients
function sendToAllClients(message) {
  console.log(`[WS Server] Broadcasting to ${connectedClients.length} clients: ${message}`);
  // In a real app, iterate through connectedClients and send
}

// WS Server subscribes to queue for messages to broadcast
mq.subscribe('server_broadcasts', sendToAllClients);

// Simulate a client connecting
connectedClients.push("client1");
connectedClients.push("client2");

// --- Backend Service Component ---
function processNewOrder(orderId) {
  console.log(`[Backend] Order ${orderId} processed.`);
  const notification = `New order #${orderId} confirmed!`;
  // Backend publishes to queue, WS server will pick it up
  mq.publish('server_broadcasts', notification);
}

// Simulate a new order event in the backend
processNewOrder(1001);
processNewOrder(1002);

Real-world Bridging Use Cases

Bridging Knowledge Check

Recap: Queues & WebSockets