Building Real-time Applications with WebSockets

Learn how to implement real-time features in your applications using WebSockets and Socket.io

Real-time functionality has become essential for modern web applications. Whether you're building collaborative tools, live dashboards, or instant messaging systems, WebSockets provide the foundation for seamless real-time communication. During my work at UN-Habitat building live data monitoring systems for 12 global cities, I discovered the power and complexity of implementing robust real-time architectures.

Why WebSockets Matter for Enterprise Applications

When we started building the UN-Habitat data pipeline dashboard, traditional HTTP requests weren't sufficient. We needed:

• Real-time data updates for 10+ TB of monthly processing
• Live notifications for system alerts across time zones
• Collaborative features for multiple users monitoring the same data
• Sub-second latency for critical infrastructure monitoring

This led us to implement a comprehensive WebSocket architecture that could handle enterprise-scale real-time requirements.

Understanding WebSocket Fundamentals

The WebSocket Handshake

WebSockets start as HTTP requests and upgrade to persistent connections:

// Client-side WebSocket connection
const socket = new WebSocket('wss://your-server.com/data-stream')

socket.onopen = (event) => {
  console.log('Connected to real-time data stream')
  // Send authentication or initialization data
  socket.send(JSON.stringify({
    type: 'auth',
    token: 'your-auth-token',
    cityId: 'nairobi'
  }))
}

socket.onmessage = (event) => {
  const data = JSON.parse(event.data)
  handleRealTimeUpdate(data)
}

socket.onclose = (event) => {
  console.log('Connection closed:', event.code, event.reason)
  // Implement reconnection logic
  setTimeout(() => reconnect(), 1000)
}

socket.onerror = (error) => {
  console.error('WebSocket error:', error)
}

Server-Side Implementation with Node.js

For the UN-Habitat dashboard, we used Node.js with the ws library:

// Server-side WebSocket implementation
const WebSocket = require('ws')
const server = require('http').createServer()
const wss = new WebSocket.Server({ server })

// Track connected clients by city
const cityConnections = new Map()

wss.on('connection', (ws, request) => {
  console.log('New connection from:', request.socket.remoteAddress)
  
  let userCity = null
  let isAuthenticated = false
  
  ws.on('message', async (message) => {
    try {
      const data = JSON.parse(message)
      
      switch (data.type) {
        case 'auth':
          // Verify authentication token
          const user = await authenticateUser(data.token)
          if (user) {
            isAuthenticated = true
            userCity = data.cityId
            
            // Add to city-specific connection pool
            if (!cityConnections.has(userCity)) {
              cityConnections.set(userCity, new Set())
            }
            cityConnections.get(userCity).add(ws)
            
            // Send initial data
            const initialData = await getCityData(userCity)
            ws.send(JSON.stringify({
              type: 'initial_data',
              data: initialData
            }))
          } else {
            ws.send(JSON.stringify({ type: 'auth_error' }))
            ws.close()
          }
          break
          
        case 'subscribe':
          if (isAuthenticated) {
            subscribeToDataStream(userCity, data.filters)
          }
          break
      }
    } catch (error) {
      console.error('Message handling error:', error)
    }
  })
  
  ws.on('close', () => {
    if (userCity && cityConnections.has(userCity)) {
      cityConnections.get(userCity).delete(ws)
    }
  })
})

Real-Time Data Broadcasting

City-Specific Data Streams

Our system needed to broadcast different data to different cities:

// Data broadcasting system
class DataBroadcaster {
  constructor() {
    this.cityConnections = new Map()
    this.dataProcessors = new Map()
  }
  
  // Process incoming data from AWS pipeline
  async processIncomingData(cityId, rawData) {
    try {
      // Process and validate data
      const processedData = await this.processData(rawData)
      
      // Broadcast to all connected clients for this city
      this.broadcastToCityClients(cityId, {
        type: 'data_update',
        timestamp: Date.now(),
        data: processedData
      })
      
      // Check for alerts and notifications
      const alerts = await this.checkForAlerts(processedData)
      if (alerts.length > 0) {
        this.broadcastToCityClients(cityId, {
          type: 'alerts',
          alerts: alerts
        })
      }
      
    } catch (error) {
      console.error('Data processing error:', error)
      this.broadcastToCityClients(cityId, {
        type: 'error',
        message: 'Data processing failed'
      })
    }
  }
  
  broadcastToCityClients(cityId, message) {
    const connections = this.cityConnections.get(cityId)
    if (!connections) return
    
    const messageStr = JSON.stringify(message)
    connections.forEach(ws => {
      if (ws.readyState === WebSocket.OPEN) {
        ws.send(messageStr)
      }
    })
  }
  
  // Integration with AWS EventBridge
  setupAWSIntegration() {
    // Listen for EventBridge events
    eventBridge.on('data-processed', (event) => {
      const { cityId, data } = event.detail
      this.processIncomingData(cityId, data)
    })
  }
}

const broadcaster = new DataBroadcaster()
broadcaster.setupAWSIntegration()

Client-Side Real-Time Components

React Hook for WebSocket Management

// Custom hook for WebSocket connections
import { useEffect, useRef, useState, useCallback } from 'react'

interface UseWebSocketOptions {
  url: string
  onMessage?: (data: any) => void
  onError?: (error: Event) => void
  reconnectInterval?: number
  maxReconnectAttempts?: number
}

export function useWebSocket({
  url,
  onMessage,
  onError,
  reconnectInterval = 1000,
  maxReconnectAttempts = 5
}: UseWebSocketOptions) {
  const [socket, setSocket] = useState<WebSocket | null>(null)
  const [isConnected, setIsConnected] = useState(false)
  const [reconnectCount, setReconnectCount] = useState(0)
  const reconnectTimeoutRef = useRef<NodeJS.Timeout>()
  
  const connect = useCallback(() => {
    try {
      const ws = new WebSocket(url)
      
      ws.onopen = () => {
        setIsConnected(true)
        setReconnectCount(0)
        console.log('WebSocket connected')
      }
      
      ws.onmessage = (event) => {
        try {
          const data = JSON.parse(event.data)
          onMessage?.(data)
        } catch (error) {
          console.error('Failed to parse WebSocket message:', error)
        }
      }
      
      ws.onclose = () => {
        setIsConnected(false)
        setSocket(null)
        
        // Attempt reconnection
        if (reconnectCount < maxReconnectAttempts) {
          reconnectTimeoutRef.current = setTimeout(() => {
            setReconnectCount(prev => prev + 1)
            connect()
          }, reconnectInterval * Math.pow(2, reconnectCount)) // Exponential backoff
        }
      }
      
      ws.onerror = (error) => {
        console.error('WebSocket error:', error)
        onError?.(error)
      }
      
      setSocket(ws)
    } catch (error) {
      console.error('Failed to create WebSocket connection:', error)
    }
  }, [url, onMessage, onError, reconnectCount, maxReconnectAttempts, reconnectInterval])
  
  useEffect(() => {
    connect()
    
    return () => {
      if (reconnectTimeoutRef.current) {
        clearTimeout(reconnectTimeoutRef.current)
      }
      socket?.close()
    }
  }, [connect])
  
  const sendMessage = useCallback((message: any) => {
    if (socket && isConnected) {
      socket.send(JSON.stringify(message))
    }
  }, [socket, isConnected])
  
  return { socket, isConnected, sendMessage, reconnectCount }
}

Real-Time Dashboard Component

// Real-time city dashboard component
interface CityDashboardProps {
  cityId: string
}

export function CityDashboard({ cityId }: CityDashboardProps) {
  const [dashboardData, setDashboardData] = useState(null)
  const [alerts, setAlerts] = useState([])
  const [connectionStatus, setConnectionStatus] = useState('connecting')
  
  const handleWebSocketMessage = useCallback((data: any) => {
    switch (data.type) {
      case 'initial_data':
        setDashboardData(data.data)
        setConnectionStatus('connected')
        break
        
      case 'data_update':
        setDashboardData(prev => ({
          ...prev,
          ...data.data,
          lastUpdated: data.timestamp
        }))
        break
        
      case 'alerts':
        setAlerts(prev => [...prev, ...data.alerts])
        // Show toast notifications for critical alerts
        data.alerts.forEach(alert => {
          if (alert.severity === 'critical') {
            showNotification(alert.message)
          }
        })
        break
        
      case 'error':
        console.error('Server error:', data.message)
        setConnectionStatus('error')
        break
    }
  }, [])
  
  const { isConnected, sendMessage } = useWebSocket({
    url: `wss://your-server.com/city-stream`,
    onMessage: handleWebSocketMessage,
    onError: () => setConnectionStatus('error')
  })
  
  useEffect(() => {
    if (isConnected) {
      // Authenticate and subscribe to city data
      sendMessage({
        type: 'auth',
        token: getAuthToken(),
        cityId
      })
    }
  }, [isConnected, cityId, sendMessage])
  
  return (
    <div className="dashboard">
      <div className="connection-status">
        <StatusIndicator status={connectionStatus} />
        {connectionStatus === 'connected' && (
          <span>Live data • {cityId}</span>
        )}
      </div>
      
      {dashboardData && (
        <div className="dashboard-grid">
          <MetricsCard data={dashboardData.metrics} />
          <ChartsSection data={dashboardData.charts} />
          <AlertsPanel alerts={alerts} />
        </div>
      )}
    </div>
  )
}

Production Considerations

Scaling WebSocket Connections

For enterprise applications serving thousands of users:

// Redis adapter for multi-server scaling
const redis = require('redis')
const { createAdapter } = require('@socket.io/redis-adapter')

// Create Redis clients
const pubClient = redis.createClient({ host: 'localhost', port: 6379 })
const subClient = pubClient.duplicate()

// Use Redis adapter for Socket.IO
io.adapter(createAdapter(pubClient, subClient))

// Horizontal scaling with load balancing
const cluster = require('cluster')
const numCPUs = require('os').cpus().length

if (cluster.isMaster) {
  // Fork workers
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork()
  }
  
  cluster.on('exit', (worker) => {
    console.log(`Worker ${worker.process.pid} died`)
    cluster.fork()
  })
} else {
  // Worker processes
  const server = createServer()
  server.listen(process.env.PORT || 3000)
}

Error Handling and Resilience

// Comprehensive error handling
class ResilientWebSocketManager {
  constructor(url, options = {}) {
    this.url = url
    this.options = {
      maxRetries: 5,
      retryDelay: 1000,
      heartbeatInterval: 30000,
      ...options
    }
    this.retryCount = 0
    this.heartbeatTimer = null
    this.isConnecting = false
  }
  
  connect() {
    if (this.isConnecting) return
    
    this.isConnecting = true
    this.socket = new WebSocket(this.url)
    
    this.socket.onopen = () => {
      this.isConnecting = false
      this.retryCount = 0
      this.startHeartbeat()
      this.onOpen?.()
    }
    
    this.socket.onmessage = (event) => {
      const data = JSON.parse(event.data)
      
      // Handle heartbeat responses
      if (data.type === 'pong') {
        return
      }
      
      this.onMessage?.(data)
    }
    
    this.socket.onclose = () => {
      this.stopHeartbeat()
      this.isConnecting = false
      
      if (this.retryCount < this.options.maxRetries) {
        setTimeout(() => {
          this.retryCount++
          this.connect()
        }, this.options.retryDelay * Math.pow(2, this.retryCount))
      }
    }
  }
  
  startHeartbeat() {
    this.heartbeatTimer = setInterval(() => {
      if (this.socket.readyState === WebSocket.OPEN) {
        this.socket.send(JSON.stringify({ type: 'ping' }))
      }
    }, this.options.heartbeatInterval)
  }
  
  stopHeartbeat() {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer)
      this.heartbeatTimer = null
    }
  }
}

Real-World Performance Results

After implementing our WebSocket architecture at UN-Habitat:

Performance Metrics

• Connection establishment: < 100ms average
• Message latency: < 50ms for city data updates
• Concurrent connections: 500+ users per city
• Uptime: 99.9% availability across 12 cities
• Data throughput: 10GB+ daily real-time processing

System Impact

• User engagement: 300% increase in dashboard usage
• Response time: 80% reduction in incident response
• Data accuracy: Real-time validation prevented 90% of data errors
• Operational efficiency: 60% reduction in manual monitoring tasks

Advanced Patterns

Room-Based Broadcasting

// Organize users into rooms for targeted messaging
class RoomManager {
  constructor() {
    this.rooms = new Map()
  }
  
  joinRoom(socket, roomId, userData) {
    if (!this.rooms.has(roomId)) {
      this.rooms.set(roomId, new Set())
    }
    
    this.rooms.get(roomId).add(socket)
    socket.userData = userData
    socket.currentRoom = roomId
    
    // Notify room members
    this.broadcastToRoom(roomId, {
      type: 'user_joined',
      user: userData
    }, socket)
  }
  
  leaveRoom(socket) {
    if (socket.currentRoom && this.rooms.has(socket.currentRoom)) {
      this.rooms.get(socket.currentRoom).delete(socket)
      
      this.broadcastToRoom(socket.currentRoom, {
        type: 'user_left',
        user: socket.userData
      })
    }
  }
  
  broadcastToRoom(roomId, message, excludeSocket = null) {
    const room = this.rooms.get(roomId)
    if (!room) return
    
    const messageStr = JSON.stringify(message)
    room.forEach(socket => {
      if (socket !== excludeSocket && socket.readyState === WebSocket.OPEN) {
        socket.send(messageStr)
      }
    })
  }
}

Message Queuing for Reliability

// Ensure message delivery with queuing
class MessageQueue {
  constructor() {
    this.queues = new Map()
  }
  
  enqueue(userId, message) {
    if (!this.queues.has(userId)) {
      this.queues.set(userId, [])
    }
    
    this.queues.get(userId).push({
      ...message,
      timestamp: Date.now(),
      id: generateMessageId()
    })
  }
  
  dequeue(userId, messageId) {
    const queue = this.queues.get(userId)
    if (queue) {
      const index = queue.findIndex(msg => msg.id === messageId)
      if (index !== -1) {
        queue.splice(index, 1)
      }
    }
  }
  
  getQueuedMessages(userId) {
    return this.queues.get(userId) || []
  }
  
  clearOldMessages(maxAge = 24 * 60 * 60 * 1000) { // 24 hours
    const cutoff = Date.now() - maxAge
    
    this.queues.forEach((queue, userId) => {
      this.queues.set(userId, queue.filter(msg => msg.timestamp > cutoff))
    })
  }
}

Security Best Practices

Authentication and Authorization

// Secure WebSocket authentication
const jwt = require('jsonwebtoken')

function authenticateWebSocket(socket, token) {
  try {
    const decoded = jwt.verify(token, process.env.JWT_SECRET)
    socket.userId = decoded.userId
    socket.permissions = decoded.permissions
    return true
  } catch (error) {
    return false
  }
}

// Rate limiting for WebSocket messages
const rateLimiter = new Map()

function checkRateLimit(userId, limit = 100, window = 60000) {
  const now = Date.now()
  const userLimits = rateLimiter.get(userId) || { count: 0, resetTime: now + window }
  
  if (now > userLimits.resetTime) {
    userLimits.count = 0
    userLimits.resetTime = now + window
  }
  
  if (userLimits.count >= limit) {
    return false
  }
  
  userLimits.count++
  rateLimiter.set(userId, userLimits)
  return true
}

Conclusion

WebSockets have transformed how we build real-time applications at enterprise scale. The key lessons from implementing the UN-Habitat system:

1. Plan for scale from day one - Design your architecture to handle growth
2. Implement robust error handling - Network issues are inevitable
3. Use proper authentication - Secure your real-time endpoints
4. Monitor performance - Track connection counts, message latency, and errors
5. Test thoroughly - Real-time systems have complex edge cases

Whether you're building collaborative tools, live dashboards, or messaging systems, WebSockets provide the foundation for creating engaging, responsive user experiences that keep users connected and informed in real-time.

---

This article is based on real implementation experience building enterprise-scale real-time systems for UN-Habitat's global urban planning initiatives, serving thousands of users across 12 cities with TB-scale data processing requirements.