diff --git a/callback.go b/callback.go index ea0978fc..226cbd2f 100644 --- a/callback.go +++ b/callback.go @@ -18,6 +18,7 @@ import ( "github.com/pion/webrtc/v4" "github.com/livekit/protocol/livekit" + "github.com/livekit/server-sdk-go/v2/datatrack" ) // ParticipantAttributesChangedFunc is callback for Participant attribute change event. @@ -44,6 +45,8 @@ type ParticipantCallback struct { OnTrackSubscriptionFailed func(sid string, rp *RemoteParticipant) OnTrackPublished func(publication *RemoteTrackPublication, rp *RemoteParticipant) OnTrackUnpublished func(publication *RemoteTrackPublication, rp *RemoteParticipant) + OnDataTrackPublished func(track *datatrack.RemoteTrack, rp *RemoteParticipant) + OnDataTrackUnpublished func(track *datatrack.RemoteTrack, rp *RemoteParticipant) OnDataReceived func(data []byte, params DataReceiveParams) // Deprecated: Use OnDataPacket instead OnDataPacket func(data DataPacket, params DataReceiveParams) OnTranscriptionReceived func(transcriptionSegments []*TranscriptionSegment, p Participant, publication TrackPublication) @@ -66,6 +69,8 @@ func NewParticipantCallback() *ParticipantCallback { OnTrackSubscriptionFailed: func(sid string, rp *RemoteParticipant) {}, OnTrackPublished: func(publication *RemoteTrackPublication, rp *RemoteParticipant) {}, OnTrackUnpublished: func(publication *RemoteTrackPublication, rp *RemoteParticipant) {}, + OnDataTrackPublished: func(track *datatrack.RemoteTrack, rp *RemoteParticipant) {}, + OnDataTrackUnpublished: func(track *datatrack.RemoteTrack, rp *RemoteParticipant) {}, OnDataReceived: func(data []byte, params DataReceiveParams) {}, OnDataPacket: func(data DataPacket, params DataReceiveParams) {}, OnTranscriptionReceived: func(transcriptionSegments []*TranscriptionSegment, p Participant, publication TrackPublication) {}, @@ -113,6 +118,12 @@ func (cb *ParticipantCallback) Merge(other *ParticipantCallback) { if other.OnTrackUnpublished != nil { cb.OnTrackUnpublished = other.OnTrackUnpublished } + if other.OnDataTrackPublished != nil { + cb.OnDataTrackPublished = other.OnDataTrackPublished + } + if other.OnDataTrackUnpublished != nil { + cb.OnDataTrackUnpublished = other.OnDataTrackUnpublished + } if other.OnDataReceived != nil { cb.OnDataReceived = other.OnDataReceived } diff --git a/datatrack/datatrack.go b/datatrack/datatrack.go index bf38b992..843a1bab 100644 --- a/datatrack/datatrack.go +++ b/datatrack/datatrack.go @@ -12,8 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package datatrack is the public API for data tracks. Publish a track with -// [github.com/livekit/server-sdk-go/v2.LocalParticipant.PublishDataTrack]. +// Package datatrack is the public API for data tracks. package datatrack import ( @@ -23,6 +22,10 @@ import ( type ( // LocalTrack is a data track published by the local participant. LocalTrack = backend.LocalTrack + // RemoteTrack is a data track published by a remote participant. + RemoteTrack = backend.RemoteTrack + // Stream delivers the frames of a subscription. + Stream = backend.Stream // Info describes a published data track. Info = backend.Info // Frame is a frame published on a data track. @@ -33,6 +36,13 @@ type ( // PublishOption customizes a publication. PublishOption = backend.PublishOption + // SubscribeOptions configure a subscription. + SubscribeOptions = backend.SubscribeOptions + // SubscribeOption customizes a subscription. + SubscribeOption = backend.SubscribeOption + // PipelineOptions configure how a remote track's packets are reassembled. + PipelineOptions = backend.PipelineOptions + // SchemaID identifies a data track schema. SchemaID = backend.SchemaID // SchemaEncoding is the encoding of a schema definition: a WellKnownSchemaEncoding or a @@ -53,14 +63,15 @@ type ( ) var ( - ErrNotAllowed = backend.ErrNotAllowed - ErrDuplicateName = backend.ErrDuplicateName - ErrInvalidName = backend.ErrInvalidName - ErrLimitReached = backend.ErrLimitReached - ErrPublishTimeout = backend.ErrPublishTimeout - ErrDisconnected = backend.ErrDisconnected - ErrUnpublished = backend.ErrUnpublished - ErrQueueFull = backend.ErrQueueFull + ErrNotAllowed = backend.ErrNotAllowed + ErrDuplicateName = backend.ErrDuplicateName + ErrInvalidName = backend.ErrInvalidName + ErrLimitReached = backend.ErrLimitReached + ErrPublishTimeout = backend.ErrPublishTimeout + ErrSubscribeTimeout = backend.ErrSubscribeTimeout + ErrDisconnected = backend.ErrDisconnected + ErrUnpublished = backend.ErrUnpublished + ErrQueueFull = backend.ErrQueueFull ErrSchemaMissingFrameEncoding = backend.ErrSchemaMissingFrameEncoding ErrSchemaMissingID = backend.ErrSchemaMissingID @@ -101,6 +112,12 @@ func WithFrameEncoding(encoding FrameEncoding) PublishOption { return backend.WithFrameEncoding(encoding) } +// WithBufferSize sets the number of received frames buffered for the subscriber. It has no +// effect when the track already has an active subscription. +func WithBufferSize(frames int) SubscribeOption { + return backend.WithBufferSize(frames) +} + // UserTimestampNow returns the current Unix time in milliseconds, the interpretation // Frame.DurationSinceTimestamp expects. func UserTimestampNow() *uint64 { diff --git a/examples/datatrack/subscriber/main.go b/examples/datatrack/subscriber/main.go new file mode 100644 index 00000000..8c4c69a2 --- /dev/null +++ b/examples/datatrack/subscriber/main.go @@ -0,0 +1,90 @@ +package main + +import ( + "context" + "fmt" + "os" + "os/signal" + "syscall" + + "github.com/livekit/protocol/logger" + lksdk "github.com/livekit/server-sdk-go/v2" + "github.com/livekit/server-sdk-go/v2/datatrack" + "github.com/urfave/cli/v3" +) + +func main() { + logger.InitFromConfig(&logger.Config{Level: "info"}, "datatrack-subscriber") + lksdk.SetLogger(logger.GetLogger()) + + cmd := &cli.Command{ + Name: "datatrack-subscriber", + Usage: "subscribe to data tracks and log received frames", + Flags: []cli.Flag{ + &cli.StringFlag{Name: "host", Usage: "livekit server host", Required: true, Sources: cli.EnvVars("LIVEKIT_URL")}, + &cli.StringFlag{Name: "api-key", Usage: "livekit api key", Required: true, Sources: cli.EnvVars("LIVEKIT_API_KEY")}, + &cli.StringFlag{Name: "api-secret", Usage: "livekit api secret", Required: true, Sources: cli.EnvVars("LIVEKIT_API_SECRET")}, + &cli.StringFlag{Name: "room-name", Usage: "room name", Required: true}, + &cli.StringFlag{Name: "identity", Usage: "participant identity", Value: "subscriber"}, + }, + Action: run, + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + if err := cmd.Run(ctx, os.Args); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run(ctx context.Context, cmd *cli.Command) error { + // Subscribe to any published data tracks + callback := &lksdk.RoomCallback{ + ParticipantCallback: lksdk.ParticipantCallback{ + OnDataTrackPublished: func(track *datatrack.RemoteTrack, rp *lksdk.RemoteParticipant) { + go subscribe(ctx, track) + }, + }, + } + + room, err := lksdk.ConnectToRoom(cmd.String("host"), lksdk.ConnectInfo{ + APIKey: cmd.String("api-key"), + APISecret: cmd.String("api-secret"), + RoomName: cmd.String("room-name"), + ParticipantIdentity: cmd.String("identity"), + }, callback) + if err != nil { + return err + } + defer room.Disconnect() + + <-ctx.Done() + return nil +} + +// subscribe subscribes to the given data track and logs received frames. +func subscribe(ctx context.Context, track *datatrack.RemoteTrack) { + lgr := logger.GetLogger().WithValues("trackSID", track.Info().SID, "trackName", track.Info().Name, "publisher", track.PublisherIdentity()) + lgr.Infow("subscribing") + + stream, err := track.Subscribe(ctx) + if err != nil { + lgr.Warnw("failed to subscribe", err) + return + } + defer stream.Close() + + for { + frame, err := stream.Next(ctx) + if err != nil { + break + } + lgr.Infow("received frame", "bytes", len(frame.Payload)) + + if latency, ok := frame.DurationSinceTimestamp(); ok { + lgr.Infow("latency", "duration", latency) + } + } + lgr.Infow("unsubscribed") +} diff --git a/internal/datatrack/errors.go b/internal/datatrack/errors.go index cf465473..4c98fc4d 100644 --- a/internal/datatrack/errors.go +++ b/internal/datatrack/errors.go @@ -17,12 +17,13 @@ package datatrack import "errors" var ( - ErrNotAllowed = errors.New("data track publishing unauthorized") - ErrDuplicateName = errors.New("track name already taken") - ErrInvalidName = errors.New("track name invalid") - ErrLimitReached = errors.New("data track publication limit reached") - ErrPublishTimeout = errors.New("timed out publishing data track") - ErrDisconnected = errors.New("room disconnected") - ErrUnpublished = errors.New("track unpublished") - ErrQueueFull = errors.New("queue full") + ErrNotAllowed = errors.New("data track publishing unauthorized") + ErrDuplicateName = errors.New("track name already taken") + ErrInvalidName = errors.New("track name invalid") + ErrLimitReached = errors.New("data track publication limit reached") + ErrPublishTimeout = errors.New("timed out publishing data track") + ErrSubscribeTimeout = errors.New("timed out subscribing to data track") + ErrDisconnected = errors.New("room disconnected") + ErrUnpublished = errors.New("track unpublished") + ErrQueueFull = errors.New("queue full") ) diff --git a/internal/datatrack/helpers_test.go b/internal/datatrack/helpers_test.go new file mode 100644 index 00000000..660a3cf9 --- /dev/null +++ b/internal/datatrack/helpers_test.go @@ -0,0 +1,64 @@ +// 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 datatrack + +import ( + "bytes" + "testing" + "time" + + dtp "github.com/livekit/protocol/datatrack" +) + +// testE2EEPrefix marks payloads "encrypted" by prefixingEncryptor. +var testE2EEPrefix = []byte{0xde, 0xad, 0xbe, 0xef} + +// prefixingEncryptor marks payloads instead of encrypting them. +type prefixingEncryptor struct{} + +func (prefixingEncryptor) Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) { + return append(bytes.Clone(testE2EEPrefix), payload...), dtp.ExtensionE2EE{}, nil +} + +// prefixStrippingDecryptor undoes prefixingEncryptor. +type prefixStrippingDecryptor struct{} + +func (prefixStrippingDecryptor) Decrypt(payload []byte, _ dtp.ExtensionE2EE) ([]byte, error) { + return payload[len(testE2EEPrefix):], nil +} + +// expectEvent receives from ch or fails fast. The 500 ms timeout is generous +// enough to avoid flakes on slow CI while keeping failures quick. +func expectEvent[T any](t *testing.T, ch <-chan T) T { + t.Helper() + select { + case event := <-ch: + return event + case <-time.After(500 * time.Millisecond): + var zero T + t.Fatalf("timed out waiting for %T", zero) + return zero + } +} + +// expectNoEvent fails if ch delivers anything within the grace period. +func expectNoEvent[T any](t *testing.T, ch <-chan T) { + t.Helper() + select { + case event := <-ch: + t.Fatalf("unexpected event %v", event) + case <-time.After(50 * time.Millisecond): + } +} diff --git a/internal/datatrack/local_test.go b/internal/datatrack/local_test.go index e0b76941..34702cfe 100644 --- a/internal/datatrack/local_test.go +++ b/internal/datatrack/local_test.go @@ -21,7 +21,6 @@ import ( "testing" "time" - dtp "github.com/livekit/protocol/datatrack" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/utils/pointer" "github.com/stretchr/testify/require" @@ -56,27 +55,6 @@ func (f *fakeLocalTransport) SendFrame(frame FramePackets) { f.frames <- frame } -// prefixingEncryptor marks payloads instead of encrypting them. -type prefixingEncryptor struct{} - -func (prefixingEncryptor) Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) { - return append([]byte{0xde, 0xad, 0xbe, 0xef}, payload...), dtp.ExtensionE2EE{}, nil -} - -// expectEvent receives from ch or fails fast. The 500 ms timeout is generous -// enough to avoid flakes on slow CI while keeping failures quick. -func expectEvent[T any](t *testing.T, ch <-chan T) T { - t.Helper() - select { - case event := <-ch: - return event - case <-time.After(500 * time.Millisecond): - var zero T - t.Fatalf("timed out waiting for %T", zero) - return zero - } -} - // publishAsync calls Publish on another goroutine and delivers its outcome on the returned channel. func publishAsync(ctx context.Context, m *LocalManager, name string) <-chan publishResult { result := make(chan publishResult, 1) @@ -273,8 +251,8 @@ func TestLocalManager_PublishWithE2EE(t *testing.T) { packets := expectEvent(t, transport.frames) packet, err := parsePacket(packets[0]) require.NoError(t, err) - require.Equal(t, []byte{0xde, 0xad, 0xbe, 0xef}, packet.Payload[:4]) - require.Equal(t, []byte{1, 2, 3, 4, 5}, packet.Payload[4:]) + require.Equal(t, testE2EEPrefix, packet.Payload[:len(testE2EEPrefix)]) + require.Equal(t, []byte{1, 2, 3, 4, 5}, packet.Payload[len(testE2EEPrefix):]) require.NotNil(t, extensionsOf(&packet.Header).E2EE) } diff --git a/internal/datatrack/proto.go b/internal/datatrack/proto.go index 112f19f5..59443a9d 100644 --- a/internal/datatrack/proto.go +++ b/internal/datatrack/proto.go @@ -17,6 +17,7 @@ import ( "fmt" "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" ) var ( @@ -273,3 +274,41 @@ func publishRejectionFromRequestResponse(msg *livekit.RequestResponse) (rejectio } return rejection, true } + +// subscriptionUpdate asks the SFU to start or stop delivering a track's packets. +type subscriptionUpdate struct { + sid livekit.TrackID + subscribe bool +} + +func (u subscriptionUpdate) toProto() *livekit.UpdateDataSubscription { + return &livekit.UpdateDataSubscription{Updates: []*livekit.UpdateDataSubscription_Update{{ + TrackSid: string(u.sid), + Subscribe: u.subscribe, + }}} +} + +// publicationUpdatesFromProto maps each remote participant to the data tracks it publishes. The +// local participant is skipped and a disconnected participant contributes an empty list. Tracks +// that fail to convert are dropped with a warning. +func publicationUpdatesFromProto(participants []*livekit.ParticipantInfo, localIdentity string, log logger.Logger) map[string][]Info { + updates := make(map[string][]Info, len(participants)) + for _, participant := range participants { + if participant.GetIdentity() == localIdentity { + continue + } + infos := make([]Info, 0, len(participant.GetDataTracks())) + if participant.GetState() != livekit.ParticipantInfo_DISCONNECTED { + for _, msg := range participant.GetDataTracks() { + info, err := infoFromProto(msg) + if err != nil { + log.Warnw("ignoring invalid data track info", err, "participant", participant.GetIdentity()) + continue + } + infos = append(infos, info) + } + } + updates[participant.GetIdentity()] = infos + } + return updates +} diff --git a/internal/datatrack/proto_test.go b/internal/datatrack/proto_test.go index 116ab0c3..73390dff 100644 --- a/internal/datatrack/proto_test.go +++ b/internal/datatrack/proto_test.go @@ -18,6 +18,7 @@ import ( "testing" "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/stretchr/testify/require" ) @@ -141,3 +142,24 @@ func TestProto_PublishRejectionFromRequestResponse(t *testing.T) { require.Equal(t, trackHandle(1), rejection.handle) require.ErrorIs(t, rejection.err, ErrNotAllowed) } + +func TestProto_SubscriptionUpdateToProto(t *testing.T) { + update := subscriptionUpdate{sid: "DTR_1234", subscribe: true}.toProto() + require.Len(t, update.GetUpdates(), 1) + require.Equal(t, "DTR_1234", update.GetUpdates()[0].GetTrackSid()) + require.True(t, update.GetUpdates()[0].GetSubscribe()) +} + +func TestProto_PublicationUpdatesFromProto(t *testing.T) { + participants := []*livekit.ParticipantInfo{ + {Identity: "local", DataTracks: []*livekit.DataTrackInfo{{PubHandle: 1, Sid: "DTR_0000", Name: "mine"}}}, + {Identity: "publisher", DataTracks: []*livekit.DataTrackInfo{{PubHandle: 1, Sid: "DTR_1234", Name: "track1"}}}, + {Identity: "leaving", State: livekit.ParticipantInfo_DISCONNECTED, DataTracks: []*livekit.DataTrackInfo{{PubHandle: 1, Sid: "DTR_4567", Name: "stale"}}}, + } + + updates := publicationUpdatesFromProto(participants, "local", logger.GetLogger()) + require.Len(t, updates, 2) + require.Len(t, updates["publisher"], 1) + require.Equal(t, livekit.TrackID("DTR_1234"), updates["publisher"][0].SID) + require.Empty(t, updates["leaving"]) +} diff --git a/internal/datatrack/remote.go b/internal/datatrack/remote.go new file mode 100644 index 00000000..08fc638e --- /dev/null +++ b/internal/datatrack/remote.go @@ -0,0 +1,791 @@ +// 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 datatrack + +import ( + "context" + "errors" + "io" + "sync" + "sync/atomic" + "time" + + "github.com/frostbyte73/core" + dtp "github.com/livekit/protocol/datatrack" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +const ( + defaultBufferSize = 16 + defaultMaxPartialFrames = 1 + packetBufferCount = 16 + subscribeTimeout = 10 * time.Second +) + +// RemoteTransport carries what the remote manager produces: subscription requests to the SFU and +// publication events for the application. +type RemoteTransport interface { + SendUpdateSubscription(req *livekit.UpdateDataSubscription) error + OnTrackPublished(track *RemoteTrack) + OnTrackUnpublished(track *RemoteTrack) +} + +// Decryptor opens end-to-end encrypted frame payloads. +type Decryptor interface { + Decrypt(payload []byte, e2ee dtp.ExtensionE2EE) ([]byte, error) +} + +// SubscribeOptions configure a subscription. +type SubscribeOptions struct { + // BufferSize is the number of received frames buffered for the subscriber; older frames are + // dropped when it is exceeded. Zero is clamped to one. + BufferSize int +} + +// SubscribeOption customizes a subscription. +type SubscribeOption func(*SubscribeOptions) + +// WithBufferSize sets the number of received frames buffered for the subscriber. It has no +// effect when the track already has an active subscription. +func WithBufferSize(frames int) SubscribeOption { + return func(options *SubscribeOptions) { + options.BufferSize = frames + } +} + +// PipelineOptions configure how a remote track's packets are reassembled. +type PipelineOptions struct { + // MaxPartialFrames is the number of frames reassembled concurrently. Higher values tolerate + // more reordering at the cost of buffering. Zero is clamped to one. + MaxPartialFrames int +} + +type RemoteManagerParams struct { + Transport RemoteTransport + // Decryptor opens frames of tracks that use end-to-end encryption; nil disables decryption. + Decryptor Decryptor + Logger logger.Logger +} + +// RemoteManager tracks the data tracks published by remote participants and the local +// participant's subscriptions to them. Methods are safe for concurrent use. +type RemoteManager struct { + params RemoteManagerParams + log atomic.Pointer[logger.Logger] + + // mu guards descriptors and subHandles. Each RemoteTrack guards its own state: the manager + // reaches it through methods, which may take the track's lock while mu is held, and a track + // never calls into the manager while holding its own lock. Methods suffixed Locked expect mu. + mu sync.Mutex + descriptors map[livekit.TrackID]*RemoteTrack + subHandles map[trackHandle]*RemoteTrack +} + +func NewRemoteManager(params RemoteManagerParams) *RemoteManager { + if params.Logger == nil { + params.Logger = logger.GetLogger() + } + m := &RemoteManager{ + params: params, + descriptors: make(map[livekit.TrackID]*RemoteTrack), + subHandles: make(map[trackHandle]*RemoteTrack), + } + m.SetLogger(params.Logger) + return m +} + +// SetLogger replaces the logger given at construction. +func (m *RemoteManager) SetLogger(l logger.Logger) { + m.log.Store(&l) +} + +func (m *RemoteManager) logger() logger.Logger { + return *m.log.Load() +} + +// HandleParticipantUpdate applies the data tracks listed for each participant in a +// ParticipantUpdate; tracks a participant no longer lists are unpublished. +func (m *RemoteManager) HandleParticipantUpdate(participants []*livekit.ParticipantInfo, localIdentity string) { + m.handlePublicationUpdates(publicationUpdatesFromProto(participants, localIdentity, m.logger())) +} + +// HandleParticipantSnapshot applies a complete list of the room's participants, as carried by a +// JoinResponse. Publishers absent from the list are treated as gone. +func (m *RemoteManager) HandleParticipantSnapshot(participants []*livekit.ParticipantInfo, localIdentity string) { + updates := publicationUpdatesFromProto(participants, localIdentity, m.logger()) + m.mu.Lock() + for _, track := range m.descriptors { + if _, present := updates[track.publisherIdentity]; !present { + updates[track.publisherIdentity] = nil + } + } + m.mu.Unlock() + m.handlePublicationUpdates(updates) +} + +func (m *RemoteManager) handlePublicationUpdates(updates map[string][]Info) { + if len(updates) == 0 { + return + } + var published, unpublished, resubscribe []*RemoteTrack + + m.mu.Lock() + for publisherIdentity, infos := range updates { + sidsInUpdate := make(map[livekit.TrackID]struct{}, len(infos)) + for _, info := range infos { + sidsInUpdate[info.SID] = struct{}{} + if _, known := m.descriptors[info.SID]; known { + continue + } + if track, reassigned := m.reassignSIDLocked(publisherIdentity, info); reassigned { + if track != nil { + resubscribe = append(resubscribe, track) + } + continue + } + track := newRemoteTrack(m, info, publisherIdentity) + m.descriptors[info.SID] = track + published = append(published, track) + } + + for sid, track := range m.descriptors { + if track.publisherIdentity != publisherIdentity { + continue + } + if _, present := sidsInUpdate[sid]; !present { + delete(m.descriptors, sid) + if handle, wasActive := track.end(ErrUnpublished); wasActive { + delete(m.subHandles, handle) + } + unpublished = append(unpublished, track) + } + } + } + m.mu.Unlock() + + for _, track := range resubscribe { + track.resendSubscription() + } + for _, track := range published { + m.params.Transport.OnTrackPublished(track) + } + for _, track := range unpublished { + m.params.Transport.OnTrackUnpublished(track) + } +} + +// reassignSIDLocked detects a track republished under a new SID after its publisher's full +// reconnect: publisher identity and handle are stable across republications. It reports whether +// the SID was reassigned and, if a subscription must be re-requested under the new SID, the +// track to resubscribe. +func (m *RemoteManager) reassignSIDLocked(publisherIdentity string, info Info) (*RemoteTrack, bool) { + var track *RemoteTrack + for _, candidate := range m.descriptors { + if candidate.publisherIdentity == publisherIdentity && candidate.pubHandle == info.pubHandle { + track = candidate + break + } + } + if track == nil { + return nil, false + } + oldSID, subscribed, ok := track.reassignSID(info) + if !ok { + return nil, false + } + delete(m.descriptors, oldSID) + m.descriptors[info.SID] = track + if subscribed { + // the SFU does not carry subscriptions across the publisher's full reconnect + return track, true + } + return nil, true +} + +func schemaEqual(a, b *SchemaID) bool { + if a == nil || b == nil { + return a == b + } + return *a == *b +} + +// HandleSubscriberHandles records the handles the SFU assigned to requested subscriptions, which +// activates pending subscriptions. +func (m *RemoteManager) HandleSubscriberHandles(msg *livekit.DataTrackSubscriberHandles) { + mapping, err := subscriberHandlesFromProto(msg) + if err != nil { + m.logger().Warnw("ignoring invalid data track subscriber handles", err) + return + } + + m.mu.Lock() + defer m.mu.Unlock() + for handle, sid := range mapping { + track, known := m.descriptors[sid] + if !known { + m.logger().Warnw("subscriber handle for unknown data track", nil, "sid", sid) + continue + } + previous, replaced, ok := track.activate(handle) + if !ok { + m.logger().Warnw("subscriber handle for data track without subscription", nil, "sid", sid) + continue + } + if replaced { + delete(m.subHandles, previous) + } + m.subHandles[handle] = track + } +} + +// releaseHandle forgets the handle of a subscription the track ended, unless the SFU has since +// assigned the same handle to the track again. +func (m *RemoteManager) releaseHandle(track *RemoteTrack, handle trackHandle) { + m.mu.Lock() + defer m.mu.Unlock() + if m.subHandles[handle] == track && !track.usesHandle(handle) { + delete(m.subHandles, handle) + } +} + +// HandlePacket routes a packet received on the data channel to its track's subscribers. +func (m *RemoteManager) HandlePacket(data []byte) { + packet, err := parsePacket(data) + if err != nil { + m.logger().Warnw("dropping invalid data track packet", err) + return + } + + handle := trackHandle(packet.Handle) + m.mu.Lock() + track := m.subHandles[handle] + m.mu.Unlock() + if track == nil { + m.logger().Warnw("dropping data track packet without subscription", nil, "handle", packet.Handle) + return + } + track.deliver(handle, packet) +} + +// ResendSubscriptionUpdates re-requests every pending and active subscription after a full +// reconnect. +func (m *RemoteManager) ResendSubscriptionUpdates() { + m.mu.Lock() + tracks := make([]*RemoteTrack, 0, len(m.descriptors)) + for _, track := range m.descriptors { + tracks = append(tracks, track) + } + m.mu.Unlock() + + for _, track := range tracks { + track.resendSubscription() + } +} + +// Shutdown ends every subscription and marks every track unpublished. The manager can be used +// again for a new session. +func (m *RemoteManager) Shutdown() { + m.mu.Lock() + for _, track := range m.descriptors { + track.end(ErrDisconnected) + } + clear(m.descriptors) + clear(m.subHandles) + m.mu.Unlock() +} + +func (m *RemoteManager) sendSubscriptionUpdate(update subscriptionUpdate) error { + err := m.params.Transport.SendUpdateSubscription(update.toProto()) + if err != nil { + m.logger().Warnw("could not send data track subscription update", err, "sid", update.sid, "subscribe", update.subscribe) + } + return err +} + +type subscriptionState int + +const ( + subscriptionNone subscriptionState = iota + subscriptionPending + subscriptionActive +) + +type subscribeResult struct { + stream *Stream + err error +} + +// RemoteTrack is a data track published by a remote participant. Methods are safe for concurrent +// use. +type RemoteTrack struct { + manager *RemoteManager + publisherIdentity string + pubHandle trackHandle + unpublished core.Fuse + maxPartialFrames atomic.Int64 + + // sendMu orders the track's subscription updates as its subscription changed. It is taken + // before mu is released, so a later change cannot send ahead of an earlier one. + sendMu sync.Mutex + + // mu guards the fields below. It is never held while calling into the manager. Methods + // suffixed Locked expect it. + mu sync.Mutex + info Info + subscription subscriptionState + waiters []chan subscribeResult + bufferSize int + subHandle trackHandle + // packets feeds the worker goroutine of the active subscription + packets chan *dtp.Packet + streams map[*Stream]struct{} + // streamList holds an immutable snapshot of streams, read by the subscription's worker without + // the lock. Each subscription has its own, so a draining worker cannot reach later streams. + streamList *atomic.Pointer[[]*Stream] +} + +func newRemoteTrack(manager *RemoteManager, info Info, publisherIdentity string) *RemoteTrack { + track := &RemoteTrack{ + manager: manager, + publisherIdentity: publisherIdentity, + pubHandle: info.pubHandle, + info: info, + streams: make(map[*Stream]struct{}), + } + track.maxPartialFrames.Store(defaultMaxPartialFrames) + track.streamList = new(atomic.Pointer[[]*Stream]) + track.streamList.Store(&[]*Stream{}) + return track +} + +// Info returns a snapshot of the track's metadata. The SID changes when the publisher completes a +// full reconnect. +func (t *RemoteTrack) Info() Info { + t.mu.Lock() + defer t.mu.Unlock() + return t.info +} + +// PublisherIdentity is the identity of the participant who published the track. +func (t *RemoteTrack) PublisherIdentity() string { + return t.publisherIdentity +} + +// IsPublished reports whether the track is still published. +func (t *RemoteTrack) IsPublished() bool { + return !t.unpublished.IsBroken() +} + +// Unpublished is closed once the track is no longer published, whether by its publisher, the SFU, +// or disconnecting from the room. +func (t *RemoteTrack) Unpublished() <-chan struct{} { + return t.unpublished.Watch() +} + +// SetPipelineOptions configures how the track's packets are reassembled. The options apply to all +// current and future subscriptions and take effect with the next packet. +func (t *RemoteTrack) SetPipelineOptions(options PipelineOptions) { + if options.MaxPartialFrames < 1 { + t.manager.logger().Warnw("zero is not a valid value for MaxPartialFrames, using one", nil) + options.MaxPartialFrames = 1 + } + t.maxPartialFrames.Store(int64(options.MaxPartialFrames)) +} + +// Subscribe starts receiving the track's frames. Only the first subscription talks to the SFU; +// later ones share the pipeline and miss frames delivered before they were made. A 10 second +// deadline applies when ctx has none. Do not call it inline from the OnDataTrackPublished callback, +// which runs on the goroutine that delivers the SFU's response. +func (t *RemoteTrack) Subscribe(ctx context.Context, opts ...SubscribeOption) (*Stream, error) { + options := SubscribeOptions{BufferSize: defaultBufferSize} + for _, opt := range opts { + opt(&options) + } + if options.BufferSize < 1 { + t.manager.logger().Warnw("zero is not a valid buffer size, using one", nil) + options.BufferSize = 1 + } + + _, hasDeadline := ctx.Deadline() + ctx, cancel := withDefaultTimeout(ctx, subscribeTimeout) + defer cancel() + + t.mu.Lock() + if t.unpublished.IsBroken() { + t.mu.Unlock() + return nil, ErrUnpublished + } + if t.subscription == subscriptionActive { + stream := t.addStreamLocked(options.BufferSize) + t.mu.Unlock() + return stream, nil + } + waiter := make(chan subscribeResult, 1) + t.waiters = append(t.waiters, waiter) + var request *subscriptionUpdate + if t.subscription == subscriptionNone { + t.subscription = subscriptionPending + t.bufferSize = options.BufferSize + request = &subscriptionUpdate{sid: t.info.SID, subscribe: true} + } + if err := t.unlockAndSend(request); err != nil { + // fail the subscription the request was for, unless it already ended + t.mu.Lock() + if t.subscription == subscriptionPending { + t.subscription = subscriptionNone + for _, waiter := range t.waiters { + waiter <- subscribeResult{err: err} + } + t.waiters = nil + } + t.mu.Unlock() + } + + select { + case res := <-waiter: + return res.stream, res.err + case <-ctx.Done(): + var withdraw *subscriptionUpdate + t.mu.Lock() + t.removeWaiterLocked(waiter) + if t.subscription == subscriptionPending && len(t.waiters) == 0 { + t.subscription = subscriptionNone + withdraw = &subscriptionUpdate{sid: t.info.SID, subscribe: false} + } + t.unlockAndSend(withdraw) + select { + case res := <-waiter: + // the answer came first and was delivered under the lock + if res.stream != nil { + res.stream.Close() + } + default: + } + if !hasDeadline && errors.Is(ctx.Err(), context.DeadlineExceeded) { + return nil, ErrSubscribeTimeout + } + return nil, ctx.Err() + } +} + +func (t *RemoteTrack) removeWaiterLocked(waiter chan subscribeResult) { + for i, w := range t.waiters { + if w == waiter { + t.waiters = append(t.waiters[:i], t.waiters[i+1:]...) + return + } + } +} + +// reassignSID moves the track to the SID it was republished under after its publisher's full +// reconnect. It reports the SID replaced and whether a subscription must be re-requested, or false +// when the info differs in more than the SID. +func (t *RemoteTrack) reassignSID(info Info) (oldSID livekit.TrackID, subscribed bool, ok bool) { + t.mu.Lock() + defer t.mu.Unlock() + // other than the SID, the info should not have changed + if t.info.Name != info.Name || t.info.UsesE2EE != info.UsesE2EE || + !schemaEqual(t.info.Schema, info.Schema) || t.info.FrameEncoding != info.FrameEncoding { + t.manager.logger().Warnw("data track info mismatch, treating as new publication", nil, "sid", t.info.SID) + return "", false, false + } + oldSID = t.info.SID + t.manager.logger().Infow("data track SID reassigned", "oldSid", oldSID, "newSid", info.SID) + t.info.SID = info.SID + return oldSID, t.subscription != subscriptionNone, true +} + +// resendSubscription re-requests the track's pending or active subscription. +func (t *RemoteTrack) resendSubscription() { + t.mu.Lock() + var request *subscriptionUpdate + if t.subscription != subscriptionNone { + request = &subscriptionUpdate{sid: t.info.SID, subscribe: true} + } + t.unlockAndSend(request) +} + +// unlockAndSend releases mu and sends update, if any, ordered after the updates of earlier +// subscription changes. +func (t *RemoteTrack) unlockAndSend(update *subscriptionUpdate) error { + if update == nil { + t.mu.Unlock() + return nil + } + t.sendMu.Lock() + defer t.sendMu.Unlock() + t.mu.Unlock() + return t.manager.sendSubscriptionUpdate(*update) +} + +// usesHandle reports whether the track's active subscription is routed by handle. +func (t *RemoteTrack) usesHandle(handle trackHandle) bool { + t.mu.Lock() + defer t.mu.Unlock() + return t.subscription == subscriptionActive && t.subHandle == handle +} + +// activate records the handle the SFU assigned to the subscription. A pending subscription +// becomes active and every waiter receives a stream; an active one takes the new handle, as +// happens after a full reconnect, and the handle it replaces is returned. It reports false when +// the track has no subscription. +func (t *RemoteTrack) activate(handle trackHandle) (previous trackHandle, replaced bool, ok bool) { + t.mu.Lock() + defer t.mu.Unlock() + switch t.subscription { + case subscriptionNone: + return 0, false, false + case subscriptionActive: + previous, t.subHandle = t.subHandle, handle + return previous, true, true + } + + var decryptor Decryptor + if t.info.UsesE2EE { + decryptor = t.manager.params.Decryptor + } + t.packets = make(chan *dtp.Packet, packetBufferCount) + t.streamList = new(atomic.Pointer[[]*Stream]) + t.streamList.Store(&[]*Stream{}) + go t.runPipeline(t.packets, t.streamList, newRemotePipeline(decryptor, t.manager.logger())) + t.subHandle = handle + t.subscription = subscriptionActive + for _, waiter := range t.waiters { + waiter <- subscribeResult{stream: t.addStreamLocked(t.bufferSize)} + } + t.waiters = nil + return 0, false, true +} + +// deliver hands a packet to the subscription's worker. The packet is dropped when the subscription +// ended, or moved to another handle, since the manager looked the track up. +func (t *RemoteTrack) deliver(handle trackHandle, packet *dtp.Packet) { + t.mu.Lock() + defer t.mu.Unlock() + if t.subscription != subscriptionActive || t.subHandle != handle { + t.manager.logger().Warnw("dropping data track packet without subscription", nil, "handle", packet.Handle) + return + } + // the send happens under the lock so the channel is never closed underneath it + select { + case t.packets <- packet: + default: + t.manager.logger().Debugw("dropping data track packet, pipeline is behind", "sid", t.info.SID) + } +} + +// runPipeline reassembles the subscription's packets on its own goroutine, so tracks never wait on +// one another, and delivers completed frames to every stream. It ends when packets is closed. +func (t *RemoteTrack) runPipeline(packets <-chan *dtp.Packet, streamList *atomic.Pointer[[]*Stream], pipeline *remotePipeline) { + for packet := range packets { + frame, ok := pipeline.processPacket(packet, int(t.maxPartialFrames.Load())) + if !ok { + continue + } + for _, stream := range *streamList.Load() { + stream.push(frame) + } + } +} + +// deactivateLocked stops the worker of the active subscription. +func (t *RemoteTrack) deactivateLocked() { + if t.packets != nil { + close(t.packets) + t.packets = nil + } +} + +func (t *RemoteTrack) refreshStreamListLocked() { + list := make([]*Stream, 0, len(t.streams)) + for stream := range t.streams { + list = append(list, stream) + } + t.streamList.Store(&list) +} + +func (t *RemoteTrack) addStreamLocked(bufferSize int) *Stream { + stream := &Stream{track: t, buf: make([]Frame, bufferSize), ready: make(chan struct{}, 1)} + t.streams[stream] = struct{}{} + t.refreshStreamListLocked() + return stream +} + +// end marks the track unpublished, fails waiters with err, and closes every stream. It reports the +// handle of the active subscription it ended, if there was one. +func (t *RemoteTrack) end(err error) (handle trackHandle, wasActive bool) { + t.mu.Lock() + defer t.mu.Unlock() + handle, wasActive = t.subHandle, t.subscription == subscriptionActive + t.unpublished.Break() + for _, waiter := range t.waiters { + waiter <- subscribeResult{err: err} + } + t.waiters = nil + for stream := range t.streams { + stream.close() + } + clear(t.streams) + t.refreshStreamListLocked() + t.subscription = subscriptionNone + t.deactivateLocked() + return handle, wasActive +} + +// removeStream is the subscriber side of Stream.Close: the last stream to leave ends the SFU +// subscription. +func (t *RemoteTrack) removeStream(stream *Stream) { + var ( + withdraw *subscriptionUpdate + handle trackHandle + ) + t.mu.Lock() + stream.close() + if _, present := t.streams[stream]; present { + delete(t.streams, stream) + t.refreshStreamListLocked() + if len(t.streams) == 0 && t.subscription == subscriptionActive { + t.subscription = subscriptionNone + t.deactivateLocked() + handle = t.subHandle + withdraw = &subscriptionUpdate{sid: t.info.SID, subscribe: false} + } + } + t.unlockAndSend(withdraw) + + if withdraw != nil { + t.manager.releaseHandle(t, handle) + } +} + +// Stream delivers the frames of one subscription. +type Stream struct { + track *RemoteTrack + mu sync.Mutex + buf []Frame + head int + n int + ready chan struct{} + closed bool +} + +// Next blocks until a frame arrives. It returns io.EOF once the stream is closed, the track is +// unpublished, or the room disconnects, and ctx's error when ctx ends first. +func (s *Stream) Next(ctx context.Context) (Frame, error) { + for { + s.mu.Lock() + if s.n > 0 { + frame := s.buf[s.head] + s.buf[s.head] = Frame{} + s.head = (s.head + 1) % len(s.buf) + s.n-- + s.mu.Unlock() + return frame, nil + } + closed := s.closed + s.mu.Unlock() + if closed { + return Frame{}, io.EOF + } + select { + case <-s.ready: + case <-ctx.Done(): + return Frame{}, ctx.Err() + } + } +} + +// Close ends the subscription. It is safe to call more than once. +func (s *Stream) Close() { + s.track.removeStream(s) +} + +// push buffers a frame, dropping the oldest buffered frame when full. +func (s *Stream) push(frame Frame) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + if s.n == len(s.buf) { + s.buf[s.head] = frame + s.head = (s.head + 1) % len(s.buf) + } else { + s.buf[(s.head+s.n)%len(s.buf)] = frame + s.n++ + } + select { + case s.ready <- struct{}{}: + default: + } +} + +func (s *Stream) close() { + s.mu.Lock() + defer s.mu.Unlock() + if !s.closed { + s.closed = true + close(s.ready) + } +} + +// remotePipeline turns a subscription's packets back into frames. It is owned by one goroutine. +type remotePipeline struct { + decryptor Decryptor + log logger.Logger + depacketizer *depacketizer +} + +func newRemotePipeline(decryptor Decryptor, log logger.Logger) *remotePipeline { + return &remotePipeline{decryptor: decryptor, log: log, depacketizer: newDepacketizer()} +} + +// processPacket reports whether the packet completed a frame. +func (p *remotePipeline) processPacket(packet *dtp.Packet, maxPartialFrames int) (Frame, bool) { + result := p.depacketizer.push(*packet, depacketizerPushOptions{maxPartialFrames: maxPartialFrames}) + + if result.drop != nil { + p.log.Debugw("data track frame dropped", "reason", result.drop.Error()) + } + if result.frame == nil { + return Frame{}, false + } + + frame := Frame{Payload: result.frame.payload, UserTimestamp: result.frame.extensions.UserTimestamp} + if p.decryptor == nil { + return frame, true + } + e2ee := result.frame.extensions.E2EE + if e2ee == nil { + p.log.Errorw("dropping data track frame without E2EE extension", nil) + return Frame{}, false + } + payload, err := p.decryptor.Decrypt(frame.Payload, *e2ee) + if err != nil { + p.log.Errorw("dropping data track frame that failed to decrypt", err) + return Frame{}, false + } + frame.Payload = payload + return frame, true +} + +// withDefaultTimeout applies deadline d to ctx when it has none. The returned cancel must be called. +func withDefaultTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + if _, ok := ctx.Deadline(); ok { + return ctx, func() {} + } + return context.WithTimeout(ctx, d) +} diff --git a/internal/datatrack/remote_test.go b/internal/datatrack/remote_test.go new file mode 100644 index 00000000..db4d1a74 --- /dev/null +++ b/internal/datatrack/remote_test.go @@ -0,0 +1,654 @@ +// 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 datatrack + +import ( + "bytes" + "context" + "errors" + "io" + "testing" + "time" + + dtp "github.com/livekit/protocol/datatrack" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/stretchr/testify/require" +) + +// fakeRemoteTransport records what the manager sends and emits, in order. +type fakeRemoteTransport struct { + subscriptionUpdates chan *livekit.UpdateDataSubscription + published chan *RemoteTrack + unpublished chan *RemoteTrack + // sendErr, when set, fails subscription updates after recording them + sendErr error +} + +func newFakeRemoteTransport() *fakeRemoteTransport { + return &fakeRemoteTransport{ + subscriptionUpdates: make(chan *livekit.UpdateDataSubscription, 16), + published: make(chan *RemoteTrack, 16), + unpublished: make(chan *RemoteTrack, 16), + } +} + +func (f *fakeRemoteTransport) SendUpdateSubscription(req *livekit.UpdateDataSubscription) error { + f.subscriptionUpdates <- req + return f.sendErr +} + +func (f *fakeRemoteTransport) OnTrackPublished(track *RemoteTrack) { + f.published <- track +} + +func (f *fakeRemoteTransport) OnTrackUnpublished(track *RemoteTrack) { + f.unpublished <- track +} + +func expectFrame(t *testing.T, stream *Stream) Frame { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + frame, err := stream.Next(ctx) + require.NoError(t, err) + return frame +} + +func expectNoFrame(t *testing.T, stream *Stream) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _, err := stream.Next(ctx) + require.ErrorIs(t, err, context.DeadlineExceeded) +} + +func expectClosed(t *testing.T, stream *Stream) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err := stream.Next(ctx) + require.ErrorIs(t, err, io.EOF) +} + +// publishTrack simulates the SFU announcing a publication and returns the track handed to the +// application. +func publishTrack(t *testing.T, m *RemoteManager, transport *fakeRemoteTransport, publisherIdentity string, info Info) *RemoteTrack { + t.Helper() + m.handlePublicationUpdates(map[string][]Info{publisherIdentity: {info}}) + return expectEvent(t, transport.published) +} + +func expectSubscriptionUpdate(t *testing.T, transport *fakeRemoteTransport) (livekit.TrackID, bool) { + t.Helper() + msg := expectEvent(t, transport.subscriptionUpdates) + require.Len(t, msg.GetUpdates(), 1) + update := msg.GetUpdates()[0] + return livekit.TrackID(update.GetTrackSid()), update.GetSubscribe() +} + +// assignHandle simulates the SFU assigning a subscriber handle. +func assignHandle(m *RemoteManager, handle trackHandle, sid livekit.TrackID) { + m.HandleSubscriberHandles(&livekit.DataTrackSubscriberHandles{ + SubHandles: map[uint32]*livekit.DataTrackSubscriberHandles_PublishedDataTrack{ + uint32(handle): {TrackSid: string(sid)}, + }, + }) +} + +// subscribeAsync calls Subscribe on another goroutine and delivers its outcome on the returned channel. +func subscribeAsync(ctx context.Context, track *RemoteTrack, opts ...SubscribeOption) <-chan subscribeResult { + result := make(chan subscribeResult, 1) + go func() { + stream, err := track.Subscribe(ctx, opts...) + result <- subscribeResult{stream: stream, err: err} + }() + return result +} + +// subscribeAndActivate subscribes, answers the SFU request with handle, and returns the stream. +func subscribeAndActivate(t *testing.T, m *RemoteManager, transport *fakeRemoteTransport, track *RemoteTrack, handle trackHandle) *Stream { + t.Helper() + result := subscribeAsync(context.Background(), track) + + sid, subscribe := expectSubscriptionUpdate(t, transport) + require.True(t, subscribe) + require.Equal(t, track.Info().SID, sid) + + assignHandle(m, handle, sid) + + res := expectEvent(t, result) + require.NoError(t, res.err) + return res.stream +} + +func rawPacket(t *testing.T, marker FrameMarker, handle trackHandle, sequence, frameNumber uint16, payload []byte, extensions Extensions) []byte { + t.Helper() + header := testHeader() + marker.apply(&header) + header.Handle = uint16(handle) + header.SequenceNumber = sequence + header.FrameNumber = frameNumber + extensions.apply(&header) + packet := dtp.Packet{Header: header, Payload: payload} + raw, err := packet.Marshal() + require.NoError(t, err) + return raw +} + +func singlePacket(t *testing.T, handle trackHandle, payload []byte, extensions Extensions) []byte { + return rawPacket(t, FrameMarkerSingle, handle, 0, 0, payload, extensions) +} + +// pushInterleavedTwoFramePair pushes Start(frame1), Start(frame2), Final(frame1), Final(frame2) +// through the manager to exercise the depacketizer's concurrent partial frame handling. +func pushInterleavedTwoFramePair(t *testing.T, m *RemoteManager, handle trackHandle, frameOne, frameOneStart uint16, frameOnePayloads [2][]byte, frameTwo, frameTwoStart uint16, frameTwoPayloads [2][]byte) { + t.Helper() + push := func(frameNumber, sequence uint16, marker FrameMarker, payload []byte) { + m.HandlePacket(rawPacket(t, marker, handle, sequence, frameNumber, payload, Extensions{})) + } + push(frameOne, frameOneStart, FrameMarkerStart, frameOnePayloads[0]) + push(frameTwo, frameTwoStart, FrameMarkerStart, frameTwoPayloads[0]) + push(frameOne, frameOneStart+1, FrameMarkerFinal, frameOnePayloads[1]) + push(frameTwo, frameTwoStart+1, FrameMarkerFinal, frameTwoPayloads[1]) +} + +func TestRemotePipeline_ProcessPacket(t *testing.T) { + const payloadLen = 1024 + pipeline := newRemotePipeline(nil, logger.GetLogger()) + + header := testHeader() + FrameMarkerSingle.apply(&header) + packet := &dtp.Packet{Header: header, Payload: bytes.Repeat([]byte{0xab}, payloadLen)} + + frame, ok := pipeline.processPacket(packet, defaultMaxPartialFrames) + require.True(t, ok, "should return a frame") + require.Len(t, frame.Payload, payloadLen) +} + +func TestRemoteManager_Shutdown(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + track := publishTrack(t, m, transport, "id", Info{SID: "DTR_1234", pubHandle: 1, Name: "test"}) + pending := subscribeAsync(context.Background(), track) + expectSubscriptionUpdate(t, transport) + + m.Shutdown() + + require.ErrorIs(t, expectEvent(t, pending).err, ErrDisconnected) + require.False(t, track.IsPublished()) + expectEvent(t, track.Unpublished()) +} + +func TestRemoteManager_Subscribe(t *testing.T) { + publisherIdentity, trackName, trackSID := "publisher", "track", livekit.TrackID("DTR_1234") + subHandle := trackHandle(0x1234) + + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + // Simulate track published + track := publishTrack(t, m, transport, publisherIdentity, Info{SID: trackSID, pubHandle: 1, Name: trackName}) + require.True(t, track.IsPublished()) + require.Equal(t, trackName, track.Info().Name) + require.Equal(t, trackSID, track.Info().SID) + require.Equal(t, publisherIdentity, track.PublisherIdentity()) + + result := subscribeAsync(context.Background(), track) + + sid, subscribe := expectSubscriptionUpdate(t, transport) + require.True(t, subscribe) + require.Equal(t, trackSID, sid) + time.Sleep(20 * time.Millisecond) + + // Simulate SFU reply + assignHandle(m, subHandle, trackSID) + + res := expectEvent(t, result) + require.NoError(t, res.err) + require.NotNil(t, res.stream) +} + +func TestRemoteManager_TrackPublicationAddAndRemove(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + trackSID := livekit.TrackID("DTR_1234") + + // Simulate track published + track := publishTrack(t, m, transport, "identity1", Info{SID: trackSID, pubHandle: 1, Name: "test"}) + require.Equal(t, trackSID, track.Info().SID) + require.Equal(t, "test", track.Info().Name) + require.True(t, track.IsPublished()) + + // Simulate track unpublished + m.handlePublicationUpdates(map[string][]Info{"identity1": nil}) + + expectEvent(t, track.Unpublished()) + require.False(t, track.IsPublished()) + + unpublished := expectEvent(t, transport.unpublished) + require.Equal(t, trackSID, unpublished.Info().SID) +} + +func TestRemoteManager_SfuPublicationUpdatesIdempotent(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + info := Info{SID: "DTR_1234", pubHandle: 1, Name: "test"} + + // Simulate three identical publication updates + for range 3 { + m.handlePublicationUpdates(map[string][]Info{"identity1": {info}}) + } + + expectEvent(t, transport.published) + + // No second publication should appear + m.Shutdown() + expectNoEvent(t, transport.published) +} + +func TestRemoteManager_SidReassignmentDoesNotRepublish(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + pubHandle := trackHandle(7) + oldSID, newSID := livekit.TrackID("DTR_1234"), livekit.TrackID("DTR_5678") + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: oldSID, pubHandle: pubHandle, Name: "test"}) + require.Equal(t, oldSID, track.Info().SID) + + // Simulate publisher full reconnect: same track, new SID + m.handlePublicationUpdates(map[string][]Info{"id": {{SID: newSID, pubHandle: pubHandle, Name: "test"}}}) + + // No publish/unpublish should appear + m.Shutdown() + expectNoEvent(t, transport.published) + expectNoEvent(t, transport.unpublished) + require.Equal(t, newSID, track.Info().SID) +} + +func TestRemoteManager_SidReassignmentResubscribesActiveSubscription(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + pubHandle := trackHandle(7) + oldSID, newSID := livekit.TrackID("DTR_1234"), livekit.TrackID("DTR_5678") + oldSubHandle, newSubHandle := trackHandle(0x1001), trackHandle(0x1002) + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: oldSID, pubHandle: pubHandle, Name: "test"}) + + // Subscribe to the track + stream := subscribeAndActivate(t, m, transport, track, oldSubHandle) + + // Simulate publisher full reconnect: same track, new SID + m.handlePublicationUpdates(map[string][]Info{"id": {{SID: newSID, pubHandle: pubHandle, Name: "test"}}}) + + // Manager should re-subscribe under the new SID + sid, subscribe := expectSubscriptionUpdate(t, transport) + require.True(t, subscribe) + require.Equal(t, newSID, sid) + require.Equal(t, newSID, track.Info().SID) + require.True(t, track.IsPublished()) + + // Simulate SFU assigning a new subscriber handle + assignHandle(m, newSubHandle, newSID) + + // Frames received on the new handle reach the existing subscriber + m.HandlePacket(singlePacket(t, newSubHandle, []byte{1, 2, 3, 4, 5}, Extensions{})) + + frame := expectFrame(t, stream) + require.Equal(t, []byte{1, 2, 3, 4, 5}, frame.Payload) +} + +func TestRemoteManager_SubscribeReceivesFrame(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + trackSID, subHandle := livekit.TrackID("DTR_1234"), trackHandle(0x1234) + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test"}) + + // Subscribe to the track + stream := subscribeAndActivate(t, m, transport, track, subHandle) + + // Simulate receiving a single-frame packet + m.HandlePacket(singlePacket(t, subHandle, []byte{1, 2, 3, 4, 5}, Extensions{})) + + frame := expectFrame(t, stream) + require.Equal(t, []byte{1, 2, 3, 4, 5}, frame.Payload) +} + +func TestRemoteManager_SubscribeWithE2EE(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport, Decryptor: prefixStrippingDecryptor{}}) + + trackSID, subHandle := livekit.TrackID("DTR_1234"), trackHandle(0x1234) + + // Simulate track published (with e2ee) + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test", UsesE2EE: true}) + + // Subscribe to the track + stream := subscribeAndActivate(t, m, transport, track, subHandle) + + // Simulate receiving an encrypted single-frame packet + payload := append(bytes.Clone(testE2EEPrefix), 1, 2, 3, 4, 5) + m.HandlePacket(singlePacket(t, subHandle, payload, Extensions{E2EE: &dtp.ExtensionE2EE{}})) + + // Payload should have fake encryption prefix stripped by decryptor + frame := expectFrame(t, stream) + require.Equal(t, []byte{1, 2, 3, 4, 5}, frame.Payload) +} + +func TestRemoteManager_SubscribeFanOutToMultipleSubscribers(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + trackSID, subHandle := livekit.TrackID("DTR_1234"), trackHandle(0x1234) + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test"}) + + // First subscriber triggers SFU interaction + stream1 := subscribeAndActivate(t, m, transport, track, subHandle) + + // Additional subscribers attach directly (no further SFU interaction) + stream2, err := track.Subscribe(context.Background()) + require.NoError(t, err) + stream3, err := track.Subscribe(context.Background()) + require.NoError(t, err) + expectNoEvent(t, transport.subscriptionUpdates) + + // Simulate receiving a single-frame packet + m.HandlePacket(singlePacket(t, subHandle, []byte{1, 2, 3, 4, 5}, Extensions{})) + + // All subscribers should receive the same frame + for _, stream := range []*Stream{stream1, stream2, stream3} { + frame := expectFrame(t, stream) + require.Equal(t, []byte{1, 2, 3, 4, 5}, frame.Payload) + } +} + +func TestRemoteManager_SubscribeUnknownTrackFails(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + // A track that is no longer published cannot be subscribed to + track := publishTrack(t, m, transport, "id", Info{SID: "DTR_1234", pubHandle: 1, Name: "test"}) + m.handlePublicationUpdates(map[string][]Info{"id": nil}) + expectEvent(t, transport.unpublished) + + _, err := track.Subscribe(context.Background()) + require.ErrorIs(t, err, ErrUnpublished) +} + +// Should fail Subscribe with the error of a request that could not be sent, rather than time out. +func TestRemoteManager_SubscribeFailsWhenRequestNotSent(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + trackSID := livekit.TrackID("DTR_1234") + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test"}) + + // Subscribe while the signal connection is down + sendErr := errors.New("signal transport is not connected") + transport.sendErr = sendErr + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err := track.Subscribe(ctx) + require.ErrorIs(t, err, sendErr) + _, subscribe := expectSubscriptionUpdate(t, transport) + require.True(t, subscribe) + + // Once the connection is back, subscribing requests the subscription again + transport.sendErr = nil + result := subscribeAsync(context.Background(), track) + _, subscribe = expectSubscriptionUpdate(t, transport) + require.True(t, subscribe) + + assignHandle(m, 0x1234, trackSID) + res := expectEvent(t, result) + require.NoError(t, res.err) +} + +func TestRemoteManager_UnpublishTerminatesPendingSubscription(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + trackSID := livekit.TrackID("DTR_1234") + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test"}) + + // Subscribe (enters pending state) + result := subscribeAsync(context.Background(), track) + _, subscribe := expectSubscriptionUpdate(t, transport) + require.True(t, subscribe) + + // Simulate track unpublished before SFU assigns a handle + m.handlePublicationUpdates(map[string][]Info{"id": nil}) + + require.ErrorIs(t, expectEvent(t, result).err, ErrUnpublished) + + unpublished := expectEvent(t, transport.unpublished) + require.Equal(t, trackSID, unpublished.Info().SID) +} + +func TestRemoteManager_UnpublishTerminatesActiveSubscription(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + trackSID, subHandle := livekit.TrackID("DTR_1234"), trackHandle(0x1234) + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test"}) + + // Subscribe to the track + stream := subscribeAndActivate(t, m, transport, track, subHandle) + + // Simulate track unpublished while subscription is active + m.handlePublicationUpdates(map[string][]Info{"id": nil}) + + expectClosed(t, stream) + + unpublished := expectEvent(t, transport.unpublished) + require.Equal(t, trackSID, unpublished.Info().SID) +} + +func TestRemoteManager_AllSubscribersDroppedTerminatesSfuSubscription(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + trackSID, subHandle := livekit.TrackID("DTR_1234"), trackHandle(0x1234) + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test"}) + + // Subscribe to the track + stream := subscribeAndActivate(t, m, transport, track, subHandle) + + // Close the only subscriber + stream.Close() + + // Manager should request SFU to unsubscribe + sid, subscribe := expectSubscriptionUpdate(t, transport) + require.False(t, subscribe) + require.Equal(t, trackSID, sid) +} + +// Should send the unsubscribe of the last subscriber before a concurrent resubscribe, so the SFU +// does not cancel the new subscription. +func TestRemoteManager_ResubscribeAfterLastSubscriberDroppedKeepsOrder(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + trackSID, subHandle := livekit.TrackID("DTR_1234"), trackHandle(0x1234) + + // Simulate track published + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test"}) + + // Subscribe to the track + stream := subscribeAndActivate(t, m, transport, track, subHandle) + + // Hold the manager so closing the last subscriber stalls after ending the subscription + m.mu.Lock() + closed := make(chan struct{}) + go func() { + stream.Close() + close(closed) + }() + + // The unsubscribe must not wait on the manager + sid, subscribe := expectSubscriptionUpdate(t, transport) + require.False(t, subscribe) + require.Equal(t, trackSID, sid) + + // Subscribe again while the close is still in progress + result := subscribeAsync(context.Background(), track) + sid, subscribe = expectSubscriptionUpdate(t, transport) + require.True(t, subscribe) + require.Equal(t, trackSID, sid) + + m.mu.Unlock() + expectEvent(t, closed) + + assignHandle(m, subHandle+1, trackSID) + res := expectEvent(t, result) + require.NoError(t, res.err) +} + +// blockingDecryptor pauses the worker until release is closed, signaling entered first, so the +// test below can hold the previous subscription's worker mid-frame. +type blockingDecryptor struct{ entered, release chan struct{} } + +func (d blockingDecryptor) Decrypt(payload []byte, _ dtp.ExtensionE2EE) ([]byte, error) { + select { + case d.entered <- struct{}{}: + default: + } + <-d.release + return payload, nil +} + +// Should not deliver frames of an ended subscription to the streams of a later one, even when its +// worker is still draining. +func TestRemoteManager_ResubscribeDoesNotReceivePreviousSubscriptionFrames(t *testing.T) { + decryptor := blockingDecryptor{entered: make(chan struct{}, 1), release: make(chan struct{})} + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport, Decryptor: decryptor}) + + trackSID := livekit.TrackID("DTR_1234") + oldHandle, newHandle := trackHandle(0x1234), trackHandle(0x1235) + e2ee := Extensions{E2EE: &dtp.ExtensionE2EE{}} + + // Simulate track published (with e2ee) + track := publishTrack(t, m, transport, "id", Info{SID: trackSID, pubHandle: 1, Name: "test", UsesE2EE: true}) + + // Subscribe, and pause the worker on a frame + oldStream := subscribeAndActivate(t, m, transport, track, oldHandle) + m.HandlePacket(singlePacket(t, oldHandle, []byte{1}, e2ee)) + expectEvent(t, decryptor.entered) + + // Close the only subscriber, ending the subscription + oldStream.Close() + _, subscribe := expectSubscriptionUpdate(t, transport) + require.False(t, subscribe) + + // Subscribe again, then let the previous worker finish its frame + newStream := subscribeAndActivate(t, m, transport, track, newHandle) + close(decryptor.release) + expectNoFrame(t, newStream) + + // The new subscription still receives its own frames + m.HandlePacket(singlePacket(t, newHandle, []byte{2}, e2ee)) + frame := expectFrame(t, newStream) + require.Equal(t, []byte{2}, frame.Payload) +} + +// Should depacketize multiple interleaved partial frames when MaxPartialFrames is set before subscribe. +func TestRemoteManager_MaxPartialFramesSetBeforeSubscribe(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + subHandle := trackHandle(0x1234) + track := publishTrack(t, m, transport, "id", Info{SID: "DTR_1234", pubHandle: 1, Name: "test"}) + + // Configure the track BEFORE any subscribe + track.SetPipelineOptions(PipelineOptions{MaxPartialFrames: 3}) + + stream := subscribeAndActivate(t, m, transport, track, subHandle) + + // Two interleaved partial frames: Start(1), Start(2), Final(1), Final(2). With the default + // MaxPartialFrames of 1 frame 1 would be evicted by frame 2; with 3 both frames coexist and emerge. + pushInterleavedTwoFramePair(t, m, subHandle, 1, 0, [2][]byte{{0xa1}, {0xa2}}, 2, 100, [2][]byte{{0xb1}, {0xb2}}) + + require.Equal(t, []byte{0xa1, 0xa2}, expectFrame(t, stream).Payload) + require.Equal(t, []byte{0xb1, 0xb2}, expectFrame(t, stream).Payload) +} + +// Should pick up MaxPartialFrames live on an already-active subscription. +func TestRemoteManager_MaxPartialFramesSetLive(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + subHandle := trackHandle(0x1234) + track := publishTrack(t, m, transport, "id", Info{SID: "DTR_1234", pubHandle: 1, Name: "test"}) + + stream := subscribeAndActivate(t, m, transport, track, subHandle) + + // Subscription is now active; flip the cap on the live pipeline + track.SetPipelineOptions(PipelineOptions{MaxPartialFrames: 3}) + + pushInterleavedTwoFramePair(t, m, subHandle, 1, 0, [2][]byte{{0xa1}, {0xa2}}, 2, 100, [2][]byte{{0xb1}, {0xb2}}) + + require.Equal(t, []byte{0xa1, 0xa2}, expectFrame(t, stream).Payload) + require.Equal(t, []byte{0xb1, 0xb2}, expectFrame(t, stream).Payload) +} + +// Should drop the older partial frame by default (no MaxPartialFrames set). +func TestRemoteManager_DefaultDropsOlderPartialFrame(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + subHandle := trackHandle(0x1234) + track := publishTrack(t, m, transport, "id", Info{SID: "DTR_1234", pubHandle: 1, Name: "test"}) + + stream := subscribeAndActivate(t, m, transport, track, subHandle) + + // Default cap of 1: Start(2) evicts Start(1), so Final(1) is unknown and only frame 2 makes it through + pushInterleavedTwoFramePair(t, m, subHandle, 1, 0, [2][]byte{{0xa1}, {0xa2}}, 2, 100, [2][]byte{{0xb1}, {0xb2}}) + + require.Equal(t, []byte{0xb1, 0xb2}, expectFrame(t, stream).Payload) + expectNoFrame(t, stream) +} + +func TestStream_DropsOldestWhenFull(t *testing.T) { + stream := &Stream{buf: make([]Frame, 2), ready: make(chan struct{}, 1)} + for i := byte(1); i <= 3; i++ { + stream.push(Frame{Payload: []byte{i}}) + } + require.Equal(t, []byte{2}, expectFrame(t, stream).Payload) + require.Equal(t, []byte{3}, expectFrame(t, stream).Payload) + expectNoFrame(t, stream) + stream.close() + expectClosed(t, stream) +} diff --git a/remotedatatracktransport.go b/remotedatatracktransport.go new file mode 100644 index 00000000..6b94bea0 --- /dev/null +++ b/remotedatatracktransport.go @@ -0,0 +1,59 @@ +// 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 ( + "sync" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/server-sdk-go/v2/datatrack" +) + +// remoteDataTrackTransport connects the remote data track manager to the engine and the room's +// callbacks. +type remoteDataTrackTransport struct { + room *Room + // publishers maps each published track to its publisher, which a disconnect removes from the + // room before the track is unpublished + publishers sync.Map // *datatrack.RemoteTrack -> *RemoteParticipant +} + +func (t *remoteDataTrackTransport) SendUpdateSubscription(req *livekit.UpdateDataSubscription) error { + return t.room.engine.SendUpdateDataSubscription(req) +} + +// OnTrackPublished runs the callbacks on the signal goroutine, as for media tracks, so they must +// not block. In particular, Subscribe waits on a signal response and must not be called inline. +func (t *remoteDataTrackTransport) OnTrackPublished(track *datatrack.RemoteTrack) { + rp := t.room.GetParticipantByIdentity(track.PublisherIdentity()) + if rp != nil { + t.publishers.Store(track, rp) + } + if rp != nil { + rp.Callback.OnDataTrackPublished(track, rp) + } + t.room.callback.OnDataTrackPublished(track, rp) +} + +func (t *remoteDataTrackTransport) OnTrackUnpublished(track *datatrack.RemoteTrack) { + rp := t.room.GetParticipantByIdentity(track.PublisherIdentity()) + if publisher, ok := t.publishers.LoadAndDelete(track); ok && rp == nil { + rp = publisher.(*RemoteParticipant) + } + if rp != nil { + rp.Callback.OnDataTrackUnpublished(track, rp) + } + t.room.callback.OnDataTrackUnpublished(track, rp) +} diff --git a/remotedatatracktransport_test.go b/remotedatatracktransport_test.go new file mode 100644 index 00000000..50694c4f --- /dev/null +++ b/remotedatatracktransport_test.go @@ -0,0 +1,74 @@ +// 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 ( + "testing" + "time" + + "github.com/livekit/protocol/livekit" + "github.com/stretchr/testify/require" + + "github.com/livekit/server-sdk-go/v2/datatrack" +) + +// A participant's disconnect removes it from the room before its data tracks are unpublished; the +// unpublish callbacks must still receive it. +func TestDataTrackUnpublishedOnParticipantDisconnect(t *testing.T) { + roomUnpublished := make(chan *RemoteParticipant, 1) + room := NewRoom(&RoomCallback{ + ParticipantCallback: ParticipantCallback{ + OnDataTrackUnpublished: func(_ *datatrack.RemoteTrack, rp *RemoteParticipant) { + roomUnpublished <- rp + }, + }, + }) + t.Cleanup(room.engine.Close) + + info := &livekit.ParticipantInfo{ + Sid: "PA_remote", + Identity: "remote", + State: livekit.ParticipantInfo_ACTIVE, + DataTracks: []*livekit.DataTrackInfo{ + {Sid: "DTR_1234", PubHandle: 1, Name: "test"}, + }, + } + room.OnParticipantUpdate([]*livekit.ParticipantInfo{info}) + rp := room.GetParticipantByIdentity("remote") + require.NotNil(t, rp) + + participantUnpublished := make(chan struct{}, 1) + rp.Callback.OnDataTrackUnpublished = func(*datatrack.RemoteTrack, *RemoteParticipant) { + participantUnpublished <- struct{}{} + } + + room.OnParticipantUpdate([]*livekit.ParticipantInfo{{ + Sid: info.Sid, + Identity: info.Identity, + State: livekit.ParticipantInfo_DISCONNECTED, + }}) + + select { + case <-participantUnpublished: + case <-time.After(time.Second): + t.Fatal("participant callback not called") + } + select { + case got := <-roomUnpublished: + require.Same(t, rp, got) + case <-time.After(time.Second): + t.Fatal("room callback not called") + } +} diff --git a/room.go b/room.go index acbef166..7a3633db 100644 --- a/room.go +++ b/room.go @@ -35,6 +35,7 @@ import ( protosignalling "github.com/livekit/protocol/signalling" "github.com/livekit/server-sdk-go/v2/e2ee" + "github.com/livekit/server-sdk-go/v2/internal/datatrack" "github.com/livekit/server-sdk-go/v2/signalling" "github.com/livekit/mediatransportutil/pkg/pacer" @@ -322,6 +323,7 @@ type Room struct { sid string name string LocalParticipant *LocalParticipant + remoteDataTracks *datatrack.RemoteManager callback *RoomCallback sidReady chan struct{} disconnectReason livekit.DisconnectReason @@ -366,6 +368,7 @@ func NewRoom(callback *RoomCallback) *Room { r.engine = NewRTCEngine(r.useSinglePeerConnection, r, r.getLocalParticipantSID, r.regionURLProvider) r.LocalParticipant = newLocalParticipant(r.engine, r.callback, r.serverInfo, r.log) + r.remoteDataTracks = datatrack.NewRemoteManager(datatrack.RemoteManagerParams{Transport: &remoteDataTrackTransport{room: r}, Logger: r.log}) return r } @@ -394,6 +397,7 @@ func (r *Room) SetLogger(l protoLogger.Logger) { r.log = l r.engine.SetLogger(l) r.LocalParticipant.SetLogger(l) + r.remoteDataTracks.SetLogger(l) r.lock.RLock() for _, rp := range r.remoteParticipants { rp.SetLogger(l) @@ -789,6 +793,7 @@ func (r *Room) cleanup() { r.engine.Close() r.LocalParticipant.closeTracks() r.LocalParticipant.dataTracks.Shutdown() + r.remoteDataTracks.Shutdown() r.setSid("", true) r.byteStreamHandlers.Clear() @@ -1063,6 +1068,7 @@ func (r *Room) OnRoomJoined( r.clearParticipantDefers(livekit.ParticipantID(pi.Sid), pi) // no need to run participant defers here, since we are connected for the first time } + r.remoteDataTracks.HandleParticipantUpdate(otherParticipants, r.LocalParticipant.Identity()) } func (r *Room) OnDisconnected(reason livekit.DisconnectReason) { @@ -1109,6 +1115,8 @@ func (r *Room) OnRestarted( r.LocalParticipant.republishTracks() r.LocalParticipant.dataTracks.RepublishTracks() + r.remoteDataTracks.HandleParticipantSnapshot(otherParticipants, r.LocalParticipant.Identity()) + r.remoteDataTracks.ResendSubscriptionUpdates() } func (r *Room) OnRestartConnected() { @@ -1187,6 +1195,7 @@ func (r *Room) OnParticipantUpdate(participants []*livekit.ParticipantInfo) { r.runParticipantDefers(newSid, rp) } } + r.remoteDataTracks.HandleParticipantUpdate(participants, r.LocalParticipant.Identity()) } func (r *Room) OnParticipantDisconnect(rp *RemoteParticipant, reason livekit.DisconnectReason) { @@ -1302,6 +1311,7 @@ func (r *Room) OnRoomMoved(moved *livekit.RoomMovedResponse) { infos = append(infos, moved.Participant) infos = append(infos, moved.OtherParticipants...) r.OnParticipantUpdate(infos) + r.remoteDataTracks.HandleParticipantSnapshot(moved.OtherParticipants, r.LocalParticipant.Identity()) } func (r *Room) OnTrackRemoteMuted(msg *livekit.MuteTrackRequest) { @@ -1426,9 +1436,11 @@ func (r *Room) OnUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit. } func (r *Room) OnDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) { + r.remoteDataTracks.HandleSubscriberHandles(dataTrackSubscriberHandles) } func (r *Room) OnDataTrackPacket(data []byte) { + r.remoteDataTracks.HandlePacket(data) } func (r *Room) OnStreamHeader(streamHeader *livekit.DataStream_Header, participantIdentity string) {