Building Real-Time Features Server-Sent Events

Server-Sent Events (SSE) provide an efficient way to implement real-time features with one-way server-to-client communication. Let's explore how to build robust real-time features using SSE.

Understanding Server-Sent Events

SSE offers several advantages for real-time communication:

• Automatic reconnection
• Event filtering
• Native browser support
• Efficient one-way communication
• Simple implementation

Basic Implementation

1. Server Setup

Create a basic SSE server with Node.js:

import express from 'express';

const app = express();

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

  // Send initial connection message
  res.write('data: Connected to event stream\n\n');

  // Keep connection alive
  const intervalId = setInterval(() => {
    res.write('data: heartbeat\n\n');
  }, 15000);

  // Clean up on client disconnect
  req.on('close', () => {
    clearInterval(intervalId);
  });
});

app.listen(3000, () => {
  console.log('SSE server running on port 3000');
});

2. Client Implementation

Implement the client-side EventSource:

class EventSourceWrapper {
  constructor(url) {
    this.url = url;
    this.eventSource = null;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 5;
  }

  connect() {
    this.eventSource = new EventSource(this.url);

    this.eventSource.onopen = () => {
      console.log('Connected to event source');
      this.reconnectAttempts = 0;
    };

    this.eventSource.onerror = () => {
      this.handleError();
    };

    return this.eventSource;
  }

  handleError() {
    if (this.reconnectAttempts < this.maxReconnectAttempts) {
      this.reconnectAttempts++;
      setTimeout(() => this.connect(), 1000 * this.reconnectAttempts);
    }
  }
}

Advanced Features

1. Event Types

Implement custom event types:

// Server
function sendEvent(res, event, data) {
  res.write(`event: ${event}\n`);
  res.write(`data: ${JSON.stringify(data)}\n\n`);
}

app.get('/events', (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream');

  // Send different types of events
  sendEvent(res, 'userJoined', { 
    userId: 123, 
    username: 'john' 
  });

  sendEvent(res, 'message', {
    text: 'Hello world',
    timestamp: Date.now()
  });
});

// Client
const events = new EventSourceWrapper('/events');
const source = events.connect();

source.addEventListener('userJoined', (e) => {
  const data = JSON.parse(e.data);
  console.log(`User ${data.username} joined`);
});

source.addEventListener('message', (e) => {
  const data = JSON.parse(e.data);
  console.log(`New message: ${data.text}`);
});

2. Authentication

Implement secure SSE connections:

// Server
app.get('/events', authenticateUser, (req, res) => {
  const userId = req.user.id;

  res.setHeader('Content-Type', 'text/event-stream');

  // Send user-specific events
  const userEvents = getUserEvents(userId);
  userEvents.on('update', (data) => {
    sendEvent(res, 'userUpdate', data);
  });

  req.on('close', () => {
    userEvents.removeAllListeners();
  });
});

// Client
class AuthenticatedEventSource {
  constructor(url, token) {
    this.url = new URL(url);
    this.url.searchParams.append('token', token);
  }

  connect() {
    return new EventSource(this.url.toString());
  }
}

Performance Optimization

1. Event Batching

Implement event batching for efficiency:

class EventBatcher {
  constructor(interval = 1000) {
    this.events = [];
    this.interval = interval;
  }

  addEvent(event) {
    this.events.push(event);
  }

  async flush(res) {
    if (this.events.length === 0) return;

    const batch = this.events.splice(0);
    sendEvent(res, 'batch', batch);
  }

  start(res) {
    this.intervalId = setInterval(
      () => this.flush(res),
      this.interval
    );
  }

  stop() {
    clearInterval(this.intervalId);
  }
}

2. Connection Management

Implement connection pooling:

class ConnectionPool {
  constructor() {
    this.connections = new Map();
  }

  add(userId, res) {
    this.connections.set(userId, res);

    res.on('close', () => {
      this.connections.delete(userId);
    });
  }

  broadcast(event, data) {
    for (const res of this.connections.values()) {
      sendEvent(res, event, data);
    }
  }

  sendToUser(userId, event, data) {
    const res = this.connections.get(userId);
    if (res) {
      sendEvent(res, event, data);
    }
  }
}

Error Handling

1. Retry Logic

Implement robust retry handling:

class RetryHandler {
  constructor(maxAttempts = 5) {
    this.attempts = 0;
    this.maxAttempts = maxAttempts;
  }

  shouldRetry() {
    return this.attempts < this.maxAttempts;
  }

  getDelay() {
    // Exponential backoff
    return Math.min(1000 * Math.pow(2, this.attempts), 30000);
  }

  async retry(callback) {
    while (this.shouldRetry()) {
      try {
        return await callback();
      } catch (error) {
        this.attempts++;
        if (!this.shouldRetry()) throw error;
        await new Promise(resolve => 
          setTimeout(resolve, this.getDelay())
        );
      }
    }
  }
}

2. Error Recovery

Implement error recovery mechanisms:

class ErrorRecovery {
  constructor(eventSource) {
    this.eventSource = eventSource;
    this.lastEventId = null;
  }

  handleError() {
    if (this.lastEventId) {
      const url = new URL(this.eventSource.url);
      url.searchParams.set('lastEventId', this.lastEventId);

      this.eventSource = new EventSource(url.toString());
      this.setupEventHandlers();
    }
  }

  setupEventHandlers() {
    this.eventSource.addEventListener('message', (e) => {
      this.lastEventId = e.lastEventId;
    });

    this.eventSource.onerror = () => this.handleError();
  }
}

Monitoring and Debugging

1. Connection Monitoring

Implement connection monitoring:

class ConnectionMonitor {
  constructor() {
    this.metrics = {
      activeConnections: 0,
      totalConnections: 0,
      errors: 0
    };
  }

  trackConnection() {
    this.metrics.activeConnections++;
    this.metrics.totalConnections++;
  }

  trackDisconnection() {
    this.metrics.activeConnections--;
  }

  trackError() {
    this.metrics.errors++;
  }

  getMetrics() {
    return { ...this.metrics };
  }
}

2. Debug Logging

Implement comprehensive logging:

class EventLogger {
  constructor() {
    this.logs = [];
  }

  log(event, data) {
    this.logs.push({
      timestamp: new Date(),
      event,
      data
    });
  }

  getLogs(filter = {}) {
    return this.logs.filter(log => {
      for (const [key, value] of Object.entries(filter)) {
        if (log[key] !== value) return false;
      }
      return true;
    });
  }
}

Best Practices

1. Connection Management

• Implement heartbeats
• Handle reconnection
• Monitor connections
• Clean up resources

1. Error Handling

• Implement retry logic
• Handle disconnections
• Log errors
• Recover gracefully

1. Performance

• Batch events
• Optimize payload size
• Monitor memory usage
• Handle backpressure

Conclusion

Server-Sent Events provide an efficient way to implement real-time features:

1. Benefits

• Simple implementation
• Native browser support
• Automatic reconnection
• Efficient communication

1. Implementation Tips

• Handle errors properly
• Implement monitoring
• Optimize performance
• Maintain security

1. Best Practices

• Keep connections alive
• Implement retry logic
• Monitor performance
• Handle disconnections

Remember to:

• Test thoroughly
• Monitor connections
• Handle errors gracefully
• Optimize performance