From ead28ea6ebe02d5de2e8261e006c6460e5f0c2e4 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Thu, 3 Sep 2026 16:42:28 -0700 Subject: [PATCH 1/5] Add data track end-to-end tests --- datatrack_integration_test.go | 455 ++++++++++++++++++++++++++++++++++ 1 file changed, 455 insertions(+) create mode 100644 datatrack_integration_test.go diff --git a/datatrack_integration_test.go b/datatrack_integration_test.go new file mode 100644 index 00000000..02f8126f --- /dev/null +++ b/datatrack_integration_test.go @@ -0,0 +1,455 @@ +// Copyright 2026 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 lksdk + +import ( + "bytes" + "context" + "fmt" + "testing" + "time" + + "github.com/google/uuid" + "github.com/livekit/protocol/auth" + "github.com/stretchr/testify/require" + + "github.com/livekit/server-sdk-go/v2/datatrack" + "github.com/livekit/server-sdk-go/v2/e2ee" +) + +// testRoom is one connection to the shared test room; published is fed by OnDataTrackPublished. +type testRoom struct { + room *Room + published <-chan *datatrack.RemoteTrack +} + +// testRoomOptions configure one connection, the counterpart of Rust's TestRoomOptions. +type testRoomOptions struct { + grants auth.VideoGrant + connect []ConnectOption +} + +func defaultTestRoomOptions() testRoomOptions { + return testRoomOptions{grants: auth.VideoGrant{RoomJoin: true}} +} + +// testRooms creates count connections to a shared room. +func testRooms(t *testing.T, count int) []testRoom { + options := make([]testRoomOptions, count) + for i := range options { + options[i] = defaultTestRoomOptions() + } + return testRoomsWithOptions(t, options...) +} + +// testRoomsWithOptions creates one connection per option to a shared room and waits until every +// participant sees the others. +func testRoomsWithOptions(t *testing.T, options ...testRoomOptions) []testRoom { + t.Helper() + roomName := fmt.Sprintf("test_room_%s", uuid.NewString()) + + rooms := make([]testRoom, 0, len(options)) + for id, option := range options { + option.grants.Room = roomName + token, err := auth.NewAccessToken(apiKey, apiSecret). + SetValidFor(30 * time.Minute). + SetVideoGrant(&option.grants). + SetIdentity(fmt.Sprintf("p%d", id)). + SetName(fmt.Sprintf("Participant %d", id)). + ToJWT() + require.NoError(t, err, "Failed to generate JWT") + + published := make(chan *datatrack.RemoteTrack, 16) + callback := &RoomCallback{ParticipantCallback: ParticipantCallback{ + OnDataTrackPublished: func(track *datatrack.RemoteTrack, _ *RemoteParticipant) { + published <- track + }, + }} + room, err := ConnectToRoomWithToken(host, token, callback, option.connect...) + require.NoError(t, err, "Failed to connect to room") + t.Cleanup(room.Disconnect) + rooms = append(rooms, testRoom{room: room, published: published}) + } + + // Wait for participant visibility across all room connections. When using a local SFU, this + // takes significantly longer and can lead to intermittently failing tests. + allConnected := time.Now() + require.Eventually(t, func() bool { + for _, r := range rooms { + if len(r.room.GetRemoteParticipants()) != len(rooms)-1 { + return false + } + } + return true + }, 5*time.Second, 10*time.Millisecond, "Not all participants became visible") + t.Logf("All participants visible after %v", time.Since(allConnected)) + + return rooms +} + +// waitForRemoteTrack waits for the first remote data track to be published. +func waitForRemoteTrack(t *testing.T, r testRoom) *datatrack.RemoteTrack { + t.Helper() + select { + case track := <-r.published: + return track + case <-time.After(5 * time.Second): + t.Fatal("No track published") + return nil + } +} + +// receiveFrame waits for the next frame on the stream. +func receiveFrame(t *testing.T, stream *datatrack.Stream, timeout time.Duration) datatrack.Frame { + t.Helper() + select { + case frame, ok := <-stream.Frames(): + require.True(t, ok, "Stream closed before a frame was received") + return frame + case <-time.After(timeout): + t.Fatal("No frame received") + return datatrack.Frame{} + } +} + +// pushEvery pushes payload on track at the given interval until ctx ends. +func pushEvery(ctx context.Context, track *datatrack.LocalTrack, payload []byte, interval time.Duration) { + for { + _ = track.TryPush(datatrack.Frame{Payload: payload}) + select { + case <-ctx.Done(): + return + case <-time.After(interval): + } + } +} + +func TestDataTrack(t *testing.T) { + for _, tc := range []struct { + name string + payloadLen int + }{ + {"single_packet", 8_192}, + {"multi_packet", 196_608}, + } { + t.Run(tc.name, func(t *testing.T) { + rooms := testRooms(t, 2) + pubRoom, subRoom := rooms[1], rooms[0] + pubIdentity := pubRoom.room.LocalParticipant.Identity() + + localTrack, err := pubRoom.room.LocalParticipant.PublishDataTrack(context.Background(), "my_track") + require.NoError(t, err) + t.Log("Track published") + + remoteTrack := waitForRemoteTrack(t, subRoom) + t.Logf("Got remote track: %s", remoteTrack.Info().SID) + + const payloadValue = 0xfa + + require.True(t, localTrack.IsPublished()) + require.False(t, localTrack.Info().UsesE2EE) + require.Equal(t, "my_track", localTrack.Info().Name) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + go pushEvery(ctx, localTrack, bytes.Repeat([]byte{payloadValue}, tc.payloadLen), 50*time.Millisecond) + + require.True(t, remoteTrack.IsPublished()) + require.False(t, remoteTrack.Info().UsesE2EE) + require.Equal(t, "my_track", remoteTrack.Info().Name) + require.Equal(t, pubIdentity, remoteTrack.PublisherIdentity()) + + stream, err := remoteTrack.Subscribe(ctx) + require.NoError(t, err) + defer stream.Close() + + frame := receiveFrame(t, stream, 15*time.Second) + require.Len(t, frame.Payload, tc.payloadLen) + require.Equal(t, bytes.Repeat([]byte{payloadValue}, tc.payloadLen), frame.Payload) + require.Nil(t, frame.UserTimestamp) + require.True(t, remoteTrack.IsPublished()) + }) + } +} + +func TestDataTrack_PublishManyTracks(t *testing.T) { + const trackCount = 256 + + room := testRooms(t, 1)[0].room + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + tracks := make([]*datatrack.LocalTrack, 0, trackCount) + start := time.Now() + for idx := range trackCount { + name := fmt.Sprintf("track_%d", idx) + track, err := room.LocalParticipant.PublishDataTrack(ctx, name) + require.NoError(t, err) + + require.True(t, track.IsPublished()) + require.Equal(t, name, track.Info().Name) + + tracks = append(tracks, track) + } + elapsed := time.Since(start) + t.Logf("Publishing %d tracks took %v (average %v per track)", trackCount, elapsed, elapsed/trackCount) + + for _, track := range tracks { + // Publish a single large frame per track. + require.NoError(t, track.TryPush(datatrack.Frame{Payload: bytes.Repeat([]byte{0xfa}, 196_608)})) + } +} + +func TestDataTrack_PublishUnauthorized(t *testing.T) { + options := defaultTestRoomOptions() + options.grants.SetCanPublishData(false) + room := testRoomsWithOptions(t, options)[0].room + + _, err := room.LocalParticipant.PublishDataTrack(context.Background(), "my_track") + require.ErrorIs(t, err, datatrack.ErrNotAllowed) +} + +func TestDataTrack_PublishDuplicateName(t *testing.T) { + room := testRooms(t, 1)[0].room + + first, err := room.LocalParticipant.PublishDataTrack(context.Background(), "first") + require.NoError(t, err) + defer first.Unpublish() + + _, err = room.LocalParticipant.PublishDataTrack(context.Background(), "first") + require.ErrorIs(t, err, datatrack.ErrDuplicateName) +} + +func TestDataTrack_PublishWithSchemaMetadata(t *testing.T) { + for _, tc := range []struct { + name string + schemaEncoding datatrack.SchemaEncoding + frameEncoding datatrack.FrameEncoding + }{ + {"well_known", datatrack.SchemaEncodingJSONSchema, datatrack.FrameEncodingJSON}, + {"custom", datatrack.CustomSchemaEncoding("a"), datatrack.CustomFrameEncoding("b")}, + } { + t.Run(tc.name, func(t *testing.T) { + rooms := testRooms(t, 2) + pubRoom, subRoom := rooms[1], rooms[0] + + schemaID := datatrack.SchemaID{Name: "my_schema", Encoding: tc.schemaEncoding} + localTrack, err := pubRoom.room.LocalParticipant.PublishDataTrack(context.Background(), "my_track", + datatrack.WithSchema(schemaID), datatrack.WithFrameEncoding(tc.frameEncoding)) + require.NoError(t, err) + require.Equal(t, &schemaID, localTrack.Info().Schema) + require.Equal(t, tc.frameEncoding, localTrack.Info().FrameEncoding) + + // The subscriber should observe the same schema and frame encoding metadata. + remoteTrack := waitForRemoteTrack(t, subRoom) + require.Equal(t, &schemaID, remoteTrack.Info().Schema) + require.Equal(t, tc.frameEncoding, remoteTrack.Info().FrameEncoding) + }) + } +} + +func TestDataTrack_E2EE(t *testing.T) { + const sharedSecret = "password" + payload := bytes.Repeat([]byte{0xfa}, 196_608) + + encrypted := func() testRoomOptions { + keyProvider := e2ee.NewExternalKeyProvider() + require.NoError(t, keyProvider.SetKeyFromPassphrase(sharedSecret, 0)) + options := defaultTestRoomOptions() + options.connect = []ConnectOption{WithDataEncryption(&EncryptionOptions{KeyProvider: keyProvider})} + return options + } + rooms := testRoomsWithOptions(t, encrypted(), encrypted()) + pubRoom, subRoom := rooms[1], rooms[0] + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + track, err := pubRoom.room.LocalParticipant.PublishDataTrack(ctx, "my_track") + require.NoError(t, err) + require.True(t, track.Info().UsesE2EE) + go pushEvery(ctx, track, payload, 125*time.Millisecond) + + remoteTrack := waitForRemoteTrack(t, subRoom) + require.True(t, remoteTrack.Info().UsesE2EE) + stream, err := remoteTrack.Subscribe(ctx) + require.NoError(t, err) + defer stream.Close() + + frame := receiveFrame(t, stream, 5*time.Second) + require.Equal(t, payload, frame.Payload) +} + +func TestDataTrack_PublishedState(t *testing.T) { + // How long to leave the track published. + const publishDuration = 500 * time.Millisecond + + rooms := testRooms(t, 2) + pubRoom, subRoom := rooms[1], rooms[0] + + track, err := pubRoom.room.LocalParticipant.PublishDataTrack(context.Background(), "my_track") + require.NoError(t, err) + require.True(t, track.IsPublished()) + + remoteTrack := waitForRemoteTrack(t, subRoom) + require.True(t, remoteTrack.IsPublished()) + + start := time.Now() + time.AfterFunc(publishDuration, track.Unpublish) + + select { + case <-remoteTrack.Unpublished(): + case <-time.After(5 * time.Second): + t.Fatal("Track was not unpublished") + } + elapsed := time.Since(start) + require.InDelta(t, publishDuration, elapsed, float64(20*time.Millisecond)) + require.False(t, remoteTrack.IsPublished()) +} + +func TestDataTrack_Resubscribe(t *testing.T) { + const iterations = 10 + + rooms := testRooms(t, 2) + pubRoom, subRoom := rooms[1], rooms[0] + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + track, err := pubRoom.room.LocalParticipant.PublishDataTrack(ctx, "my_track") + require.NoError(t, err) + go pushEvery(ctx, track, bytes.Repeat([]byte{0xfa}, 64), 50*time.Millisecond) + + remoteTrack := waitForRemoteTrack(t, subRoom) + + successfulSubscriptions := 0 + for range iterations { + stream, err := remoteTrack.Subscribe(ctx) + require.NoError(t, err) + + // Ensure we can at least get one frame. + frame := receiveFrame(t, stream, 5*time.Second) + require.NotEmpty(t, frame.Payload) + successfulSubscriptions++ + + stream.Close() + time.Sleep(50 * time.Millisecond) + } + require.Equal(t, iterations, successfulSubscriptions) +} + +func TestDataTrack_FrameWithUserTimestamp(t *testing.T) { + rooms := testRooms(t, 2) + pubRoom, subRoom := rooms[1], rooms[0] + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + track, err := pubRoom.room.LocalParticipant.PublishDataTrack(ctx, "my_track") + require.NoError(t, err) + go func() { + for { + _ = track.TryPush(datatrack.Frame{Payload: bytes.Repeat([]byte{0xfa}, 64), UserTimestamp: datatrack.UserTimestampNow()}) + select { + case <-ctx.Done(): + return + case <-time.After(50 * time.Millisecond): + } + } + }() + + remoteTrack := waitForRemoteTrack(t, subRoom) + stream, err := remoteTrack.Subscribe(ctx) + require.NoError(t, err) + defer stream.Close() + + // Ensure we can at least get one frame. + frame := receiveFrame(t, stream, 5*time.Second) + require.NotEmpty(t, frame.Payload) + duration, ok := frame.DurationSinceTimestamp() + require.True(t, ok, "Missing timestamp") + require.Less(t, duration, time.Second) +} + +// faultScenarios mirrors Rust's SignalReconnect and ForceTcp cases. Pion's ForceTCP simulation is a +// no-op, so the full reconnect is driven by SimulateNodeFailure instead. +var faultScenarios = []struct { + name string + scenario SimulateScenario + fullReconnect bool +}{ + {"signal_reconnect", SimulateSignalReconnect, false}, + {"full_reconnect", SimulateNodeFailure, true}, +} + +func TestDataTrack_SubscriberSideFault(t *testing.T) { + for _, tc := range faultScenarios { + t.Run(tc.name, func(t *testing.T) { + rooms := testRooms(t, 2) + pubRoom, subRoom := rooms[1], rooms[0] + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + track, err := pubRoom.room.LocalParticipant.PublishDataTrack(ctx, "my_track") + require.NoError(t, err) + go pushEvery(ctx, track, bytes.Repeat([]byte{0xfa}, 64), 50*time.Millisecond) + + remoteTrack := waitForRemoteTrack(t, subRoom) + stream, err := remoteTrack.Subscribe(ctx) + require.NoError(t, err) + defer stream.Close() + + // TODO: this should also evaluate what happens if a track subscription is removed + // during a full reconnect event. + subRoom.room.Simulate(tc.scenario) + require.True(t, remoteTrack.IsPublished()) + + // Ensure we can at least get one frame. + frame := receiveFrame(t, stream, 15*time.Second) + require.NotEmpty(t, frame.Payload) + }) + } +} + +func TestDataTrack_PublisherSideFault(t *testing.T) { + for _, tc := range faultScenarios { + t.Run(tc.name, func(t *testing.T) { + pubRoom := testRooms(t, 1)[0] + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + track, err := pubRoom.room.LocalParticipant.PublishDataTrack(ctx, "my_track") + require.NoError(t, err) + initialSID := track.Info().SID + + pubRoom.room.Simulate(tc.scenario) + require.True(t, track.IsPublished(), "Should still be reported as published") + + if tc.fullReconnect { + // Republish (full reconnect → new session → new sid) is async. Poll up to 8s for + // the new sid instead of unconditionally sleeping a fixed window. + require.Eventually(t, func() bool { + return track.Info().SID != initialSID + }, 8*time.Second, 100*time.Millisecond, "Should have new SID after full reconnect (still %s)", initialSID) + } + + require.True(t, track.IsPublished(), "Should still be reported as published") + require.NoError(t, track.TryPush(datatrack.Frame{Payload: bytes.Repeat([]byte{0xfa}, 64)}), "Should be able to push frame") + }) + } +} From 931f7bb0f66a8aa3930ebaf7e37b9dce2e6d974b Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 22 Sep 2026 16:55:45 -0700 Subject: [PATCH 2/5] Update API --- datatrack_integration_test.go | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/datatrack_integration_test.go b/datatrack_integration_test.go index 02f8126f..cb7ff94a 100644 --- a/datatrack_integration_test.go +++ b/datatrack_integration_test.go @@ -114,14 +114,11 @@ func waitForRemoteTrack(t *testing.T, r testRoom) *datatrack.RemoteTrack { // receiveFrame waits for the next frame on the stream. func receiveFrame(t *testing.T, stream *datatrack.Stream, timeout time.Duration) datatrack.Frame { t.Helper() - select { - case frame, ok := <-stream.Frames(): - require.True(t, ok, "Stream closed before a frame was received") - return frame - case <-time.After(timeout): - t.Fatal("No frame received") - return datatrack.Frame{} - } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + frame, err := stream.Next(ctx) + require.NoError(t, err, "No frame received") + return frame } // pushEvery pushes payload on track at the given interval until ctx ends. From 5f3a2045414fab03cc54fff2d954eac80974891b Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:20:39 -0700 Subject: [PATCH 3/5] Remove workaround for local SFU --- datatrack_integration_test.go | 16 +--------------- 1 file changed, 1 insertion(+), 15 deletions(-) diff --git a/datatrack_integration_test.go b/datatrack_integration_test.go index cb7ff94a..d33161fb 100644 --- a/datatrack_integration_test.go +++ b/datatrack_integration_test.go @@ -54,8 +54,7 @@ func testRooms(t *testing.T, count int) []testRoom { return testRoomsWithOptions(t, options...) } -// testRoomsWithOptions creates one connection per option to a shared room and waits until every -// participant sees the others. +// testRoomsWithOptions creates one connection per option to a shared room. func testRoomsWithOptions(t *testing.T, options ...testRoomOptions) []testRoom { t.Helper() roomName := fmt.Sprintf("test_room_%s", uuid.NewString()) @@ -83,19 +82,6 @@ func testRoomsWithOptions(t *testing.T, options ...testRoomOptions) []testRoom { rooms = append(rooms, testRoom{room: room, published: published}) } - // Wait for participant visibility across all room connections. When using a local SFU, this - // takes significantly longer and can lead to intermittently failing tests. - allConnected := time.Now() - require.Eventually(t, func() bool { - for _, r := range rooms { - if len(r.room.GetRemoteParticipants()) != len(rooms)-1 { - return false - } - } - return true - }, 5*time.Second, 10*time.Millisecond, "Not all participants became visible") - t.Logf("All participants visible after %v", time.Since(allConnected)) - return rooms } From 8f24d2ad976ca3ac5395abbae77ca0955d08db98 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:36:32 -0700 Subject: [PATCH 4/5] Lowercase assertion messages --- datatrack_integration_test.go | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/datatrack_integration_test.go b/datatrack_integration_test.go index d33161fb..4e937a7a 100644 --- a/datatrack_integration_test.go +++ b/datatrack_integration_test.go @@ -68,7 +68,7 @@ func testRoomsWithOptions(t *testing.T, options ...testRoomOptions) []testRoom { SetIdentity(fmt.Sprintf("p%d", id)). SetName(fmt.Sprintf("Participant %d", id)). ToJWT() - require.NoError(t, err, "Failed to generate JWT") + require.NoError(t, err, "failed to generate JWT") published := make(chan *datatrack.RemoteTrack, 16) callback := &RoomCallback{ParticipantCallback: ParticipantCallback{ @@ -77,7 +77,7 @@ func testRoomsWithOptions(t *testing.T, options ...testRoomOptions) []testRoom { }, }} room, err := ConnectToRoomWithToken(host, token, callback, option.connect...) - require.NoError(t, err, "Failed to connect to room") + require.NoError(t, err, "failed to connect to room") t.Cleanup(room.Disconnect) rooms = append(rooms, testRoom{room: room, published: published}) } @@ -92,7 +92,7 @@ func waitForRemoteTrack(t *testing.T, r testRoom) *datatrack.RemoteTrack { case track := <-r.published: return track case <-time.After(5 * time.Second): - t.Fatal("No track published") + t.Fatal("no track published") return nil } } @@ -103,7 +103,7 @@ func receiveFrame(t *testing.T, stream *datatrack.Stream, timeout time.Duration) ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() frame, err := stream.Next(ctx) - require.NoError(t, err, "No frame received") + require.NoError(t, err, "no frame received") return frame } @@ -134,10 +134,10 @@ func TestDataTrack(t *testing.T) { localTrack, err := pubRoom.room.LocalParticipant.PublishDataTrack(context.Background(), "my_track") require.NoError(t, err) - t.Log("Track published") + t.Log("track published") remoteTrack := waitForRemoteTrack(t, subRoom) - t.Logf("Got remote track: %s", remoteTrack.Info().SID) + t.Logf("got remote track: %s", remoteTrack.Info().SID) const payloadValue = 0xfa @@ -188,7 +188,7 @@ func TestDataTrack_PublishManyTracks(t *testing.T) { tracks = append(tracks, track) } elapsed := time.Since(start) - t.Logf("Publishing %d tracks took %v (average %v per track)", trackCount, elapsed, elapsed/trackCount) + t.Logf("publishing %d tracks took %v (average %v per track)", trackCount, elapsed, elapsed/trackCount) for _, track := range tracks { // Publish a single large frame per track. @@ -296,7 +296,7 @@ func TestDataTrack_PublishedState(t *testing.T) { select { case <-remoteTrack.Unpublished(): case <-time.After(5 * time.Second): - t.Fatal("Track was not unpublished") + t.Fatal("track was not unpublished") } elapsed := time.Since(start) require.InDelta(t, publishDuration, elapsed, float64(20*time.Millisecond)) @@ -363,7 +363,7 @@ func TestDataTrack_FrameWithUserTimestamp(t *testing.T) { frame := receiveFrame(t, stream, 5*time.Second) require.NotEmpty(t, frame.Payload) duration, ok := frame.DurationSinceTimestamp() - require.True(t, ok, "Missing timestamp") + require.True(t, ok, "missing timestamp") require.Less(t, duration, time.Second) } @@ -421,18 +421,18 @@ func TestDataTrack_PublisherSideFault(t *testing.T) { initialSID := track.Info().SID pubRoom.room.Simulate(tc.scenario) - require.True(t, track.IsPublished(), "Should still be reported as published") + require.True(t, track.IsPublished(), "should still be reported as published") if tc.fullReconnect { // Republish (full reconnect → new session → new sid) is async. Poll up to 8s for // the new sid instead of unconditionally sleeping a fixed window. require.Eventually(t, func() bool { return track.Info().SID != initialSID - }, 8*time.Second, 100*time.Millisecond, "Should have new SID after full reconnect (still %s)", initialSID) + }, 8*time.Second, 100*time.Millisecond, "should have new SID after full reconnect (still %s)", initialSID) } - require.True(t, track.IsPublished(), "Should still be reported as published") - require.NoError(t, track.TryPush(datatrack.Frame{Payload: bytes.Repeat([]byte{0xfa}, 64)}), "Should be able to push frame") + require.True(t, track.IsPublished(), "should still be reported as published") + require.NoError(t, track.TryPush(datatrack.Frame{Payload: bytes.Repeat([]byte{0xfa}, 64)}), "should be able to push frame") }) } } From 6866b862a803a3ab9d1ee3bb5f9b807d0c31fdf8 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 20 Oct 2026 17:50:35 -0700 Subject: [PATCH 5/5] Upgrade livekit/protocol to v1.53.0 --- go.mod | 12 ++++++------ go.sum | 24 ++++++++++++------------ internal/datatrack/packet.go | 5 +++-- 3 files changed, 21 insertions(+), 20 deletions(-) diff --git a/go.mod b/go.mod index 45a79eb8..a4e6c622 100644 --- a/go.mod +++ b/go.mod @@ -10,7 +10,7 @@ require ( github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/media-sdk v0.1.1 github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0 - github.com/livekit/protocol v1.52.2-0.20261006000721-750fff5dfc0b + github.com/livekit/protocol v1.53.0 github.com/magefile/mage v1.17.2 github.com/moby/buildkit v0.33.1 github.com/moby/patternmatcher v0.6.1 @@ -82,10 +82,10 @@ require ( github.com/jfreymuth/vorbis v1.0.2 // indirect github.com/joho/godotenv v1.5.1 github.com/jxskiss/base62 v1.1.0 // indirect - github.com/klauspost/compress v1.19.2 // indirect + github.com/klauspost/compress v1.20.1 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/lithammer/shortuuid/v4 v4.3.0 // indirect - github.com/livekit/psrpc v0.8.0 // indirect + github.com/livekit/psrpc v0.8.2 // indirect github.com/maxbrunsfeld/counterfeiter/v6 v6.14.0 // indirect github.com/moby/locker v1.0.1 // indirect github.com/moby/sys/signal v0.7.1 // indirect @@ -124,15 +124,15 @@ require ( go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.28.0 go.uber.org/zap/exp v0.3.0 // indirect - golang.org/x/exp v0.0.0-20260603202125-055de637280b // indirect + golang.org/x/exp v0.0.0-20261005173118-76772065c9b0 // indirect golang.org/x/net v0.59.0 // indirect golang.org/x/sync v0.24.0 golang.org/x/sys v0.48.0 // indirect golang.org/x/text v0.42.0 // indirect golang.org/x/time v0.15.0 // indirect - golang.org/x/tools v0.50.0 // indirect + golang.org/x/tools v0.51.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8 // indirect google.golang.org/grpc v1.84.0 gopkg.in/hraban/opus.v2 v2.0.0-20230925203106-0188a62cb302 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 0f599f59..cd1ca1a8 100644 --- a/go.sum +++ b/go.sum @@ -123,8 +123,8 @@ github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= github.com/jxskiss/base62 v1.1.0 h1:A5zbF8v8WXx2xixnAKD2w+abC+sIzYJX+nxmhA6HWFw= github.com/jxskiss/base62 v1.1.0/go.mod h1:HhWAlUXvxKThfOlZbcuFzsqwtF5TcqS9ru3y5GfjWAc= -github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= -github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -139,10 +139,10 @@ github.com/livekit/media-sdk v0.1.1 h1:d0glitmWU5ncIgsW/6wG4Q4iOEcpwINGl85mQjwnK github.com/livekit/media-sdk v0.1.1/go.mod h1:TuYRjSepaakL6ATsM9V2VMuksewW1PlhA32BG7Pxty0= github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0 h1:BSLyHr3Lsl71bHGHsfiYNmbohugR2yoaf9HjAZeFJnc= github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.52.2-0.20261006000721-750fff5dfc0b h1:8F84iX/JTM5CG/5oDojrAudCkc6RBcjgUl7X407o2vc= -github.com/livekit/protocol v1.52.2-0.20261006000721-750fff5dfc0b/go.mod h1:t5CV/LCsWbI9zsSTHTJKDyLtByqvvVkioSP7jgqpIGA= -github.com/livekit/psrpc v0.8.0 h1:uo3x+DL8pK2OIaCxg/s9sLUO2cIocS5HwdPhhI+m+fQ= -github.com/livekit/psrpc v0.8.0/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q= +github.com/livekit/protocol v1.53.0 h1:O5lcOxwgEqyhczAxYd2IPEYuq039+7MXzjxjdLPid8o= +github.com/livekit/protocol v1.53.0/go.mod h1:kUTAI2TrTngGOiWP6MTAOkJpru6YYGd0MP9wzlDATy4= +github.com/livekit/psrpc v0.8.2 h1:7tbkhBJoYlK5Csfkdv1XYbZjyAQeaEZgupvIXNCv7Dc= +github.com/livekit/psrpc v0.8.2/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q= github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4= github.com/mackerelio/go-osstat v0.2.8/go.mod h1:SyS3XxKdoSKJnTGTkN5Yrh6VUQVuAURACfE6y+2DN4k= github.com/magefile/mage v1.17.2 h1:fyXVu1eadI8Ap1HCCNgEhJ5McIWiYhLR8uol64ZZc40= @@ -332,8 +332,8 @@ go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= -golang.org/x/exp v0.0.0-20260603202125-055de637280b h1:v1uXiEBHo8QA0LiGCo7UgHMzHT4Kdfpl2zmtH5vaP1Q= -golang.org/x/exp v0.0.0-20260603202125-055de637280b/go.mod h1:d2fgXJLVs4dYDHUk5lwMIfzRzSrWCfGZb0ZqeLa/Vcw= +golang.org/x/exp v0.0.0-20261005173118-76772065c9b0 h1:qUp3i6M39kccSm3OMMIE3tvs7YQWO9rFeCKdk1sSaFk= +golang.org/x/exp v0.0.0-20261005173118-76772065c9b0/go.mod h1:SRjTzy5WcYqRRqquLUvHeeovSwLO0CZ3xcltZNO4Fck= golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= @@ -349,14 +349,14 @@ golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= -golang.org/x/tools v0.50.0 h1:c2ifzfcuY7L90lZ2aKd8S4K2NpASF08SZx9ZuJkHmSU= -golang.org/x/tools v0.50.0/go.mod h1:7ulVMw3831Mwi5EZD6RomGyffr4VFjuNYXf2BbCEAV0= +golang.org/x/tools v0.51.0 h1:k4Xc/1Om9jwkBJBo4NVLMSARBoWtK10mx+W5BnXCeAI= +golang.org/x/tools v0.51.0/go.mod h1:9eEncMayCV6zRMGhR5eZEC2iBx98qWcF1HZ9Z7wJOoA= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d h1:FarXi840EJWSHYTN3ERkADbPWjl307+FGrA22KAVjjc= google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d/go.mod h1:K/+WGbmBY7aNW1HDw1fJnKYo10i0DkAX6pows00dLig= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d h1:IL4hdHzcUv2l/gcg98/Rj3FbtE6axwqslOW8SW0C+S0= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8 h1:0SXBqli5dnuCKMjI/X3FM226zEaEa0GUcSCJtN1vL2o= +google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8/go.mod h1:OaIUM3+LpYcK2GXM4FTmhWoIq371Owdr+Cc7/BsYHHc= google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= diff --git a/internal/datatrack/packet.go b/internal/datatrack/packet.go index 08a1d0de..5c48ef07 100644 --- a/internal/datatrack/packet.go +++ b/internal/datatrack/packet.go @@ -19,6 +19,7 @@ import ( "fmt" dtp "github.com/livekit/protocol/datatrack" + "github.com/livekit/protocol/livekit" ) const supportedVersion = 0 @@ -86,12 +87,12 @@ func extensionsOf(h *dtp.Header) Extensions { var extensions Extensions for _, ext := range h.Extensions { switch ext.ID() { - case dtp.ExtensionE2EEID: + case uint8(livekit.DataTrackExtensionID_DTEI_E2EE): var e2ee dtp.ExtensionE2EE if e2ee.Unmarshal(ext) == nil { extensions.E2EE = &e2ee } - case dtp.ExtensionUserTimestampID: + case uint8(livekit.DataTrackExtensionID_DTEI_USER_TIMESTAMP): var userTimestamp dtp.ExtensionUserTimestamp if userTimestamp.Unmarshal(ext) == nil { timestamp := userTimestamp.Timestamp()