diff --git a/go.mod b/go.mod index 9c2d915b9..04ce5dd4a 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 78af2b233..515944102 100644 --- a/go.sum +++ b/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= diff --git a/pkg/rtc/datadowntrack.go b/pkg/rtc/datadowntrack.go index 45f5e19a8..a1b0fdc35 100644 --- a/pkg/rtc/datadowntrack.go +++ b/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" ) diff --git a/pkg/rtc/datatrack.go b/pkg/rtc/datatrack.go index 2f3629b47..1d93c296d 100644 --- a/pkg/rtc/datatrack.go +++ b/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" diff --git a/pkg/rtc/datatrack/extension_participant_sid.go b/pkg/rtc/datatrack/extension_participant_sid.go deleted file mode 100644 index 35ada18f1..000000000 --- a/pkg/rtc/datatrack/extension_participant_sid.go +++ /dev/null @@ -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 -} diff --git a/pkg/rtc/datatrack/extension_participant_sid_test.go b/pkg/rtc/datatrack/extension_participant_sid_test.go deleted file mode 100644 index 7128e8bdc..000000000 --- a/pkg/rtc/datatrack/extension_participant_sid_test.go +++ /dev/null @@ -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()) -} diff --git a/pkg/rtc/datatrack/packet.go b/pkg/rtc/datatrack/packet.go deleted file mode 100644 index 42fe1eef2..000000000 --- a/pkg/rtc/datatrack/packet.go +++ /dev/null @@ -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 -} diff --git a/pkg/rtc/datatrack/packet_test.go b/pkg/rtc/datatrack/packet_test.go deleted file mode 100644 index 8f1bc672b..000000000 --- a/pkg/rtc/datatrack/packet_test.go +++ /dev/null @@ -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) - }) -} diff --git a/pkg/rtc/datatrack/testutils.go b/pkg/rtc/datatrack/testutils.go deleted file mode 100644 index e7df0bbd9..000000000 --- a/pkg/rtc/datatrack/testutils.go +++ /dev/null @@ -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 -} diff --git a/pkg/rtc/datatrack_stats.go b/pkg/rtc/datatrack_stats.go index 5bde6f57f..37ae15640 100644 --- a/pkg/rtc/datatrack_stats.go +++ b/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" ) diff --git a/pkg/rtc/participant_data_track.go b/pkg/rtc/participant_data_track.go index eff47b82b..23a34134f 100644 --- a/pkg/rtc/participant_data_track.go +++ b/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" diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 92bc9218f..76224e731 100644 --- a/pkg/rtc/room.go +++ b/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 ( diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 0393177a9..bfd06e322 100644 --- a/pkg/rtc/types/interfaces.go +++ b/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" ) diff --git a/pkg/rtc/types/typesfakes/fake_data_track.go b/pkg/rtc/types/typesfakes/fake_data_track.go index 07a49f9b1..ddd512d92 100644 --- a/pkg/rtc/types/typesfakes/fake_data_track.go +++ b/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" ) diff --git a/pkg/rtc/types/typesfakes/fake_data_track_sender.go b/pkg/rtc/types/typesfakes/fake_data_track_sender.go index a9a69486a..f45c28e82 100644 --- a/pkg/rtc/types/typesfakes/fake_data_track_sender.go +++ b/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" ) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index e0900c54f..cd94f6f26 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/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" diff --git a/pkg/rtc/types/typesfakes/fake_local_participant_listener.go b/pkg/rtc/types/typesfakes/fake_local_participant_listener.go index 4c0a3a07a..7f6912626 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant_listener.go +++ b/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" ) diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 652c8bf14..df9f2255d 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/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" diff --git a/pkg/rtc/types/typesfakes/fake_participant_listener.go b/pkg/rtc/types/typesfakes/fake_participant_listener.go index a05a3e0f5..ac1b05bea 100644 --- a/pkg/rtc/types/typesfakes/fake_participant_listener.go +++ b/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" ) diff --git a/pkg/rtc/updatatrackmanager.go b/pkg/rtc/updatatrackmanager.go index c3af8dc50..dbf6c5a32 100644 --- a/pkg/rtc/updatatrackmanager.go +++ b/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" ) diff --git a/test/client/client.go b/test/client/client.go index bd8c76f20..7a74c4841 100644 --- a/test/client/client.go +++ b/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 diff --git a/test/client/datatrack_remote.go b/test/client/datatrack_remote.go index 2d2f66365..97097620a 100644 --- a/test/client/datatrack_remote.go +++ b/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" diff --git a/test/client/datatrack_writer.go b/test/client/datatrack_writer.go index 82c6a188b..aaadde9f6 100644 --- a/test/client/datatrack_writer.go +++ b/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)