Live Streaming

Live Streaming

Current Level: Advanced
Domain: Video Streaming / Media

Overview

Live streaming enables real-time video broadcasting. This guide covers RTMP, HLS, WebRTC, streaming servers, and cloud platforms for building scalable live streaming solutions with appropriate latency and quality.

Core Concepts

Live Streaming Protocols

NGINX-RTMP Server Setup

# nginx.conf
rtmp {
    server {
        listen 1935;
        chunk_size 4096;

        application live {
            live on;
            record off;

            # HLS
            hls on;
            hls_path /tmp/hls;
            hls_fragment 3;
            hls_playlist_length 60;

            # DASH
            dash on;
            dash_path /tmp/dash;

            # Allow publish from localhost only
            allow publish 127.0.0.1;
            deny publish all;

            # Allow play from anywhere
            allow play all;

            # Webhook on publish
            on_publish http://localhost:3000/api/stream/publish;
            on_publish_done http://localhost:3000/api/stream/done;
        }

        application recording {
            live on;
            record all;
            record_path /var/recordings;
            record_suffix -%Y-%m-%d-%H-%M-%S.flv;
        }
    }
}

http {
    server {
        listen 8080;

        location /hls {
            types {
                application/vnd.apple.mpegurl m3u8;
                video/mp2t ts;
            }
            root /tmp;
            add_header Cache-Control no-cache;
            add_header Access-Control-Allow-Origin *;
        }

        location /dash {
            root /tmp;
            add_header Cache-Control no-cache;
            add_header Access-Control-Allow-Origin *;
        }

        location /stat {
            rtmp_stat all;
            rtmp_stat_stylesheet stat.xsl;
        }
    }
}

OBS Integration

// services/stream-key.service.ts
export class StreamKeyService {
  async generateStreamKey(userId: string): Promise<StreamKey> {
    const key = crypto.randomBytes(16).toString('hex');

    const streamKey = await db.streamKey.create({
      data: {
        userId,
        key,
        expiresAt: new Date(Date.now() + 24 * 60 * 60 * 1000) // 24 hours
      }
    });

    return streamKey;
  }

  async validateStreamKey(key: string): Promise<boolean> {
    const streamKey = await db.streamKey.findUnique({
      where: { key }
    });

    if (!streamKey) return false;

    if (streamKey.expiresAt < new Date()) {
      return false;
    }

    return true;
  }

  getStreamUrl(streamKey: string): StreamUrls {
    return {
      rtmpUrl: `rtmp://${process.env.STREAM_SERVER}/live`,
      streamKey,
      playbackUrl: `https://${process.env.CDN_DOMAIN}/hls/${streamKey}.m3u8`
    };
  }
}

interface StreamUrls {
  rtmpUrl: string;
  streamKey: string;
  playbackUrl: string;
}

// OBS Settings for users
const obsSettings = {
  server: 'rtmp://stream.example.com/live',
  streamKey: 'user-stream-key-here',
  encoder: 'x264',
  bitrate: 2500, // kbps
  keyframeInterval: 2,
  preset: 'veryfast',
  profile: 'main',
  tune: 'zerolatency'
};

WebRTC for Low-Latency

// services/webrtc-stream.service.ts
import { RTCPeerConnection, RTCSessionDescription } from 'wrtc';

export class WebRTCStreamService {
  private peerConnections = new Map<string, RTCPeerConnection>();

  async createOffer(streamId: string): Promise<RTCSessionDescriptionInit> {
    const pc = new RTCPeerConnection({
      iceServers: [
        { urls: 'stun:stun.l.google.com:19302' },
        {
          urls: 'turn:turn.example.com:3478',
          username: process.env.TURN_USERNAME,
          credential: process.env.TURN_PASSWORD
        }
      ]
    });

    this.peerConnections.set(streamId, pc);

    // Add transceivers for video and audio
    pc.addTransceiver('video', { direction: 'recvonly' });
    pc.addTransceiver('audio', { direction: 'recvonly' });

    const offer = await pc.createOffer();
    await pc.setLocalDescription(offer);

    return offer;
  }

  async handleAnswer(streamId: string, answer: RTCSessionDescriptionInit): Promise<void> {
    const pc = this.peerConnections.get(streamId);
    if (!pc) throw new Error('Peer connection not found');

    await pc.setRemoteDescription(new RTCSessionDescription(answer));
  }

  async addIceCandidate(streamId: string, candidate: RTCIceCandidateInit): Promise<void> {
    const pc = this.peerConnections.get(streamId);
    if (!pc) throw new Error('Peer connection not found');

    await pc.addIceCandidate(candidate);
  }

  closePeerConnection(streamId: string): void {
    const pc = this.peerConnections.get(streamId);
    if (pc) {
      pc.close();
      this.peerConnections.delete(streamId);
    }
  }
}

// Client-side WebRTC player
async function playWebRTCStream(streamId: string) {
  const pc = new RTCPeerConnection({
    iceServers: [{ urls: 'stun:stun.l.google.com:19302' }]
  });

  pc.ontrack = (event) => {
    const video = document.getElementById('video') as HTMLVideoElement;
    video.srcObject = event.streams[0];
  };

  // Get offer from server
  const response = await fetch(`/api/stream/${streamId}/offer`, {
    method: 'POST'
  });
  const { offer } = await response.json();

  await pc.setRemoteDescription(offer);
  const answer = await pc.createAnswer();
  await pc.setLocalDescription(answer);

  // Send answer to server
  await fetch(`/api/stream/${streamId}/answer`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ answer })
  });
}

Chat Integration

// services/stream-chat.service.ts
import { Server as SocketIOServer } from 'socket.io';

export class StreamChatService {
  private io: SocketIOServer;

  constructor(io: SocketIOServer) {
    this.io = io;
    this.setupHandlers();
  }

  private setupHandlers(): void {
    this.io.on('connection', (socket) => {
      socket.on('join-stream', async (streamId: string) => {
        socket.join(`stream:${streamId}`);

        // Increment viewer count
        await this.incrementViewerCount(streamId);

        // Broadcast viewer count
        this.broadcastViewerCount(streamId);
      });

      socket.on('leave-stream', async (streamId: string) => {
        socket.leave(`stream:${streamId}`);

        await this.decrementViewerCount(streamId);
        this.broadcastViewerCount(streamId);
      });

      socket.on('chat-message', async (data: ChatMessage) => {
        // Save message
        await db.chatMessage.create({
          data: {
            streamId: data.streamId,
            userId: data.userId,
            message: data.message
          }
        });

        // Broadcast to all viewers
        this.io.to(`stream:${data.streamId}`).emit('chat-message', {
          username: data.username,
          message: data.message,
          timestamp: Date.now()
        });
      });

      socket.on('disconnect', () => {
        // Handle disconnect
      });
    });
  }

  private async incrementViewerCount(streamId: string): Promise<void> {
    await db.stream.update({
      where: { id: streamId },
      data: { viewerCount: { increment: 1 } }
    });
  }

  private async decrementViewerCount(streamId: string): Promise<void> {
    await db.stream.update({
      where: { id: streamId },
      data: { viewerCount: { decrement: 1 } }
    });
  }

  private async broadcastViewerCount(streamId: string): Promise<void> {
    const stream = await db.stream.findUnique({ where: { id: streamId } });
    
    this.io.to(`stream:${streamId}`).emit('viewer-count', {
      count: stream?.viewerCount || 0
    });
  }
}

interface ChatMessage {
  streamId: string;
  userId: string;
  username: string;
  message: string;
}

Recording Live Streams

// services/stream-recording.service.ts
import ffmpeg from 'fluent-ffmpeg';

export class StreamRecordingService {
  async startRecording(streamId: string, hlsUrl: string): Promise<void> {
    const outputPath = `/recordings/${streamId}-${Date.now()}.mp4`;

    ffmpeg(hlsUrl)
      .outputOptions([
        '-c copy',
        '-bsf:a aac_adtstoasc'
      ])
      .output(outputPath)
      .on('start', () => {
        console.log('Recording started:', streamId);
      })
      .on('end', async () => {
        console.log('Recording ended:', streamId);
        await this.uploadRecording(streamId, outputPath);
      })
      .on('error', (err) => {
        console.error('Recording error:', err);
      })
      .run();

    await db.stream.update({
      where: { id: streamId },
      data: { recording: true }
    });
  }

  private async uploadRecording(streamId: string, filePath: string): Promise<void> {
    // Upload to S3
    const s3Key = `recordings/${streamId}.mp4`;
    await uploadToS3(filePath, s3Key, 'video/mp4');

    // Save to database
    await db.recording.create({
      data: {
        streamId,
        s3Key,
        url: `${process.env.CDN_URL}/${s3Key}`
      }
    });
  }
}

AWS IVS (Interactive Video Service)

// services/aws-ivs.service.ts
import { IVSClient, CreateChannelCommand, GetStreamCommand } from '@aws-sdk/client-ivs';

export class AWSIVSService {
  private client: IVSClient;

  constructor() {
    this.client = new IVSClient({ region: process.env.AWS_REGION! });
  }

  async createChannel(name: string): Promise<IVSChannel> {
    const command = new CreateChannelCommand({
      name,
      latencyMode: 'LOW', // or 'NORMAL'
      type: 'STANDARD',
      authorized: true
    });

    const response = await this.client.send(command);

    return {
      arn: response.channel?.arn!,
      ingestEndpoint: response.channel?.ingestEndpoint!,
      playbackUrl: response.channel?.playbackUrl!,
      streamKey: response.streamKey?.value!
    };
  }

  async getStreamStatus(channelArn: string): Promise<StreamStatus> {
    const command = new GetStreamCommand({
      channelArn
    });

    const response = await this.client.send(command);

    return {
      state: response.stream?.state!,
      health: response.stream?.health!,
      viewerCount: response.stream?.viewerCount || 0
    };
  }
}

interface IVSChannel {
  arn: string;
  ingestEndpoint: string;
  playbackUrl: string;
  streamKey: string;
}

interface StreamStatus {
  state: string;
  health: string;
  viewerCount: number;
}

Adaptive Streaming

# FFmpeg command for adaptive streaming
ffmpeg -i input.mp4 \
  -c:v libx264 -c:a aac \
  -b:v:0 5000k -s:v:0 1920x1080 -profile:v:0 high \
  -b:v:1 3000k -s:v:1 1280x720 -profile:v:1 main \
  -b:v:2 1500k -s:v:2 854x480 -profile:v:2 main \
  -b:v:3 800k -s:v:3 640x360 -profile:v:3 baseline \
  -b:a 128k \
  -var_stream_map "v:0,a:0 v:1,a:0 v:2,a:0 v:3,a:0" \
  -master_pl_name master.m3u8 \
  -f hls -hls_time 6 -hls_list_size 0 \
  -hls_segment_filename "stream_%v/segment_%03d.ts" \
  stream_%v/playlist.m3u8

Best Practices

1. Low Latency - Use WebRTC for <1s latency
2. Scalability - Use CDN for distribution
3. Recording - Record all streams
4. Chat - Integrate real-time chat
5. Monitoring - Monitor stream health
6. Authentication - Secure stream keys
7. Adaptive - Support adaptive bitrate
8. Backup - Have failover streams
9. Analytics - Track viewer metrics
10. Compliance - Follow content policies

Quick Start

Basic RTMP Stream

# Stream to RTMP server
ffmpeg -re -i input.mp4 \
  -c:v libx264 -preset veryfast -maxrate 3000k \
  -bufsize 6000k -pix_fmt yuv420p -g 50 \
  -c:a aac -b:a 160k -ac 2 -ar 44100 \
  -f flv rtmp://server/live/stream_key

HLS Playback

<!-- HLS.js for browser playback -->
<script src="https://cdn.jsdelivr.net/npm/hls.js@latest"></script>
<video id="video"></video>

<script>
  const video = document.getElementById('video')
  const hls = new Hls()
  hls.loadSource('https://server/stream.m3u8')
  hls.attachMedia(video)
</script>

Production Checklist

• Streaming Protocol: Choose protocol (RTMP, HLS, WebRTC)
• Streaming Server: Set up streaming server (NGINX-RTMP, Wowza)
• CDN: Use CDN for global distribution
• Adaptive Bitrate: Implement adaptive bitrate streaming
• Authentication: Secure stream keys
• Monitoring: Monitor stream health and viewer metrics
• Backup: Failover streams for reliability
• Analytics: Track viewer metrics
• Compliance: Follow content policies
• Testing: Test with various devices and networks
• Error Handling: Handle stream failures gracefully
• Documentation: Document streaming setup

Anti-patterns

❌ Don't: No Authentication

# ❌ Bad - Public stream key
rtmp://server/live/public_key  # Anyone can stream!

# ✅ Good - Secure stream key
rtmp://server/live/secure-random-key-12345
# Rotate keys regularly

❌ Don't: Single Bitrate

# ❌ Bad - One quality only
ffmpeg -i input.mp4 -b:v 3000k output.m3u8

# ✅ Good - Adaptive bitrate
ffmpeg -i input.mp4 \
  -b:v:0 3000k -b:v:1 1500k -b:v:2 800k \
  -var_stream_map "v:0 v:1 v:2" \
  output.m3u8

❌ Don't: No Monitoring

// ❌ Bad - No stream health checks
stream.on('data', (chunk) => {
  // No monitoring!
})

// ✅ Good - Monitor stream health
stream.on('data', (chunk) => {
  metrics.increment('stream.bytes')
  metrics.gauge('stream.bitrate', calculateBitrate(chunk))
})

// Alert on issues
if (bitrate < threshold) {
  alert('Low bitrate detected')
}

Integration Points

• CDN Delivery (37-video-streaming/cdn-delivery/) - CDN setup
• Adaptive Bitrate (37-video-streaming/adaptive-bitrate/) - ABR streaming
• Video Analytics (37-video-streaming/video-analytics/) - Viewer metrics

Further Reading

• NGINX-RTMP
• AWS IVS
• HLS.js
• WebRTC
• Mux
• WebRTC
• OBS Studio