diff --git a/datatrack_integration_test.go b/datatrack_integration_test.go new file mode 100644 index 00000000..4e937a7a --- /dev/null +++ b/datatrack_integration_test.go @@ -0,0 +1,438 @@ +// 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. +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}) + } + + 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() + 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. +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") + }) + } +} 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()