JavaScript: Consuming Real-time Updates with Server-Sent Events (SSE)

// Function to establish and manage a Server-Sent Events connection
function setupSSEConnection(eventSourceUrl, onMessageCallback, onErrorCallback, onOpenCallback) {
  if (!window.EventSource) {
    console.error('Your browser does not support Server-Sent Events.');
    if (onErrorCallback) onErrorCallback(new Error('Browser not supported.'));
    return null;
  }

  const eventSource = new EventSource(eventSourceUrl);
  console.log(`Attempting to connect to SSE stream: ${eventSourceUrl}`);

  // Event listener for generic messages (unnamed events)
  eventSource.onmessage = (event) => {
    console.log('Received generic message:', event.data);
    try {
      const parsedData = JSON.parse(event.data);
      if (onMessageCallback) onMessageCallback(parsedData, 'message');
    } catch (e) {
      if (onMessageCallback) onMessageCallback(event.data, 'message'); // Send raw if not JSON
      console.warn('Could not parse SSE data as JSON:', event.data);
    }
  };

  // Event listener for a specific custom event type (e.g., 'new_notification')
  eventSource.addEventListener('new_notification', (event) => {
    console.log('Received custom event "new_notification":', event.data);
    try {
      const notificationData = JSON.parse(event.data);
      if (onMessageCallback) onMessageCallback(notificationData, 'new_notification');
    } catch (e) {
      if (onMessageCallback) onMessageCallback(event.data, 'new_notification');
      console.warn('Could not parse custom event data as JSON:', event.data);
    }
  });

  // Event listener for connection opening
  eventSource.onopen = (event) => {
    console.log('SSE connection opened successfully.');
    if (onOpenCallback) onOpenCallback(event);
  };

  // Event listener for errors
  eventSource.onerror = (event) => {
    console.error('SSE Error:', event);
    if (event.readyState === EventSource.CLOSED) {
      console.log('SSE connection closed unexpectedly.');
    }
    if (onErrorCallback) onErrorCallback(event);
    eventSource.close(); // Close the connection on error to prevent constant retries if unrecoverable
  };

  // Return the EventSource instance to allow manual closing if needed
  return eventSource;
}

// Example Usage:
const SSE_ENDPOINT = 'http://localhost:3001/events'; // Your backend SSE endpoint

// Callback functions for handling different events
const handleSSEMessage = (data, eventType) => {
  console.log(`[${eventType}] Received data:`, data);
  // Example: Update UI with the real-time data
  const outputDiv = document.getElementById('sse-output');
  if (outputDiv) {
    const p = document.createElement('p');
    p.textContent = `[${new Date().toLocaleTimeString()}] ${eventType}: ${JSON.stringify(data)}`;
    outputDiv.prepend(p); // Add new message at the top
  }
};

const handleSSEError = (error) => {
  console.error('SSE connection error:', error);
  // Example: Display an error message to the user
  const statusDiv = document.getElementById('sse-status');
  if (statusDiv) statusDiv.textContent = 'SSE connection failed or closed. Please refresh.';
};

const handleSSEOpen = () => {
  console.log('SSE connection is ready!');
  const statusDiv = document.getElementById('sse-status');
  if (statusDiv) statusDiv.textContent = 'Connected to SSE stream.';
};

// Start the SSE connection
let currentEventSource = null;
document.addEventListener('DOMContentLoaded', () => {
  // Add a simple UI for output and status
  document.body.innerHTML += `
    <div id="sse-container" style="font-family: monospace; max-width: 600px; margin: 20px auto; border: 1px solid #ccc; padding: 15px;">
      <h2>Server-Sent Events Demo</h2>
      <p>Status: <span id="sse-status" style="font-weight: bold; color: blue;">Connecting...</span></p>
      <div id="sse-output" style="max-height: 300px; overflow-y: auto; border: 1px solid #eee; padding: 10px; background-color: #f9f9f9;">
        <!-- Messages will appear here -->
      </div>
      <button id="closeSse" style="margin-top: 10px; padding: 8px 15px; cursor: pointer;">Close SSE Connection</button>
    </div>
  `;

  currentEventSource = setupSSEConnection(
    SSE_ENDPOINT,
    handleSSEMessage,
    handleSSEError,
    handleSSEOpen
  );

  document.getElementById('closeSse').addEventListener('click', () => {
    if (currentEventSource) {
      currentEventSource.close();
      console.log('SSE connection manually closed.');
      const statusDiv = document.getElementById('sse-status');
      if (statusDiv) statusDiv.textContent = 'SSE connection closed by user.';
    }
  });
});

// For a simple Node.js backend to test this snippet:
/*
// backend.js
const express = require('express');
const cors = require('cors'); // Required for cross-origin SSE

const app = express();
const PORT = 3001;

app.use(cors()); // Enable CORS for all routes

app.get('/events', (req, res) => {
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');
    res.setHeader('Access-Control-Allow-Origin', '*'); // Necessary for CORS

    let messageCount = 0;
    const intervalId = setInterval(() => {
        messageCount++;
        // Generic message
        res.write(`data: ${JSON.stringify({ id: messageCount, timestamp: new Date(), message: 'Hello from SSE!' })}

`);

        // Custom event
        if (messageCount % 3 === 0) {
            res.write(`event: new_notification
`);
            res.write(`data: ${JSON.stringify({ id: messageCount, alert: 'Important update!', type: 'info' })}

`);
        }

        if (messageCount >= 10) {
            // Optional: Close connection after some messages
            // res.end();
            // clearInterval(intervalId);
            // console.log('SSE stream ended.');
        }
    }, 3000); // Send an event every 3 seconds

    req.on('close', () => {
        clearInterval(intervalId);
        console.log('Client closed SSE connection.');
    });
});

app.listen(PORT, () => {
    console.log(`Backend SSE server running on http://localhost:${PORT}`);
});
*/