Live Streaming

Overview

Live streaming in Moqtail enables real-time delivery of continuously generated content like video streams, audio feeds, or sensor data. The LiveTrackSource class wraps a ReadableStream<MoqtObject> and handles the complexities of live object distribution.

Live tracks are optimized for forward-only delivery. Subscribers can join at any time and receive objects from the current position forward.

Creating a Live Track Source

The LiveTrackSource is designed for streaming content that is continuously produced:

import { LiveTrackSource, MoqtObject } from 'moqtail-ts'

// Create a readable stream of objects
const liveStream = new ReadableStream<MoqtObject>({
  async start(controller) {
    // Your content generation logic
    while (capturing) {
      const object = await generateNextObject()
      controller.enqueue(object)
    }
    controller.close()
  }
})

// Wrap in LiveTrackSource
const liveSource = new LiveTrackSource(liveStream)

Live Source Interface

The LiveTrackSource implements the LiveObjectSource interface:

interface LiveObjectSource {
  readonly stream: ReadableStream<MoqtObject>
  readonly largestLocation: Location | undefined
  onNewObject(listener: (obj: MoqtObject) => void): () => void
  onDone(listener: () => void): () => void
  stop(): void
}

Stream Access

The underlying ReadableStream<MoqtObject> that produces objects:

const stream = liveSource.stream

Latest Position

Tracks the highest location seen so far:

// Returns undefined until first object arrives
const latest = liveSource.largestLocation
if (latest) {
  console.log(`Latest: Group ${latest.group}, Object ${latest.object}`)
}

Event Listeners

// Listen for each new object
const unsubscribe = liveSource.onNewObject((obj) => {
  console.log(`New object: ${obj.groupId}:${obj.objectId}`)
})

// Listen for stream end
liveSource.onDone(() => {
  console.log('Live stream ended')
})

// Clean up when done
unsubscribe()

Manual Stop

Stop ingestion and release resources:

liveSource.stop()

Real-Time Video Example

Here’s a complete example streaming video from a canvas:

import {
  MOQtailClient,
  FullTrackName,
  Track,
  LiveTrackSource,
  MoqtObject,
  Location,
  ObjectForwardingPreference
} from 'moqtail-ts'

// Configuration
const KEYFRAME_INTERVAL = 30 // frames
const fullTrackName = FullTrackName.tryNew('live/conference', 'video')

// Create video encoder stream
const videoStream = new ReadableStream<MoqtObject>({
  async start(controller) {
    let groupId = 0n
    let objectId = 0n
    let frameCount = 0
    
    // Get canvas stream
    const canvas = document.getElementById('video-canvas') as HTMLCanvasElement
    const stream = canvas.captureStream(30) // 30 fps
    const track = stream.getVideoTracks()[0]
    
    // Set up video encoder
    const encoder = new VideoEncoder({
      output: (chunk, metadata) => {
        const isKeyframe = chunk.type === 'key'
        
        // Create MoqtObject from encoded chunk
        const object = MoqtObject.newWithPayload(
          fullTrackName,
          new Location(groupId, objectId),
          isKeyframe ? 0 : 8, // Higher priority for keyframes
          ObjectForwardingPreference.Subgroup,
          null,
          null,
          new Uint8Array(chunk.byteLength).fill(0) // Copy chunk data
        )
        
        controller.enqueue(object)
        
        // Advance position
        if (isKeyframe && objectId > 0n) {
          groupId++
          objectId = 0n
        } else {
          objectId++
        }
      },
      error: (error) => {
        console.error('Encoding error:', error)
        controller.error(error)
      }
    })
    
    encoder.configure({
      codec: 'vp09.00.10.08',
      width: 1280,
      height: 720,
      bitrate: 2_000_000,
      framerate: 30
    })
    
    // Encode frames
    const reader = new MediaStreamTrackProcessor({ track }).readable.getReader()
    
    try {
      while (true) {
        const { done, value: frame } = await reader.read()
        if (done) break
        
        encoder.encode(frame, { keyFrame: frameCount % KEYFRAME_INTERVAL === 0 })
        frame.close()
        frameCount++
      }
    } finally {
      await encoder.flush()
      controller.close()
    }
  }
})

// Create track with live source
const videoTrack: Track = {
  fullTrackName,
  forwardingPreference: ObjectForwardingPreference.Subgroup,
  trackSource: {
    live: new LiveTrackSource(videoStream)
  },
  publisherPriority: 0
}

// Publish
const client = await MOQtailClient.new({
  url: 'https://relay.example.com',
  supportedVersions: [0xff00000b]
})

client.addOrUpdateTrack(videoTrack)
await client.publishNamespace(['live', 'conference'])

Audio Streaming Example

Stream audio from a microphone:

import {
  LiveTrackSource,
  MoqtObject,
  Location,
  FullTrackName,
  ObjectForwardingPreference
} from 'moqtail-ts'

const fullTrackName = FullTrackName.tryNew('live/conference', 'audio')

const audioStream = new ReadableStream<MoqtObject>({
  async start(controller) {
    let groupId = 0n // Audio segment (e.g., 1 second)
    let objectId = 0n // Packet within segment
    let packetCount = 0
    
    const PACKETS_PER_SEGMENT = 50 // 20ms packets for 1s segment
    
    // Get microphone stream
    const stream = await navigator.mediaDevices.getUserMedia({ audio: true })
    const audioTrack = stream.getAudioTracks()[0]
    
    // Set up audio encoder (Opus)
    const encoder = new AudioEncoder({
      output: (chunk) => {
        const object = MoqtObject.newWithPayload(
          fullTrackName,
          new Location(groupId, objectId),
          16, // Medium priority
          ObjectForwardingPreference.Datagram, // Low latency for audio
          null,
          null,
          new Uint8Array(chunk.byteLength).fill(0) // Copy chunk data
        )
        
        controller.enqueue(object)
        
        // Move to next segment after 50 packets
        objectId++
        packetCount++
        if (packetCount >= PACKETS_PER_SEGMENT) {
          groupId++
          objectId = 0n
          packetCount = 0
        }
      },
      error: (error) => controller.error(error)
    })
    
    encoder.configure({
      codec: 'opus',
      sampleRate: 48000,
      numberOfChannels: 2,
      bitrate: 128_000
    })
    
    // Encode audio
    const reader = new MediaStreamTrackProcessor({ track: audioTrack }).readable.getReader()
    
    while (true) {
      const { done, value: frame } = await reader.read()
      if (done) break
      
      encoder.encode(frame)
      frame.close()
    }
    
    await encoder.flush()
    controller.close()
  }
})

const audioTrack: Track = {
  fullTrackName,
  forwardingPreference: ObjectForwardingPreference.Datagram,
  trackSource: {
    live: new LiveTrackSource(audioStream)
  },
  publisherPriority: 8 // High priority but below keyframes
}

Datagram Delivery

For ultra-low latency scenarios, use datagram delivery:

import { ObjectForwardingPreference } from 'moqtail-ts'

const liveTrack: Track = {
  fullTrackName: FullTrackName.tryNew('sensors/temperature', 'data'),
  forwardingPreference: ObjectForwardingPreference.Datagram,
  trackSource: {
    live: new LiveTrackSource(sensorStream)
  },
  publisherPriority: 0
}

// Enable datagrams on the client
const client = await MOQtailClient.new({
  url: 'https://relay.example.com',
  supportedVersions: [0xff00000b],
  enableDatagrams: true // Required for datagram delivery
})

Datagrams may be lost or delivered out of order. Only use for content where latency is more critical than reliability (e.g., sensor data, audio, low-latency video).

Stream Cancellation

The library implements intelligent stream cancellation:

// Automatically cancels streams for older groups
// when bandwidth is limited

// The publication system handles this internally
// when delivering subgroup objects to subscribers

Stream cancellation ensures subscribers always receive the most recent content with minimal latency, automatically dropping outdated frames during network congestion.

Handling End of Stream

Signal the end of a live track properly:

const liveStream = new ReadableStream<MoqtObject>({
  async start(controller) {
    // ... generate objects ...
    
    // When done, send end-of-track marker
    const endObject = MoqtObject.newWithStatus(
      fullTrackName,
      new Location(lastGroupId, lastObjectId + 1n),
      0,
      ObjectForwardingPreference.Subgroup,
      null,
      null,
      ObjectStatus.EndOfTrack
    )
    
    controller.enqueue(endObject)
    controller.close()
  }
})

Best Practices

GOP Structure

Start each GOP with a keyframe (objectId = 0)
Keep GOP duration consistent (e.g., 2 seconds)
Use group boundaries to enable efficient seeking

Priority Assignment

Keyframes: 0 (highest priority)
P-frames: 8-16
B-frames: 16-32
Audio: 8-16
Metadata: 32-64

Error Handling

const liveSource = new LiveTrackSource(liveStream)

liveSource.onDone(() => {
  console.log('Stream ended - cleaning up')
  client.removeTrack(track)
})

Resource Cleanup

// Stop live source when no longer needed
liveSource.stop()

// Remove track from client
client.removeTrack(track)

// Announce namespace done
await client.publishNamespaceDone(namespace)

Overview

Creating a Live Track Source

Live Source Interface

Real-Time Video Example

Audio Streaming Example

Datagram Delivery

Stream Cancellation

Handling End of Stream

Best Practices

GOP Structure

Priority Assignment

Error Handling

Resource Cleanup

Next Steps

Static Content

Hybrid Content

Documentation Index

​Overview

​Creating a Live Track Source

​Live Source Interface

​Real-Time Video Example

​Audio Streaming Example

​Datagram Delivery

​Stream Cancellation

​Handling End of Stream

​Best Practices

GOP Structure

Priority Assignment

Error Handling

Resource Cleanup

​Next Steps

Static Content

Hybrid Content

Overview

Creating a Live Track Source

Live Source Interface

Real-Time Video Example

Audio Streaming Example

Datagram Delivery

Stream Cancellation

Handling End of Stream

Best Practices

Next Steps