Real-Time JSON Streaming with WebSockets: Complete Guide 2026

# Real-Time JSON Streaming with WebSockets: Complete Guide 2026

Real-time data streaming powers modern applications like live dashboards, chat, collaborative editing, and financial tickers. This guide covers WebSockets, Server-Sent Events, and best practices for streaming JSON data in production.

Table of Contents

---

WebSockets vs SSE vs HTTP Polling

Technology Comparison

| Feature | WebSockets | Server-Sent Events | HTTP Polling |

|---------|-----------|-------------------|--------------|

| Bidirectional | ✅ Yes | ❌ No (server→client) | ✅ Yes |

| Browser Support | 99%+ | 95%+ | 100% |

| Connection | Persistent | Persistent | Repeated |

| Protocol | ws:// or wss:// | HTTP/HTTPS | HTTP/HTTPS |

| Overhead | Low | Low | High |

| Text & Binary | Both | Text only | Both |

| Auto Reconnect | Manual | Automatic | Manual |

| Proxies/Firewalls | Sometimes blocked | Works everywhere | Works everywhere |

When to Use Each

• Chat applications
• Multiplayer games
• Collaborative editing (Google Docs-style)
• Trading platforms with bidirectional updates

• Live notifications
• Stock tickers (server pushes only)
• News feeds
• Server logs streaming

• Fallback when WebSockets unavailable
• Low-frequency updates (every 30+ seconds)
• Simple implementations with no special infrastructure

---

WebSocket Implementation

Basic WebSocket Server (Node.js)

const WebSocket = require('ws');

const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', (ws) => {
  console.log('Client connected');

  // Send welcome message
  ws.send(JSON.stringify({
    type: 'welcome',
    message: 'Connected to server',
    timestamp: Date.now()
  }));

  // Handle messages
  ws.on('message', (data) => {
    const message = JSON.parse(data);
    console.log('Received:', message);

    // Echo back
    ws.send(JSON.stringify({
      type: 'response',
      data: message,
      timestamp: Date.now()
    }));
  });

  // Handle close
  ws.on('close', () => {
    console.log('Client disconnected');
  });

  // Handle errors
  ws.on('error', (error) => {
    console.error('WebSocket error:', error);
  });
});

console.log('WebSocket server running on ws://localhost:8080');

WebSocket Client (Browser)

const ws = new WebSocket('ws://localhost:8080');

ws.onopen = () => {
  console.log('Connected to server');

  // Send JSON message
  ws.send(JSON.stringify({
    type: 'subscribe',
    channel: 'updates'
  }));
};

ws.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Received:', data);

  switch(data.type) {
    case 'welcome':
      console.log(data.message);
      break;
    case 'update':
      updateUI(data);
      break;
  }
};

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

ws.onclose = () => {
  console.log('Disconnected from server');
  // Implement reconnection logic
  setTimeout(() => {
    connectWebSocket(); // Retry connection
  }, 5000);
};

Broadcasting to All Clients

function broadcast(data) {
  const message = JSON.stringify(data);

  wss.clients.forEach((client) => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(message);
    }
  });
}

// Example: Broadcast stock price update
setInterval(() => {
  broadcast({
    type: 'stock_update',
    symbol: 'AAPL',
    price: 150 + Math.random()  10,

    timestamp: Date.now()
  });
}, 1000);

const rooms = new Map();

wss.on('connection', (ws) => {
  ws.on('message', (data) => {
    const message = JSON.parse(data);

    if (message.type === 'join') {
      const room = message.room;

      if (!rooms.has(room)) {
        rooms.set(room, new Set());
      }
      rooms.get(room).add(ws);
      ws.room = room;

      console.log(Client joined room: ${room});
    } else if (message.type === 'chat') {
      broadcastToRoom(ws.room, {
        type: 'chat',
        user: message.user,
        text: message.text,
        timestamp: Date.now()
      });
    }
  });

  ws.on('close', () => {
    if (ws.room && rooms.has(ws.room)) {
      rooms.get(ws.room).delete(ws);
    }
  });
});

function broadcastToRoom(room, data) {
  const message = JSON.stringify(data);

  if (rooms.has(room)) {
    rooms.get(room).forEach((client) => {
      if (client.readyState === WebSocket.OPEN) {
        client.send(message);
      }
    });
  }
}

const express = require('express');
const app = express();

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

  // Send initial message
  res.write(data: ${JSON.stringify({ type: 'connected', time: Date.now() })}\n\n);

  // Send updates every 2 seconds
  const interval = setInterval(() => {
    const data = {
      type: 'update',
      value: Math.random()  100,
      timestamp: Date.now()
    };

    res.write(data: ${JSON.stringify(data)}\n\n);
  }, 2000);

  // Clean up on client disconnect
  req.on('close', () => {
    clearInterval(interval);
    console.log('Client disconnected');
  });
});

app.listen(3000, () => {
  console.log('SSE server running on http://localhost:3000');
});

SSE Client Implementation

const eventSource = new EventSource('http://localhost:3000/events');

eventSource.onopen = () => {
  console.log('SSE connection established');
};

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Received:', data);

  // Update UI with new data
  updateChart(data);
};

eventSource.onerror = (error) => {
  console.error('SSE error:', error);

  if (eventSource.readyState === EventSource.CLOSED) {
    console.log('SSE connection closed, reconnecting...');
    // EventSource auto-reconnects by default
  }
};

// Close connection when needed
// eventSource.close();

Named Events with SSE

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

  setInterval(() => {
    // Named event
    res.write(event: price\n);
    res.write(data: ${JSON.stringify({ symbol: 'BTC', price: 50000 })}\n\n);
  }, 1000);

  setInterval(() => {
    // Different named event
    res.write(event: alert\n);
    res.write(data: ${JSON.stringify({ message: 'Price spike!' })}\n\n);
  }, 5000);
});

// Client
const eventSource = new EventSource('/events');

eventSource.addEventListener('price', (event) => {
  const data = JSON.parse(event.data);
  updatePrice(data);
});

eventSource.addEventListener('alert', (event) => {
  const data = JSON.parse(event.data);
  showAlert(data.message);
});

---

JSON Streaming Patterns

Chunked Transfer Encoding

// Stream large JSON array in chunks
app.get('/api/users-stream', async (req, res) => {
  res.setHeader('Content-Type', 'application/json');
  res.setHeader('Transfer-Encoding', 'chunked');

  res.write('['); // Start array

  let first = true;
  const users = await db.users.stream();

  for await (const user of users) {
    if (!first) res.write(',');
    res.write(JSON.stringify(user));
    first = false;
  }

  res.write(']'); // Close array
  res.end();
});

NDJSON (Newline Delimited JSON)

// Server: Stream users as NDJSON
app.get('/api/users.ndjson', async (req, res) => {
  res.setHeader('Content-Type', 'application/x-ndjson');

  const users = await db.users.stream();

  for await (const user of users) {
    res.write(JSON.stringify(user) + '\n');
  }

  res.end();
});

// Client: Parse NDJSON stream
async function fetchNDJSON(url) {
  const response = await fetch(url);
  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });

    const lines = buffer.split('\n');
    buffer = lines.pop(); // Keep incomplete line in buffer

    for (const line of lines) {
      if (line.trim()) {
        const data = JSON.parse(line);
        processUser(data);
      }
    }
  }
}

---

Error Handling and Reconnection

Robust WebSocket Client

class RobustWebSocket {
  constructor(url, options = {}) {
    this.url = url;
    this.reconnectInterval = options.reconnectInterval || 5000;
    this.maxReconnectAttempts = options.maxReconnectAttempts || 10;
    this.reconnectAttempts = 0;
    this.handlers = {};

    this.connect();
  }

  connect() {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      console.log('WebSocket connected');
      this.reconnectAttempts = 0;
      this.trigger('open');
    };

    this.ws.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        this.trigger('message', data);
      } catch (error) {
        console.error('Invalid JSON received:', error);
      }
    };

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
      this.trigger('error', error);
    };

    this.ws.onclose = () => {
      console.log('WebSocket closed');
      this.trigger('close');
      this.reconnect();
    };
  }

  reconnect() {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.error('Max reconnection attempts reached');
      this.trigger('maxReconnectError');
      return;
    }

    this.reconnectAttempts++;
    console.log(Reconnecting (attempt ${this.reconnectAttempts})...);

    setTimeout(() => {
      this.connect();
    }, this.reconnectInterval);
  }

  send(data) {
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data));
    } else {
      console.warn('WebSocket not connected, message not sent');
    }
  }

  on(event, handler) {
    if (!this.handlers[event]) {
      this.handlers[event] = [];
    }
    this.handlers[event].push(handler);
  }

  trigger(event, data) {
    if (this.handlers[event]) {
      this.handlers[event].forEach(handler => handler(data));
    }
  }

  close() {
    this.maxReconnectAttempts = 0; // Prevent reconnection
    this.ws.close();
  }
}

// Usage
const ws = new RobustWebSocket('ws://localhost:8080', {
  reconnectInterval: 3000,
  maxReconnectAttempts: 5
});

ws.on('open', () => {
  ws.send({ type: 'subscribe', channel: 'updates' });
});

ws.on('message', (data) => {
  console.log('Received:', data);
});

---

Scalability and Load Balancing

Redis Pub/Sub for Multi-Server Scaling

const Redis = require('ioredis');
const WebSocket = require('ws');

const pub = new Redis();
const sub = new Redis();

const wss = new WebSocket.Server({ port: 8080 });

// Subscribe to Redis channel
sub.subscribe('updates');

sub.on('message', (channel, message) => {
  // Broadcast to all connected clients on this server
  wss.clients.forEach((client) => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(message);
    }
  });
});

// When server receives message, publish to Redis
wss.on('connection', (ws) => {
  ws.on('message', (data) => {
    // Publish to all servers via Redis
    pub.publish('updates', data);
  });
});

Sticky Sessions with Load Balancer

# Nginx configuration for WebSocket load balancing
upstream websocket {
    ip_hash; # Sticky sessions
    server 10.0.0.1:8080;
    server 10.0.0.2:8080;
    server 10.0.0.3:8080;
}

server {
    listen 80;

    location /ws {
        proxy_pass http://websocket;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
    }
}

---

Production Best Practices

Heartbeat / Ping-Pong

// Server
wss.on('connection', (ws) => {
  ws.isAlive = true;

  ws.on('pong', () => {
    ws.isAlive = true;
  });
});

// Ping all clients every 30 seconds
setInterval(() => {
  wss.clients.forEach((ws) => {
    if (ws.isAlive === false) {
      return ws.terminate(); // Close dead connections
    }

    ws.isAlive = false;
    ws.ping();
  });
}, 30000);

Rate Limiting

const rateLimitMap = new Map();

wss.on('connection', (ws, req) => {
  const ip = req.socket.remoteAddress;

  ws.on('message', (data) => {
    const now = Date.now();
    const clientData = rateLimitMap.get(ip) || { count: 0, resetTime: now + 60000 };

    if (now > clientData.resetTime) {
      clientData.count = 0;
      clientData.resetTime = now + 60000;
    }

    clientData.count++;

    if (clientData.count > 100) { // Max 100 messages per minute
      ws.send(JSON.stringify({
        type: 'error',
        message: 'Rate limit exceeded'
      }));
      return;
    }

    rateLimitMap.set(ip, clientData);

    // Process message
    handleMessage(data);
  });
});

Message Validation

const Ajv = require('ajv');
const ajv = new Ajv();

const messageSchema = {
  type: 'object',
  properties: {
    type: { type: 'string', enum: ['subscribe', 'unsubscribe', 'message'] },
    data: { type: 'object' }
  },
  required: ['type']
};

const validate = ajv.compile(messageSchema);

ws.on('message', (data) => {
  let message;

  try {
    message = JSON.parse(data);
  } catch (error) {
    ws.send(JSON.stringify({ error: 'Invalid JSON' }));
    return;
  }

  if (!validate(message)) {
    ws.send(JSON.stringify({ error: 'Invalid message format', details: validate.errors }));
    return;
  }

  // Process valid message
  handleMessage(message);
});

---

Conclusion

Real-time JSON streaming transforms user experiences with instant updates. Key takeaways:

• WebSockets for bidirectional, low-latency apps
• SSE for server-push scenarios
• HTTP polling as fallback

• Implement reconnection logic
• Use heartbeats to detect dead connections
• Scale with Redis Pub/Sub
• Rate limit to prevent abuse
• Validate all incoming messages

Start with simple implementations, then add robustness as your application scales.

Related Resources

• JSON in Node.js
• JSON APIs Guide
• Large JSON Files