Move data track packet serialization to protocol (#4829)

* Pin protocol to data track serialization branch

* Use datatrack package from protocol repo

* Update protocol sha

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
 	github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095
-	github.com/livekit/protocol v1.50.5-0.20260829123501-a469dd43727b
+	github.com/livekit/protocol v1.51.1-0.20260901194019-9660e8efd0b0
 	github.com/livekit/psrpc v0.7.5
 	github.com/mackerelio/go-osstat v0.2.8
 	github.com/magefile/mage v1.17.2

go.sum

@@ -164,8 +164,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT
 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 h1:BcliKAXoMhl/nWmzQweQ5kmh4Qqagxl4s3Z5pvM/7AY=
 github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU=
-github.com/livekit/protocol v1.50.5-0.20260829123501-a469dd43727b h1:OOqCnFt5AZRVdSjYzaOzCR9KiB9IeeTPMGls+C1Ti2g=
-github.com/livekit/protocol v1.50.5-0.20260829123501-a469dd43727b/go.mod h1:x7m1nX86XfvS0YoqXnVe0jixkCmAZglR4mcbRHURSro=
+github.com/livekit/protocol v1.51.1-0.20260901194019-9660e8efd0b0 h1:HdcExqttH0EkDNyrVf7dy4McuLjuUdztWTF0a7I0ans=
+github.com/livekit/protocol v1.51.1-0.20260901194019-9660e8efd0b0/go.mod h1:x7m1nX86XfvS0YoqXnVe0jixkCmAZglR4mcbRHURSro=
 github.com/livekit/psrpc v0.7.5 h1:WxfJIQ41X1b+48A1uzc8Gy9FhYEMxBJNCpCYqPdO/Ds=
 github.com/livekit/psrpc v0.7.5/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw=
 github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1 h1:fH+v4W+NFp9FfPzON6FaUFNmazGcctaAhb2P+Ksf+1s=

pkg/rtc/datadowntrack.go

@@ -19,8 +19,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 )

pkg/rtc/datatrack.go

@@ -21,9 +21,9 @@ import (
 
 	"github.com/frostbyte73/core"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	sfuutils "github.com/livekit/livekit-server/pkg/sfu/utils"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/utils"

pkg/rtc/datatrack/extension_participant_sid.go

@@ -1,59 +0,0 @@
-// Copyright 2023 LiveKit, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package datatrack
-
-import (
-	"errors"
-
-	"github.com/livekit/protocol/livekit"
-)
-
-type ExtensionParticipantSid struct {
-	participantID livekit.ParticipantID
-}
-
-func NewExtensionParticipantSid(participantID livekit.ParticipantID) (*ExtensionParticipantSid, error) {
-	if len(participantID) >= 256 {
-		return nil, errors.New("participantID too long")
-	}
-
-	return &ExtensionParticipantSid{participantID}, nil
-}
-
-func (e *ExtensionParticipantSid) ParticipantID() livekit.ParticipantID {
-	return e.participantID
-}
-
-func (e *ExtensionParticipantSid) Marshal() (Extension, error) {
-	data := make([]byte, len(e.participantID))
-	copy(data, e.participantID)
-	return Extension{
-		id:   uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID),
-		data: data,
-	}, nil
-}
-
-func (e *ExtensionParticipantSid) Unmarshal(ext Extension) error {
-	if ext.id != uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID) {
-		return errors.New("invalid extension ID")
-	}
-
-	if len(ext.data) == 0 {
-		return errors.New("empty extension data")
-	}
-
-	e.participantID = livekit.ParticipantID(ext.data)
-	return nil
-}

pkg/rtc/datatrack/extension_participant_sid_test.go

@@ -1,46 +0,0 @@
-// Copyright 2023 LiveKit, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package datatrack
-
-import (
-	"testing"
-
-	"github.com/livekit/protocol/livekit"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestExtensionParticipantSid(t *testing.T) {
-	longTestParticipantID := livekit.ParticipantID(make([]byte, 256))
-	_, err := NewExtensionParticipantSid(longTestParticipantID)
-	require.Error(t, err)
-
-	testParticipantID := livekit.ParticipantID("test")
-	extParticipantSid, err := NewExtensionParticipantSid(testParticipantID)
-	require.NoError(t, err)
-
-	expectedExt := Extension{
-		id:   uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID),
-		data: []byte{'t', 'e', 's', 't'},
-	}
-	ext, err := extParticipantSid.Marshal()
-	require.NoError(t, err)
-	require.Equal(t, expectedExt, ext)
-
-	var unmarshaled ExtensionParticipantSid
-	err = unmarshaled.Unmarshal(ext)
-	require.NoError(t, err)
-	assert.Equal(t, testParticipantID, unmarshaled.ParticipantID())
-}

pkg/rtc/datatrack/packet.go

@@ -1,310 +0,0 @@
-// Copyright 2023 LiveKit, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package datatrack
-
-import (
-	"encoding/binary"
-	"errors"
-	"fmt"
-)
-
-var (
-	errHeaderSizeInsufficient    = errors.New("data track packet header size insufficient")
-	errBufferSizeInsufficient    = errors.New("data track packet buffer size insufficient")
-	errExtensionSizeInsufficient = errors.New("data track packet extension size insufficient")
-	errExtensionNotFound         = errors.New("data track packet extension not found")
-	errExtensionSizeTooBig       = errors.New("extension size is too big")
-)
-
-const (
-	headerLength = 12
-
-	versionShift = 5
-	versionMask  = (1 << 3) - 1
-
-	startOfFrameShift = 4
-	startOfFrameMask  = (1 << 1) - 1
-
-	finalOfFrameShift = 3
-	finalOfFrameMask  = (1 << 1) - 1
-
-	extensionsShift = 2
-	extensionsMask  = (1 << 1) - 1
-
-	handleOffset = 2
-	handleLength = 2
-
-	seqNumOffset = 4
-	seqNumLength = 2
-
-	frameNumOffset = 6
-	frameNumLength = 2
-
-	timestampOffset = 8
-	timestampLength = 4
-
-	extensionsSizeOffset = headerLength
-	extensionsSizeLength = 2
-
-	extensionIDLength   = 1
-	extensionSizeLength = 1
-)
-
-type Extension struct {
-	id   uint8
-	data []byte
-}
-
-type Header struct {
-	Version        uint8
-	IsStartOfFrame bool
-	IsFinalOfFrame bool
-	HasExtensions  bool
-	Handle         uint16
-	SequenceNumber uint16
-	FrameNumber    uint16
-	Timestamp      uint32
-	ExtensionsSize uint16
-	Extensions     []Extension
-}
-
-/*
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-	┆*  0                   1                   2                   3
-	┆*  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-	┆* |V    |S|F|X| reserved          | handle                        |
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-	┆* | sequence number               | frame number                  |
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-	┆* |                           timestamp                           |
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-	|* Extensions Size if X=1          | Extensions...                 |
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-
-	Each extension
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-	┆*  0                   1                   2                   3
-	┆*  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-	┆* | Extension ID  | Extension size| Extension payload             |
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-
-	End of all extensions
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-	|* padded to 4 byte boundary if aggregate of `Extensions Size`     |
-	|* field and all extensions do not end on a 4 byte boundary        |
-	┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-*/
-
-func (h *Header) Unmarshal(buf []byte) (int, error) {
-	if len(buf) < headerLength {
-		return 0, fmt.Errorf("%w: %d < %d", errHeaderSizeInsufficient, len(buf), headerLength)
-	}
-
-	hdrSize := headerLength
-	h.Version = buf[0] >> versionShift & versionMask
-	h.IsStartOfFrame = (buf[0] >> startOfFrameShift & startOfFrameMask) > 0
-	h.IsFinalOfFrame = (buf[0] >> finalOfFrameShift & finalOfFrameMask) > 0
-	h.HasExtensions = (buf[0] >> extensionsShift & extensionsMask) > 0
-
-	h.Handle = binary.BigEndian.Uint16(buf[handleOffset : handleOffset+handleLength])
-	h.SequenceNumber = binary.BigEndian.Uint16(buf[seqNumOffset : seqNumOffset+seqNumLength])
-	h.FrameNumber = binary.BigEndian.Uint16(buf[frameNumOffset : frameNumOffset+frameNumLength])
-	h.Timestamp = binary.BigEndian.Uint32(buf[timestampOffset : timestampOffset+timestampLength])
-
-	if h.HasExtensions {
-		if len(buf) < extensionsSizeOffset+extensionsSizeLength {
-			return 0, fmt.Errorf("%w: %d < %d", errHeaderSizeInsufficient, len(buf), extensionsSizeOffset+extensionsSizeLength)
-		}
-		extensionsSize := (int(binary.BigEndian.Uint16(buf[extensionsSizeOffset:extensionsSizeOffset+extensionsSizeLength]))+1)*4 - extensionsSizeLength
-		hdrSize += extensionsSizeLength
-
-		extensionHeaderSize := extensionIDLength + extensionSizeLength
-		remainingSize := extensionsSize
-		idx := extensionsSizeOffset + extensionsSizeLength
-		for remainingSize != 0 {
-			// read extension header
-			if len(buf[idx:]) < extensionIDLength || remainingSize < extensionIDLength {
-				return 0, fmt.Errorf("%w: %d/%d < %d", errExtensionSizeInsufficient, remainingSize, len(buf[idx:]), extensionIDLength)
-			}
-			id := buf[idx]
-			if id == 0 {
-				// end of extensions, padding has started
-				if len(buf[idx:]) < remainingSize {
-					return 0, fmt.Errorf("%w: %d/%d < %d", errExtensionSizeInsufficient, remainingSize, len(buf[idx:]), remainingSize)
-				}
-				hdrSize += remainingSize
-				break
-			}
-
-			if len(buf[idx+1:]) < extensionSizeLength || remainingSize < extensionSizeLength {
-				return 0, fmt.Errorf("%w: %d/%d < %d", errExtensionSizeInsufficient, remainingSize, len(buf[idx:]), extensionSizeLength)
-			}
-			size := int(buf[idx+1])
-
-			remainingSize -= extensionHeaderSize
-			idx += extensionHeaderSize
-			hdrSize += extensionHeaderSize
-
-			// read extension data
-			if len(buf[idx:]) < size || remainingSize < size {
-				return 0, fmt.Errorf("%w: %d/%d < %d", errExtensionSizeInsufficient, remainingSize, len(buf[idx:]), size)
-			}
-			h.Extensions = append(h.Extensions, Extension{id: id, data: buf[idx : idx+size]})
-
-			remainingSize -= size
-			idx += size
-			hdrSize += size
-		}
-		h.ExtensionsSize = uint16(extensionsSize - remainingSize)
-	}
-
-	return hdrSize, nil
-}
-
-func (h *Header) MarshalSize() int {
-	extensionsSize := 0
-	if h.HasExtensions {
-		extensionsSize += extensionsSizeLength
-		for _, ext := range h.Extensions {
-			extensionsSize += len(ext.data) + extensionIDLength + extensionSizeLength
-		}
-	}
-
-	return headerLength + (extensionsSize+3)/4*4
-}
-
-func (h *Header) MarshalTo(buf []byte) (int, error) {
-	if len(buf) < headerLength {
-		return 0, fmt.Errorf("%w: %d < %d", errHeaderSizeInsufficient, len(buf), headerLength)
-	}
-
-	hdrSize := headerLength
-	buf[0] = h.Version << versionShift
-	if h.IsStartOfFrame {
-		buf[0] |= (1 << startOfFrameShift)
-	}
-	if h.IsFinalOfFrame {
-		buf[0] |= (1 << finalOfFrameShift)
-	}
-	if h.HasExtensions {
-		buf[0] |= (1 << extensionsShift)
-	}
-
-	binary.BigEndian.PutUint16(buf[handleOffset:handleOffset+handleLength], h.Handle)
-	binary.BigEndian.PutUint16(buf[seqNumOffset:seqNumOffset+seqNumLength], h.SequenceNumber)
-	binary.BigEndian.PutUint16(buf[frameNumOffset:frameNumOffset+frameNumLength], h.FrameNumber)
-	binary.BigEndian.PutUint32(buf[timestampOffset:timestampOffset+timestampLength], h.Timestamp)
-
-	if h.HasExtensions {
-		extensionsSize := (extensionsSizeLength + h.ExtensionsSize + 3) / 4 * 4
-		binary.BigEndian.PutUint16(buf[extensionsSizeOffset:extensionsSizeOffset+extensionsSizeLength], (extensionsSize/4)-1)
-		hdrSize += extensionsSizeLength
-
-		addedSize := 0
-		idx := extensionsSizeOffset + extensionsSizeLength
-		for _, ext := range h.Extensions {
-			buf[idx] = ext.id
-			if len(ext.data) > 255 {
-				return 0, fmt.Errorf("%w: %d > 255", errExtensionSizeTooBig, len(ext.data))
-			}
-			buf[idx+extensionIDLength] = byte(len(ext.data))
-			copy(buf[idx+extensionIDLength+extensionSizeLength:], ext.data)
-
-			extSize := len(ext.data) + extensionIDLength + extensionSizeLength
-			idx += extSize
-			hdrSize += extSize
-			addedSize += extSize
-		}
-
-		paddingSize := extensionsSize - extensionsSizeLength - uint16(addedSize)
-		for i := range paddingSize {
-			buf[idx+int(i)] = 0
-		}
-		idx += int(paddingSize)
-		hdrSize += int(paddingSize)
-	}
-
-	return hdrSize, nil
-}
-
-func (h *Header) AddExtension(ext Extension) {
-	for i, existingExt := range h.Extensions {
-		if existingExt.id == ext.id {
-			h.ExtensionsSize -= uint16(len(existingExt.data) + extensionIDLength + extensionSizeLength)
-			h.Extensions[i].data = ext.data
-			h.ExtensionsSize += uint16(len(h.Extensions[i].data) + extensionIDLength + extensionSizeLength)
-			return
-		}
-	}
-
-	h.Extensions = append(h.Extensions, ext)
-	h.ExtensionsSize += uint16(len(ext.data) + extensionIDLength + extensionSizeLength)
-	h.HasExtensions = true
-}
-
-func (h *Header) GetExtension(id uint8) (Extension, error) {
-	for _, ext := range h.Extensions {
-		if ext.id == id {
-			return ext, nil
-		}
-	}
-	return Extension{}, fmt.Errorf("%w, id: %d", errExtensionNotFound, id)
-}
-
-// ----------------------------------------------------
-
-type Packet struct {
-	Header
-	Payload []byte
-}
-
-func (p *Packet) Unmarshal(buf []byte) error {
-	hdrSize, err := p.Header.Unmarshal(buf)
-	if err != nil {
-		return err
-	}
-	if hdrSize > len(buf) {
-		return fmt.Errorf("%w: %d < %d", errBufferSizeInsufficient, len(buf), hdrSize)
-	}
-
-	p.Payload = buf[hdrSize:]
-	return nil
-}
-
-func (p *Packet) Marshal() ([]byte, error) {
-	buf := make([]byte, p.Header.MarshalSize()+len(p.Payload))
-	if err := p.MarshalTo(buf); err != nil {
-		return nil, err
-	}
-
-	return buf, nil
-}
-
-func (p *Packet) MarshalTo(buf []byte) error {
-	size := p.Header.MarshalSize() + len(p.Payload)
-	if len(buf) < size {
-		return fmt.Errorf("%w: %d < %d", errBufferSizeInsufficient, len(buf), size)
-	}
-
-	hdrSize, err := p.Header.MarshalTo(buf)
-	if err != nil {
-		return err
-	}
-
-	copy(buf[hdrSize:], p.Payload)
-	return nil
-}

pkg/rtc/datatrack/packet_test.go

@@ -1,294 +0,0 @@
-// Copyright 2023 LiveKit, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package datatrack
-
-import (
-	"testing"
-
-	"github.com/livekit/protocol/livekit"
-	"github.com/stretchr/testify/require"
-)
-
-func TestPacket(t *testing.T) {
-	t.Run("without extension", func(t *testing.T) {
-		payload := make([]byte, 6)
-		for i := range len(payload) {
-			payload[i] = byte(255 - i)
-		}
-		packet := &Packet{
-			Header: Header{
-				Version:        0,
-				IsStartOfFrame: true,
-				IsFinalOfFrame: true,
-				Handle:         3333,
-				SequenceNumber: 6666,
-				FrameNumber:    9999,
-				Timestamp:      0xdeadbeef,
-			},
-			Payload: payload,
-		}
-		rawPacket, err := packet.Marshal()
-		require.NoError(t, err)
-
-		expectedRawPacket := []byte{
-			0x18, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0xff, 0xfe, 0xfd, 0xfc,
-			0xfb, 0xfa,
-		}
-		require.Equal(t, expectedRawPacket, rawPacket)
-
-		var unmarshaled Packet
-		err = unmarshaled.Unmarshal(rawPacket)
-		require.NoError(t, err)
-		require.Equal(t, packet, &unmarshaled)
-	})
-
-	t.Run("with extension", func(t *testing.T) {
-		payload := make([]byte, 4)
-		for i := range len(payload) {
-			payload[i] = byte(255 - i)
-		}
-		packet := &Packet{
-			Header: Header{
-				Version:        0,
-				IsStartOfFrame: true,
-				IsFinalOfFrame: false,
-				Handle:         3333,
-				SequenceNumber: 6666,
-				FrameNumber:    9999,
-				Timestamp:      0xdeadbeef,
-			},
-			Payload: payload,
-		}
-		if extParticipantSid, err := NewExtensionParticipantSid("test_participant"); err == nil {
-			if ext, err := extParticipantSid.Marshal(); err == nil {
-				packet.AddExtension(ext)
-			}
-		}
-		rawPacket, err := packet.Marshal()
-		require.NoError(t, err)
-
-		expectedRawPacket := []byte{
-			0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0x00, 0x04, 0x01, 0x10,
-			0x74, 0x65, 0x73, 0x74, 0x5f, 0x70, 0x61, 0x72,
-			0x74, 0x69, 0x63, 0x69, 0x70, 0x61, 0x6e, 0x74,
-			0xff, 0xfe, 0xfd, 0xfc,
-		}
-		require.Equal(t, expectedRawPacket, rawPacket)
-
-		var unmarshaled Packet
-		err = unmarshaled.Unmarshal(rawPacket)
-		require.NoError(t, err)
-		require.Equal(t, packet, &unmarshaled)
-
-		ext, err := unmarshaled.GetExtension(uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID))
-		require.NoError(t, err)
-
-		var extParticipantSid ExtensionParticipantSid
-		require.NoError(t, extParticipantSid.Unmarshal(ext))
-		require.Equal(t, livekit.ParticipantID("test_participant"), extParticipantSid.ParticipantID())
-	})
-
-	t.Run("with extension padding", func(t *testing.T) {
-		payload := make([]byte, 4)
-		for i := range len(payload) {
-			payload[i] = byte(255 - i)
-		}
-		packet := &Packet{
-			Header: Header{
-				Version:        0,
-				IsStartOfFrame: true,
-				IsFinalOfFrame: false,
-				Handle:         3333,
-				SequenceNumber: 6666,
-				FrameNumber:    9999,
-				Timestamp:      0xdeadbeef,
-			},
-			Payload: payload,
-		}
-		if extParticipantSid, err := NewExtensionParticipantSid("participant"); err == nil {
-			if ext, err := extParticipantSid.Marshal(); err == nil {
-				packet.AddExtension(ext)
-			}
-		}
-		rawPacket, err := packet.Marshal()
-		require.NoError(t, err)
-
-		expectedRawPacket := []byte{
-			0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x01, 0x0b,
-			0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70,
-			0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc,
-		}
-		require.Equal(t, expectedRawPacket, rawPacket)
-
-		var unmarshaled Packet
-		err = unmarshaled.Unmarshal(rawPacket)
-		require.NoError(t, err)
-		require.Equal(t, packet, &unmarshaled)
-
-		ext, err := unmarshaled.GetExtension(uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID))
-		require.NoError(t, err)
-
-		var extParticipantSid ExtensionParticipantSid
-		require.NoError(t, extParticipantSid.Unmarshal(ext))
-		require.Equal(t, livekit.ParticipantID("participant"), extParticipantSid.ParticipantID())
-	})
-
-	t.Run("replace extension", func(t *testing.T) {
-		payload := make([]byte, 4)
-		for i := range len(payload) {
-			payload[i] = byte(255 - i)
-		}
-		packet := &Packet{
-			Header: Header{
-				Version:        0,
-				IsStartOfFrame: true,
-				IsFinalOfFrame: false,
-				Handle:         3333,
-				SequenceNumber: 6666,
-				FrameNumber:    9999,
-				Timestamp:      0xdeadbeef,
-			},
-			Payload: payload,
-		}
-		if extParticipantSid, err := NewExtensionParticipantSid("participant"); err == nil {
-			if ext, err := extParticipantSid.Marshal(); err == nil {
-				packet.AddExtension(ext)
-			}
-		}
-		rawPacket, err := packet.Marshal()
-		require.NoError(t, err)
-
-		expectedRawPacket := []byte{
-			0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x01, 0x0b,
-			0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70,
-			0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc,
-		}
-		require.Equal(t, expectedRawPacket, rawPacket)
-
-		// replace existing extension ID and ensure that marshalled packet is updated
-		if extParticipantSid, err := NewExtensionParticipantSid("test_participant"); err == nil {
-			if ext, err := extParticipantSid.Marshal(); err == nil {
-				packet.AddExtension(ext)
-			}
-		}
-		rawPacket, err = packet.Marshal()
-		require.NoError(t, err)
-
-		expectedRawPacket = []byte{
-			0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0x00, 0x04, 0x01, 0x10,
-			0x74, 0x65, 0x73, 0x74, 0x5f, 0x70, 0x61, 0x72,
-			0x74, 0x69, 0x63, 0x69, 0x70, 0x61, 0x6e, 0x74,
-			0xff, 0xfe, 0xfd, 0xfc,
-		}
-		require.Equal(t, expectedRawPacket, rawPacket)
-
-		var unmarshaled Packet
-		err = unmarshaled.Unmarshal(rawPacket)
-		require.NoError(t, err)
-		require.Equal(t, packet, &unmarshaled)
-
-		ext, err := unmarshaled.GetExtension(uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID))
-		require.NoError(t, err)
-
-		var extParticipantSid ExtensionParticipantSid
-		require.NoError(t, extParticipantSid.Unmarshal(ext))
-		require.Equal(t, livekit.ParticipantID("test_participant"), extParticipantSid.ParticipantID())
-	})
-
-	t.Run("bad packet", func(t *testing.T) {
-		var unmarshaled Packet
-		// extensions size too small
-		badPacket := []byte{
-			0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0x00, 0x02, 0x01, 0x0b,
-			0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70,
-			0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc,
-		}
-		err := unmarshaled.Unmarshal(badPacket)
-		require.Error(t, err)
-
-		// get an invalid extension id
-		badPacket = []byte{
-			0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x02, 0x0b,
-			0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70,
-			0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc,
-		}
-		err = unmarshaled.Unmarshal(badPacket)
-		require.NoError(t, err)
-		_, err = unmarshaled.GetExtension(uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID))
-		require.Error(t, err)
-
-		// extension payload size bigger than payload
-		badPacket = []byte{
-			0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x01, 0x0d,
-			0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70,
-			0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc,
-		}
-		err = unmarshaled.Unmarshal(badPacket)
-		require.Error(t, err)
-
-		// extension payload size smaller than payload
-		badPacket = []byte{
-			0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f,
-			0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x01, 0x07,
-			0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70,
-			0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc,
-		}
-		err = unmarshaled.Unmarshal(badPacket)
-		require.Error(t, err)
-	})
-
-	t.Run("oversized extension padding does not panic", func(t *testing.T) {
-		var unmarshaled Packet
-		// HasExtensions set, extensionsSize describes more bytes than present,
-		// terminated by a 0x00 padding id -> hdrSize would exceed len(buf)
-		badPacket := []byte{
-			0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-		}
-		err := unmarshaled.Unmarshal(badPacket)
-		require.Error(t, err)
-	})
-
-	t.Run("extensions size wraparound does not panic", func(t *testing.T) {
-		var unmarshaled Packet
-		// 0xFFFF extensions-size field wraps (raw+1)*4 uint16 arithmetic to a huge
-		// remainingSize; the 0x00 padding id must not push hdrSize past len(buf)
-		badPacket := []byte{
-			0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-			0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0x00,
-		}
-		err := unmarshaled.Unmarshal(badPacket)
-		require.Error(t, err)
-	})
-
-	t.Run("truncated extensions size field does not panic", func(t *testing.T) {
-		var unmarshaled Packet
-		// HasExtensions set but buffer too short to hold the extensionsSize field
-		badPacket := []byte{
-			0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-			0x00, 0x00, 0x00, 0x00, 0x00,
-		}
-		err := unmarshaled.Unmarshal(badPacket)
-		require.Error(t, err)
-	})
-}

pkg/rtc/datatrack/testutils.go

@@ -1,74 +0,0 @@
-// Copyright 2023 LiveKit, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package datatrack
-
-import (
-	"math/rand"
-	"time"
-)
-
-func GenerateRawDataPackets(handle uint16, seqNum uint16, frameNum uint16, numFrames int, frameSize int, frameDuration time.Duration) [][]byte {
-	if seqNum == 0 {
-		seqNum = uint16(rand.Intn(256) + 1)
-	}
-	if frameNum == 0 {
-		frameNum = uint16(rand.Intn(256) + 1)
-	}
-	timestamp := uint32(rand.Intn(1024))
-
-	packetsPerFrame := (frameSize + 255) / 256 // using 256 bytes of payload per packet
-	if packetsPerFrame == 0 {
-		return nil
-	}
-	numPackets := packetsPerFrame * numFrames
-	rawPackets := make([][]byte, 0, numPackets)
-	for range numFrames {
-		remainingSize := frameSize
-		for packetIdx := range packetsPerFrame {
-			payloadSize := min(remainingSize, 256)
-			payload := make([]byte, payloadSize)
-			for i := range len(payload) {
-				payload[i] = byte(255 - i)
-			}
-			packet := &Packet{
-				Header: Header{
-					Version:        0,
-					IsStartOfFrame: packetIdx == 0,
-					IsFinalOfFrame: packetIdx == packetsPerFrame-1,
-					Handle:         handle,
-					SequenceNumber: seqNum,
-					FrameNumber:    frameNum,
-					Timestamp:      timestamp,
-				},
-				Payload: payload,
-			}
-			if extParticipantSid, err := NewExtensionParticipantSid("test_participant"); err == nil {
-				if ext, err := extParticipantSid.Marshal(); err == nil {
-					packet.AddExtension(ext)
-				}
-			}
-			rawPacket, err := packet.Marshal()
-			if err == nil {
-				rawPackets = append(rawPackets, rawPacket)
-			}
-			seqNum++
-			remainingSize -= payloadSize
-		}
-		frameNum++
-		timestamp += uint32(90000 * frameDuration.Seconds())
-	}
-
-	return rawPackets
-}

pkg/rtc/datatrack_stats.go

@@ -18,7 +18,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/utils/mono"
 )

pkg/rtc/participant_data_track.go

@@ -15,8 +15,8 @@
 package rtc
 
 import (
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/utils"

pkg/rtc/room.go

@@ -40,7 +40,6 @@ import (
 	"github.com/livekit/livekit-server/pkg/agent"
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/routing"
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
@@ -48,6 +47,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 	sutils "github.com/livekit/livekit-server/pkg/utils"
+	"github.com/livekit/protocol/datatrack"
 )
 
 const (

pkg/rtc/types/interfaces.go

@@ -30,11 +30,11 @@ import (
 	"github.com/livekit/protocol/utils"
 
 	"github.com/livekit/livekit-server/pkg/routing"
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	"github.com/livekit/livekit-server/pkg/telemetry"
+	"github.com/livekit/protocol/datatrack"
 
 	"google.golang.org/protobuf/proto"
 )

pkg/rtc/types/typesfakes/fake_data_track.go

@@ -4,8 +4,8 @@ package typesfakes
 import (
 	"sync"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 )
 

pkg/rtc/types/typesfakes/fake_data_track_sender.go

@@ -4,8 +4,8 @@ package typesfakes
 import (
 	"sync"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 )
 

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -6,13 +6,13 @@ import (
 	"time"
 
 	"github.com/livekit/livekit-server/pkg/routing"
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/protocol/auth"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/observability/roomobs"

pkg/rtc/types/typesfakes/fake_local_participant_listener.go

@@ -4,8 +4,8 @@ package typesfakes
 import (
 	"sync"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 )
 

pkg/rtc/types/typesfakes/fake_participant.go

@@ -5,8 +5,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/utils"

pkg/rtc/types/typesfakes/fake_participant_listener.go

@@ -4,8 +4,8 @@ package typesfakes
 import (
 	"sync"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 )
 

pkg/rtc/updatatrackmanager.go

@@ -19,8 +19,8 @@ import (
 	"slices"
 	"sync"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 )

test/client/client.go

@@ -46,10 +46,10 @@ import (
 	"github.com/livekit/protocol/signalling"
 
 	"github.com/livekit/livekit-server/pkg/rtc"
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/protocol/datatrack"
 )
 
 type SignalRequestHandler func(msg *livekit.SignalRequest) error

test/client/datatrack_remote.go

@@ -2,7 +2,7 @@ package client
 
 import (
 	"github.com/frostbyte73/core"
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
+	"github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"go.uber.org/atomic"

test/client/datatrack_writer.go

@@ -19,8 +19,9 @@ import (
 	"math/rand"
 	"time"
 
-	"github.com/livekit/livekit-server/pkg/rtc/datatrack"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/protocol/datatrack"
+	"github.com/livekit/protocol/datatrack/datatracktest"
 	"github.com/livekit/protocol/logger"
 )
 
@@ -59,7 +60,7 @@ func (d *dataTrackWriter) writeFrames() {
 			return
 
 		default:
-			packets := datatrack.GenerateRawDataPackets(d.handle, seqNum, frameNum, 1, rand.Intn(2048)+1, 100*time.Millisecond)
+			packets := datatracktest.GenerateRawDataPackets(d.handle, seqNum, frameNum, 1, rand.Intn(2048)+1, 100*time.Millisecond)
 			for _, packet := range packets {
 				if err := d.transport.SendDataTrackMessage(packet); err != nil {
 					logger.Errorw("could not send data track packet", err)