Implement KLV datachanel for WebRTC

This implementation adds comprehensive KLV (Key-Length-Value) metadata support
to MediaMTX's WebRTC functionality. KLV metadata is transmitted via WebRTC data
channels, allowing real-time delivery of telemetry and metadata alongside video and audio streams.

Author: onorua

internal/protocols/webrtc/from_stream.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/bluenviron/gortsplib/v4/pkg/description"
 	"github.com/bluenviron/gortsplib/v4/pkg/format"
 	"github.com/bluenviron/gortsplib/v4/pkg/format/rtpav1"
 	"github.com/bluenviron/gortsplib/v4/pkg/format/rtph264"
@@ -638,6 +639,105 @@ func setupAudioTrack(
 	return nil, nil
 }
 
+// setupKLVDataChannel sets up KLV metadata transmission via WebRTC data channel
+func setupKLVDataChannel(
+	stream *stream.Stream,
+	reader stream.Reader,
+	pc *PeerConnection,
+) (format.Format, error) {
+	// Look for KLV format in the stream (using Generic format with KLV RTPMap)
+	var klvFormat format.Format
+	var klvMedia *description.Media
+
+	for _, media := range stream.Desc.Medias {
+		if media == nil {
+			continue
+		}
+		reader.Log(logger.Debug, "checking media type: %s", media.Type)
+		for _, forma := range media.Formats {
+			reader.Log(logger.Debug, "checking format: %T, codec: %s", forma, forma.Codec())
+
+			// Check for Generic format with KLV RTPMap
+			if genericFmt, ok := forma.(*format.Generic); ok {
+				reader.Log(logger.Debug, "found Generic format with RTPMap: %s", genericFmt.RTPMap())
+				if genericFmt.RTPMap() == "KLV/90000" {
+					klvFormat = genericFmt
+					klvMedia = media
+					break
+				}
+			}
+
+			// Check for KLV format (using type assertion)
+			if _, ok := forma.(*format.KLV); ok {
+				reader.Log(logger.Debug, "found KLV format")
+				klvFormat = forma
+				klvMedia = media
+				break
+			}
+
+			// Also check for internal KLV format by codec name
+			if forma.Codec() == "KLV" {
+				reader.Log(logger.Debug, "found format with KLV codec")
+				klvFormat = forma
+				klvMedia = media
+				break
+			}
+		}
+		if klvFormat != nil {
+			break
+		}
+	}
+
+	if klvFormat == nil {
+		// No KLV format found, return nil without error
+		return nil, nil
+	}
+
+	reader.Log(logger.Info, "setting up KLV metadata transmission via WebRTC data channel")
+
+	// Add reader for KLV data and send via data channel
+	stream.AddReader(
+		reader,
+		klvMedia,
+		klvFormat,
+		func(u unit.Unit) error {
+			// Handle both Generic and KLV units
+			var klvData []byte
+
+			switch tunit := u.(type) {
+			case *unit.Generic:
+				// Extract KLV data from Generic unit RTP packets
+				if tunit.RTPPackets != nil {
+					for _, pkt := range tunit.RTPPackets {
+						klvData = append(klvData, pkt.Payload...)
+					}
+				}
+			case *unit.KLV:
+				// Extract KLV data from KLV unit
+				if tunit.Unit != nil {
+					klvData = append(klvData, tunit.Unit...)
+				}
+			default:
+				return nil // Unknown unit type, skip
+			}
+
+			if len(klvData) == 0 {
+				return nil
+			}
+
+			// Send KLV data through WebRTC data channel
+			err := pc.SendKLVData(klvData)
+			if err != nil {
+				reader.Log(logger.Debug, "failed to send KLV data via data channel: %v", err)
+				// Don't return error to avoid breaking the stream
+			}
+
+			return nil
+		})
+
+	return klvFormat, nil
+}
+
 // FromStream maps a MediaMTX stream to a WebRTC connection
 func FromStream(
 	stream *stream.Stream,
@@ -658,10 +758,22 @@ func FromStream(
 		return errNoSupportedCodecsFrom
 	}
 
+	// Setup KLV metadata handling via data channel
+	klvFormat, err := setupKLVDataChannel(stream, reader, pc)
+	if err != nil {
+		reader.Log(logger.Warn, "failed to setup KLV data channel: %v", err)
+	}
+
 	n := 1
 	for _, media := range stream.Desc.Medias {
+		if media == nil {
+			continue
+		}
 		for _, forma := range media.Formats {
-			if forma != videoFormat && forma != audioFormat {
+			if forma == nil {
+				continue
+			}
+			if forma != videoFormat && forma != audioFormat && forma != klvFormat {
 				reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
 			}
 			n++

internal/protocols/webrtc/peer_connection.go

@@ -108,6 +108,10 @@ type PeerConnection struct {
 	ctx               context.Context
 	ctxCancel         context.CancelFunc
 	incomingTracks    []*IncomingTrack
+
+	// KLV data channel support
+	klvDataChannel  *webrtc.DataChannel
+	klvChannelReady chan struct{}
 }
 
 // Start starts the peer connection.
@@ -239,6 +243,7 @@ func (co *PeerConnection) Start() error {
 	co.closed = make(chan struct{})
 	co.gatheringDone = make(chan struct{})
 	co.incomingTrack = make(chan trackRecvPair)
+	co.klvChannelReady = make(chan struct{})
 
 	co.ctx, co.ctxCancel = context.WithCancel(context.Background())
 
@@ -250,7 +255,20 @@ func (co *PeerConnection) Start() error {
 				return err
 			}
 		}
+
+		// Setup KLV data channel for publishing
+		err = co.setupKLVDataChannel()
+		if err != nil {
+			co.wr.GracefulClose() //nolint:errcheck
+			return err
+		}
 	} else {
+		// Setup KLV data channel for reading (WHEP)
+		err = co.setupKLVDataChannel()
+		if err != nil {
+			co.wr.GracefulClose() //nolint:errcheck
+			return err
+		}
 		_, err = co.wr.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{
 			Direction: webrtc.RTPTransceiverDirectionRecvonly,
 		})
@@ -356,6 +374,72 @@ func (co *PeerConnection) Close() {
 	<-co.closed
 }
 
+// setupKLVDataChannel sets up a data channel for KLV metadata transmission
+func (co *PeerConnection) setupKLVDataChannel() error {
+	// Create data channel for KLV metadata
+	dataChannelInit := &webrtc.DataChannelInit{
+		Ordered:        &[]bool{true}[0],  // Ensure ordered delivery
+		MaxRetransmits: &[]uint16{3}[0],   // Allow some retransmissions for reliability
+	}
+
+	var err error
+	co.klvDataChannel, err = co.wr.CreateDataChannel("klv", dataChannelInit)
+	if err != nil {
+		return fmt.Errorf("failed to create KLV data channel: %w", err)
+	}
+
+	// Set up data channel event handlers
+	co.klvDataChannel.OnOpen(func() {
+		co.Log.Log(logger.Info, "KLV data channel opened")
+		close(co.klvChannelReady)
+	})
+
+	co.klvDataChannel.OnClose(func() {
+		co.Log.Log(logger.Info, "KLV data channel closed")
+	})
+
+	co.klvDataChannel.OnError(func(err error) {
+		co.Log.Log(logger.Warn, "KLV data channel error: %v", err)
+	})
+
+	return nil
+}
+
+// SendKLVData sends KLV metadata through the data channel
+func (co *PeerConnection) SendKLVData(klvData []byte) error {
+	if co.klvDataChannel == nil {
+		return fmt.Errorf("KLV data channel not initialized")
+	}
+
+	// Wait for channel to be ready (with timeout)
+	select {
+	case <-co.klvChannelReady:
+		// Channel is ready
+	case <-time.After(5 * time.Second):
+		return fmt.Errorf("timeout waiting for KLV data channel to be ready")
+	case <-co.ctx.Done():
+		return fmt.Errorf("context cancelled")
+	}
+
+	// Check if channel is still open
+	if co.klvDataChannel.ReadyState() != webrtc.DataChannelStateOpen {
+		return fmt.Errorf("KLV data channel is not open (state: %s)", co.klvDataChannel.ReadyState())
+	}
+
+	// Send the KLV data
+	err := co.klvDataChannel.Send(klvData)
+	if err != nil {
+		return fmt.Errorf("failed to send KLV data: %w", err)
+	}
+
+	return nil
+}
+
+// KLVChannelReady returns a channel that closes when the KLV data channel is ready
+func (co *PeerConnection) KLVChannelReady() <-chan struct{} {
+	return co.klvChannelReady
+}
+
 // CreatePartialOffer creates a partial offer.
 func (co *PeerConnection) CreatePartialOffer() (*webrtc.SessionDescription, error) {
 	offer, err := co.wr.CreateOffer(nil)

internal/servers/webrtc/reader.js

@@ -428,6 +428,7 @@ class MediaMTXWebRTCReader {
     this.pc.onicecandidate = (evt) => this.#onLocalCandidate(evt);
     this.pc.onconnectionstatechange = () => this.#onConnectionState();
     this.pc.ontrack = (evt) => this.#onTrack(evt);
+    this.pc.ondatachannel = (evt) => this.#onDataChannel(evt);
 
     return this.pc.createOffer()
       .then((offer) => {
@@ -547,6 +548,32 @@ class MediaMTXWebRTCReader {
       this.conf.onTrack(evt);
     }
   }
+
+  #onDataChannel(evt) {
+    const dataChannel = evt.channel;
+
+    if (dataChannel.label === 'klv') {
+      dataChannel.onopen = () => {
+        console.log('KLV metadata data channel opened');
+      };
+
+      dataChannel.onmessage = (event) => {
+        if (this.conf.onKLVData !== undefined) {
+          // Parse KLV data from ArrayBuffer
+          const klvData = new Uint8Array(event.data);
+          this.conf.onKLVData(klvData);
+        }
+      };
+
+      dataChannel.onclose = () => {
+        console.log('KLV metadata data channel closed');
+      };
+
+      dataChannel.onerror = (error) => {
+        console.error('KLV metadata data channel error:', error);
+      };
+    }
+  }
 }
 
 window.MediaMTXWebRTCReader = MediaMTXWebRTCReader;