diff --git a/datatrack/datatrack.go b/datatrack/datatrack.go new file mode 100644 index 00000000..bf38b992 --- /dev/null +++ b/datatrack/datatrack.go @@ -0,0 +1,108 @@ +// 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 is the public API for data tracks. Publish a track with +// [github.com/livekit/server-sdk-go/v2.LocalParticipant.PublishDataTrack]. +package datatrack + +import ( + backend "github.com/livekit/server-sdk-go/v2/internal/datatrack" +) + +type ( + // LocalTrack is a data track published by the local participant. + LocalTrack = backend.LocalTrack + // Info describes a published data track. + Info = backend.Info + // Frame is a frame published on a data track. + Frame = backend.Frame + + // PublishOptions describe a data track to publish. + PublishOptions = backend.PublishOptions + // PublishOption customizes a publication. + PublishOption = backend.PublishOption + + // SchemaID identifies a data track schema. + SchemaID = backend.SchemaID + // SchemaEncoding is the encoding of a schema definition: a WellKnownSchemaEncoding or a + // CustomSchemaEncoding. + SchemaEncoding = backend.SchemaEncoding + // WellKnownSchemaEncoding is a schema encoding defined by the protocol. + WellKnownSchemaEncoding = backend.WellKnownSchemaEncoding + // CustomSchemaEncoding is an application-defined schema encoding identified by name. + CustomSchemaEncoding = backend.CustomSchemaEncoding + + // FrameEncoding is the encoding of frames pushed on a data track: a WellKnownFrameEncoding or + // a CustomFrameEncoding. + FrameEncoding = backend.FrameEncoding + // WellKnownFrameEncoding is a frame encoding defined by the protocol. + WellKnownFrameEncoding = backend.WellKnownFrameEncoding + // CustomFrameEncoding is an application-defined frame encoding identified by name. + CustomFrameEncoding = backend.CustomFrameEncoding +) + +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 + + ErrSchemaMissingFrameEncoding = backend.ErrSchemaMissingFrameEncoding + ErrSchemaMissingID = backend.ErrSchemaMissingID + ErrSchemaIncompatible = backend.ErrSchemaIncompatible +) + +const ( + // SchemaEncodingOther is a well-known encoding not known to this SDK version. + SchemaEncodingOther = backend.SchemaEncodingOther + SchemaEncodingProtobuf = backend.SchemaEncodingProtobuf + SchemaEncodingFlatbuffer = backend.SchemaEncodingFlatbuffer + SchemaEncodingROS1Msg = backend.SchemaEncodingROS1Msg + SchemaEncodingROS2Msg = backend.SchemaEncodingROS2Msg + SchemaEncodingROS2IDL = backend.SchemaEncodingROS2IDL + SchemaEncodingOMGIDL = backend.SchemaEncodingOMGIDL + SchemaEncodingJSONSchema = backend.SchemaEncodingJSONSchema +) + +const ( + // FrameEncodingOther is a well-known encoding not known to this SDK version. + FrameEncodingOther = backend.FrameEncodingOther + FrameEncodingROS1 = backend.FrameEncodingROS1 + FrameEncodingCDR = backend.FrameEncodingCDR + FrameEncodingProtobuf = backend.FrameEncodingProtobuf + FrameEncodingFlatbuffer = backend.FrameEncodingFlatbuffer + FrameEncodingCBOR = backend.FrameEncodingCBOR + FrameEncodingMsgPack = backend.FrameEncodingMsgPack + FrameEncodingJSON = backend.FrameEncodingJSON +) + +// WithSchema associates a schema with the track's frames. +func WithSchema(schema SchemaID) PublishOption { + return backend.WithSchema(schema) +} + +// WithFrameEncoding declares the encoding of the track's frames. +func WithFrameEncoding(encoding FrameEncoding) PublishOption { + return backend.WithFrameEncoding(encoding) +} + +// UserTimestampNow returns the current Unix time in milliseconds, the interpretation +// Frame.DurationSinceTimestamp expects. +func UserTimestampNow() *uint64 { + return backend.UserTimestampNow() +} diff --git a/datatracksender.go b/datatracksender.go index 495a0153..7b1f464a 100644 --- a/datatracksender.go +++ b/datatracksender.go @@ -19,13 +19,12 @@ import ( protoLogger "github.com/livekit/protocol/logger" "github.com/pion/webrtc/v4" + + "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) const dataTrackBufferedAmountLowThreshold = 8 * 1024 -// dataTrackFramePackets is one frame serialized into data track packets. -type dataTrackFramePackets [][]byte - // dataTrackSender paces frames onto the data track channel, keeping only the freshest frame // while the channel's send buffer is above the low threshold. type dataTrackSender struct { @@ -33,7 +32,7 @@ type dataTrackSender struct { lock sync.Mutex dc *webrtc.DataChannel - frame dataTrackFramePackets + frame datatrack.FramePackets notify chan struct{} done chan struct{} } @@ -61,7 +60,7 @@ func (s *dataTrackSender) setLogger(log protoLogger.Logger) { s.log = log } -func (s *dataTrackSender) send(frame dataTrackFramePackets) { +func (s *dataTrackSender) send(frame datatrack.FramePackets) { if dropped := s.push(frame); dropped != nil { s.log.Debugw("dropping data track frame", "numPackets", len(dropped)) } @@ -78,7 +77,7 @@ func (s *dataTrackSender) stop() { close(s.done) } -func (s *dataTrackSender) push(frame dataTrackFramePackets) (dropped dataTrackFramePackets) { +func (s *dataTrackSender) push(frame datatrack.FramePackets) (dropped datatrack.FramePackets) { if len(frame) == 0 { return nil } @@ -91,7 +90,7 @@ func (s *dataTrackSender) push(frame dataTrackFramePackets) (dropped dataTrackFr return dropped } -func (s *dataTrackSender) pop() dataTrackFramePackets { +func (s *dataTrackSender) pop() datatrack.FramePackets { s.lock.Lock() defer s.lock.Unlock() @@ -103,7 +102,7 @@ func (s *dataTrackSender) pop() dataTrackFramePackets { func (s *dataTrackSender) run() { var ( dc *webrtc.DataChannel - inFlight dataTrackFramePackets + inFlight datatrack.FramePackets ) for { select { diff --git a/datatracksender_test.go b/datatracksender_test.go index a76ae0e3..f2767b2b 100644 --- a/datatracksender_test.go +++ b/datatracksender_test.go @@ -18,10 +18,12 @@ import ( "testing" "github.com/stretchr/testify/require" + + "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) -func testFrame(marker byte, packets int) dataTrackFramePackets { - frame := make(dataTrackFramePackets, packets) +func testFrame(marker byte, packets int) datatrack.FramePackets { + frame := make(datatrack.FramePackets, packets) for i := range frame { frame[i] = []byte{marker, byte(i)} } diff --git a/engine.go b/engine.go index d727c9e4..4ce6c9c4 100644 --- a/engine.go +++ b/engine.go @@ -34,6 +34,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" ) @@ -746,22 +747,24 @@ func (e *RTCEngine) GetDataChannelSub(kind livekit.DataPacket_Kind) *webrtc.Data } func (e *RTCEngine) waitUntilConnected(timeout time.Duration) error { - return waitUntilConnected(timeout, func() bool { + return waitUntilConnected(context.Background(), timeout, func() bool { return e.IsConnected() }) } -func (e *RTCEngine) ensurePublisherConnected(ensureDataReady bool) error { +// ensurePublisherConnected waits for the publisher connection, giving up when ctx ends or the +// connect timeout passes, whichever comes first. +func (e *RTCEngine) ensurePublisherConnected(ctx context.Context, ensureDataReady bool) error { e.pclock.Lock() subscriberPrimary := e.subscriberPrimary e.pclock.Unlock() connectTimeout := e.connectionManager.getConnectTimeout() if !subscriberPrimary { - return e.waitUntilConnected(connectTimeout) + return waitUntilConnected(ctx, connectTimeout, e.IsConnected) } var negotiated bool - return waitUntilConnected(connectTimeout, func() bool { + return waitUntilConnected(ctx, connectTimeout, func() bool { if publisher, ok := e.Publisher(); ok { if publisher.IsConnected() && (!ensureDataReady || e.dataPubChannelReady()) { return true @@ -1231,7 +1234,7 @@ func filterTURNServers(iceServers []*livekit.ICEServer) []*livekit.ICEServer { } func (e *RTCEngine) publishDataPacket(pck *livekit.DataPacket, kind livekit.DataPacket_Kind) error { - err := e.ensurePublisherConnected(true) + err := e.ensurePublisherConnected(context.Background(), true) if err != nil { e.log.Errorw("could not ensure publisher connected", err) return err @@ -1836,7 +1839,7 @@ func setConfiguration(pcTransport *PCTransport, configuration webrtc.Configurati } } -func waitUntilConnected(d time.Duration, test func() bool) error { +func waitUntilConnected(ctx context.Context, d time.Duration, test func() bool) error { if test() { return nil } @@ -1849,6 +1852,9 @@ func waitUntilConnected(d time.Duration, test func() bool) error { for { select { + case <-ctx.Done(): + return ctx.Err() + case <-timeout.C: return ErrConnectionTimeout @@ -1860,6 +1866,6 @@ func waitUntilConnected(d time.Duration, test func() bool) error { } } -func (e *RTCEngine) sendDataTrackFrame(frame dataTrackFramePackets) { +func (e *RTCEngine) sendDataTrackFrame(frame datatrack.FramePackets) { e.dataTrackSender.send(frame) } diff --git a/examples/datatrack/publisher/main.go b/examples/datatrack/publisher/main.go new file mode 100644 index 00000000..2586e0b7 --- /dev/null +++ b/examples/datatrack/publisher/main.go @@ -0,0 +1,88 @@ +package main + +import ( + "bytes" + "context" + "fmt" + "os" + "os/signal" + "syscall" + "time" + + "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-publisher") + lksdk.SetLogger(logger.GetLogger()) + + cmd := &cli.Command{ + Name: "datatrack-publisher", + Usage: "publish sensor data on a data track", + 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: "publisher"}, + }, + 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 { + 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"), + }, nil) + if err != nil { + return err + } + defer room.Disconnect() + + track, err := room.LocalParticipant.PublishDataTrack(ctx, "my_sensor_data") + if err != nil { + return err + } + defer track.Unpublish() + + pushFrames(ctx, track) + return nil +} + +func readSensor() []byte { + // Dynamically read some sensor data... + return bytes.Repeat([]byte{0xfa}, 256) +} + +func pushFrames(ctx context.Context, track *datatrack.LocalTrack) { + for { + logger.Infow("pushing frame") + + frame := datatrack.Frame{ + Payload: readSensor(), + UserTimestamp: datatrack.UserTimestampNow(), + } + if err := track.TryPush(frame); err != nil { + logger.Warnw("failed to push frame", err) + } + + select { + case <-ctx.Done(): + return + case <-time.After(500 * time.Millisecond): + } + } +} diff --git a/go.mod b/go.mod index ec8bc220..45a79eb8 100644 --- a/go.mod +++ b/go.mod @@ -22,6 +22,7 @@ require ( github.com/pion/webrtc/v4 v4.2.24 github.com/stretchr/testify v1.12.1 github.com/twitchtv/twirp v8.1.3+incompatible + github.com/urfave/cli/v3 v3.9.0 go.uber.org/atomic v1.12.0 golang.org/x/crypto v0.57.0 google.golang.org/protobuf v1.36.12 diff --git a/go.sum b/go.sum index 3e3fe4b1..0f599f59 100644 --- a/go.sum +++ b/go.sum @@ -282,6 +282,8 @@ github.com/tonistiigi/vt100 v0.0.0-20240514184818-90bafcd6abab h1:H6aJ0yKQ0gF49Q github.com/tonistiigi/vt100 v0.0.0-20240514184818-90bafcd6abab/go.mod h1:ulncasL3N9uLrVann0m+CDlJKWsIAP34MPcOJF6VRvc= github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJXP61mNV3/7iuU= github.com/twitchtv/twirp v8.1.3+incompatible/go.mod h1:RRJoFSAmTEh2weEqWtpPE3vFK5YBhA6bqp2l1kfCC5A= +github.com/urfave/cli/v3 v3.9.0 h1:AV9lIiPv3ukYnxunaCUsHnEozptYmDN2F0+yWqLMn/c= +github.com/urfave/cli/v3 v3.9.0/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso= github.com/vbatts/tar-split v0.12.3 h1:Cd46rkGXI3Td4yrVNwU8ripbxFaQbmesqhjBUUYAJSw= github.com/vbatts/tar-split v0.12.3/go.mod h1:sQOc6OlqGCr7HkGx/IDBeKiTIvqhmj8KffNhEXG4Nq0= github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= diff --git a/integration_test.go b/integration_test.go index 39ad8ad5..616aba6c 100644 --- a/integration_test.go +++ b/integration_test.go @@ -833,7 +833,7 @@ func TestForceTLS(t *testing.T) { pub.LocalParticipant.PublishDataPacket(UserData([]byte("test")), WithDataPublishReliable(true)) pub.Simulate(SimulateForceTLS) - require.Eventually(t, func() bool { return reconnected.Load() && pub.engine.ensurePublisherConnected(true) == nil }, 15*time.Second, 100*time.Millisecond) + require.Eventually(t, func() bool { return reconnected.Load() && pub.engine.ensurePublisherConnected(context.Background(), true) == nil }, 15*time.Second, 100*time.Millisecond) pub.log.Infow("reconnected") diff --git a/datatrack/clock.go b/internal/datatrack/clock.go similarity index 100% rename from datatrack/clock.go rename to internal/datatrack/clock.go diff --git a/datatrack/depacketizer.go b/internal/datatrack/depacketizer.go similarity index 100% rename from datatrack/depacketizer.go rename to internal/datatrack/depacketizer.go diff --git a/datatrack/depacketizer_test.go b/internal/datatrack/depacketizer_test.go similarity index 100% rename from datatrack/depacketizer_test.go rename to internal/datatrack/depacketizer_test.go diff --git a/internal/datatrack/errors.go b/internal/datatrack/errors.go new file mode 100644 index 00000000..cf465473 --- /dev/null +++ b/internal/datatrack/errors.go @@ -0,0 +1,28 @@ +// 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 "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") +) diff --git a/datatrack/frame.go b/internal/datatrack/frame.go similarity index 100% rename from datatrack/frame.go rename to internal/datatrack/frame.go diff --git a/datatrack/handle.go b/internal/datatrack/handle.go similarity index 100% rename from datatrack/handle.go rename to internal/datatrack/handle.go diff --git a/datatrack/info.go b/internal/datatrack/info.go similarity index 100% rename from datatrack/info.go rename to internal/datatrack/info.go diff --git a/internal/datatrack/local.go b/internal/datatrack/local.go new file mode 100644 index 00000000..3437c151 --- /dev/null +++ b/internal/datatrack/local.go @@ -0,0 +1,484 @@ +// 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" + "sync" + "sync/atomic" + + dtp "github.com/livekit/protocol/datatrack" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +// transportMTU is the largest packet sent over the data channel. +const transportMTU = 16_000 + +// FramePackets is one frame serialized into data track packets. +type FramePackets [][]byte + +// LocalTransport carries what the local manager produces: signal requests to the SFU and frame +// packets over the data channel. +type LocalTransport interface { + // SendPublishRequest may block until the connection is ready; it returns ctx.Err() when ctx ends first. + SendPublishRequest(ctx context.Context, req *livekit.PublishDataTrackRequest) error + SendUnpublishRequest(req *livekit.UnpublishDataTrackRequest) error + // SendFrame queues a frame's packets; it must not block. + SendFrame(frame FramePackets) +} + +// Encryptor seals frame payloads for end-to-end encryption. +type Encryptor interface { + Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) +} + +// PublishOptions describe a data track to publish. +type PublishOptions struct { + Name string + Schema *SchemaID + FrameEncoding FrameEncoding +} + +// PublishOption customizes a publication. +type PublishOption func(*PublishOptions) + +// WithSchema associates a schema with the track's frames. +func WithSchema(schema SchemaID) PublishOption { + return func(options *PublishOptions) { + options.Schema = &schema + } +} + +// WithFrameEncoding declares the encoding of the track's frames. +func WithFrameEncoding(encoding FrameEncoding) PublishOption { + return func(options *PublishOptions) { + options.FrameEncoding = encoding + } +} + +type LocalManagerParams struct { + Transport LocalTransport + // Encryptor enables end-to-end encryption for every published track; nil disables it. + Encryptor Encryptor + Logger logger.Logger +} + +// LocalManager tracks the data tracks published by the local participant. Methods are safe for +// concurrent use. +type LocalManager struct { + params LocalManagerParams + log atomic.Pointer[logger.Logger] + + mu sync.Mutex + handles handleAllocator + pending map[trackHandle]chan publishResult + active map[trackHandle]*LocalTrack +} + +type publishResult struct { + track *LocalTrack + err error +} + +func NewLocalManager(params LocalManagerParams) *LocalManager { + if params.Logger == nil { + params.Logger = logger.GetLogger() + } + m := &LocalManager{ + params: params, + pending: make(map[trackHandle]chan publishResult), + active: make(map[trackHandle]*LocalTrack), + } + m.SetLogger(params.Logger) + return m +} + +// SetLogger replaces the logger given at construction. +func (m *LocalManager) SetLogger(l logger.Logger) { + m.log.Store(&l) +} + +func (m *LocalManager) logger() logger.Logger { + return *m.log.Load() +} + +// Publish requests a publication and waits for the SFU's answer. Ending ctx abandons the request; +// should the SFU accept it afterwards, the track is unpublished right away. +func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*LocalTrack, error) { + var schemaEncoding SchemaEncoding + if options.Schema != nil { + schemaEncoding = options.Schema.Encoding + if schemaEncoding == nil { + schemaEncoding = SchemaEncodingOther + } + } + if err := validateSchema(options.FrameEncoding, schemaEncoding); err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + + m.mu.Lock() + handle, err := m.handles.get() + if err != nil { + m.mu.Unlock() + return nil, ErrLimitReached + } + result := make(chan publishResult, 1) + m.pending[handle] = result + m.mu.Unlock() + + request := publishRequest{ + handle: handle, + name: options.Name, + usesE2EE: m.params.Encryptor != nil, + schema: options.Schema, + frameEncoding: options.FrameEncoding, + } + if err := m.params.Transport.SendPublishRequest(ctx, request.toProto()); err != nil { + m.mu.Lock() + delete(m.pending, handle) + m.mu.Unlock() + return nil, err + } + + select { + case res := <-result: + return res.track, res.err + case <-ctx.Done(): + m.mu.Lock() + _, stillPending := m.pending[handle] + delete(m.pending, handle) + m.mu.Unlock() + if !stillPending { + // the answer came first and was delivered under the lock + if res := <-result; res.track != nil { + res.track.Unpublish() + } + } + return nil, ctx.Err() + } +} + +// HandlePublishResponse completes the publication the SFU accepted. +func (m *LocalManager) HandlePublishResponse(msg *livekit.PublishDataTrackResponse) { + info, err := infoFromProto(msg.GetInfo()) + if err != nil { + m.logger().Warnw("ignoring invalid publish data track response", err) + return + } + m.resolvePublish(publishResponse{handle: info.pubHandle, info: info}) +} + +// HandleRequestResponse fails the publication a rejection refers to. It reports whether the +// response was a publish rejection. +func (m *LocalManager) HandleRequestResponse(msg *livekit.RequestResponse) bool { + rejection, ok := publishRejectionFromRequestResponse(msg) + if !ok { + return false + } + m.resolvePublish(rejection) + return true +} + +func (m *LocalManager) resolvePublish(response publishResponse) { + m.mu.Lock() + if result, ok := m.pending[response.handle]; ok { + delete(m.pending, response.handle) + res := publishResult{err: response.err} + if response.err == nil { + res.track = newLocalTrack(m, response.info) + m.active[response.handle] = res.track + } + result <- res + m.mu.Unlock() + return + } + track, isActive := m.active[response.handle] + if isActive && response.err != nil { + delete(m.active, response.handle) + } + m.mu.Unlock() + + switch { + case isActive && response.err != nil: + track.markUnpublished() + m.logger().Warnw("republish failed for data track", response.err, "handle", response.handle) + case isActive: + if track.completeRepublish(response.info.SID) { + m.logger().Debugw("data track republished", "handle", response.handle, "sid", response.info.SID) + } else { + m.logger().Warnw("data track already published", nil, "handle", response.handle) + } + case response.err == nil: + // accepted after the request was abandoned, unpublish to keep the SFU consistent + m.sendUnpublish(response.handle) + } +} + +// HandleUnpublishResponse completes an unpublication, whether requested locally or by the SFU. +func (m *LocalManager) HandleUnpublishResponse(msg *livekit.UnpublishDataTrackResponse) { + handle, err := handleFromUint32(msg.GetInfo().GetPubHandle()) + if err != nil { + m.logger().Warnw("ignoring invalid unpublish data track response", err) + return + } + + m.mu.Lock() + if result, ok := m.pending[handle]; ok { + delete(m.pending, handle) + result <- publishResult{err: ErrUnpublished} + } + track, ok := m.active[handle] + delete(m.active, handle) + m.mu.Unlock() + + if ok { + track.markUnpublished() + } +} + +func (m *LocalManager) unpublish(handle trackHandle) { + m.mu.Lock() + track, ok := m.active[handle] + delete(m.active, handle) + m.mu.Unlock() + if !ok { + return + } + track.markUnpublished() + m.sendUnpublish(handle) +} + +func (m *LocalManager) sendUnpublish(handle trackHandle) { + if err := m.params.Transport.SendUnpublishRequest(unpublishRequestToProto(handle)); err != nil { + m.logger().Debugw("could not send unpublish data track request", "handle", handle, "error", err) + } +} + +// RepublishTracks re-requests every publication after a full reconnect. Pending publications fail +// with ErrDisconnected; published tracks drop frames until the SFU answers with their new SID, and +// are unpublished if it rejects them. +func (m *LocalManager) RepublishTracks() { + m.mu.Lock() + for handle, result := range m.pending { + delete(m.pending, handle) + result <- publishResult{err: ErrDisconnected} + } + requests := make([]publishRequest, 0, len(m.active)) + for handle, track := range m.active { + info := track.beginRepublish() + requests = append(requests, publishRequest{ + handle: handle, + name: info.Name, + usesE2EE: info.UsesE2EE, + schema: info.Schema, + frameEncoding: info.FrameEncoding, + }) + } + m.mu.Unlock() + + for _, request := range requests { + if err := m.params.Transport.SendPublishRequest(context.Background(), request.toProto()); err != nil { + m.logger().Warnw("could not republish data track", err, "handle", request.handle) + } + } +} + +// PublishResponsesForSyncState describes the current publications for SyncState. +func (m *LocalManager) PublishResponsesForSyncState() []*livekit.PublishDataTrackResponse { + m.mu.Lock() + infos := make([]Info, 0, len(m.active)) + for _, track := range m.active { + infos = append(infos, track.Info()) + } + m.mu.Unlock() + return publishResponsesForSyncState(infos) +} + +// Shutdown fails pending publications and unpublishes every track locally. The manager can be +// used again for a new session. +func (m *LocalManager) Shutdown() { + m.mu.Lock() + for handle, result := range m.pending { + delete(m.pending, handle) + result <- publishResult{err: ErrDisconnected} + } + tracks := make([]*LocalTrack, 0, len(m.active)) + for _, track := range m.active { + tracks = append(tracks, track) + } + clear(m.active) + m.mu.Unlock() + + for _, track := range tracks { + track.markUnpublished() + } +} + +type publishState int + +const ( + statePublished publishState = iota + stateRepublishing + stateUnpublished +) + +// LocalTrack is a data track published by the local participant. Methods are safe for concurrent +// use. +type LocalTrack struct { + manager *LocalManager + handle trackHandle + unpublished chan struct{} + + mu sync.Mutex + state publishState + info Info + pipeline localPipeline +} + +func newLocalTrack(manager *LocalManager, info Info) *LocalTrack { + var encryptor Encryptor + if info.UsesE2EE { + encryptor = manager.params.Encryptor + } + return &LocalTrack{ + manager: manager, + handle: info.pubHandle, + unpublished: make(chan struct{}), + info: info, + pipeline: newLocalPipeline(info.pubHandle, encryptor), + } +} + +// Info returns a snapshot of the track's metadata. The SID changes when the local participant +// completes a full reconnect. +func (t *LocalTrack) Info() Info { + t.mu.Lock() + defer t.mu.Unlock() + return t.info +} + +// IsPublished reports whether the track is published. A track being republished after a +// reconnect still counts as published. +func (t *LocalTrack) IsPublished() bool { + t.mu.Lock() + defer t.mu.Unlock() + return t.state != stateUnpublished +} + +// Unpublished is closed once the track is no longer published, whether by Unpublish, the SFU, or +// disconnecting from the room. +func (t *LocalTrack) Unpublished() <-chan struct{} { + return t.unpublished +} + +// TryPush sends a frame to the track's subscribers. It returns ErrUnpublished once the track is +// unpublished and ErrQueueFull while it is being republished after a reconnect. Frames are never +// retained, so the caller may reuse the payload immediately. Frames with empty payloads are dropped. +func (t *LocalTrack) TryPush(frame Frame) error { + t.mu.Lock() + switch t.state { + case stateRepublishing: + t.mu.Unlock() + return ErrQueueFull + case stateUnpublished: + t.mu.Unlock() + return ErrUnpublished + } + packets, err := t.pipeline.processFrame(frame) + t.mu.Unlock() + if err != nil { + return err + } + t.manager.params.Transport.SendFrame(packets) + return nil +} + +// Unpublish stops the publication. It is safe to call more than once. +func (t *LocalTrack) Unpublish() { + t.manager.unpublish(t.handle) +} + +func (t *LocalTrack) markUnpublished() { + t.mu.Lock() + defer t.mu.Unlock() + if t.state == stateUnpublished { + return + } + t.state = stateUnpublished + close(t.unpublished) +} + +func (t *LocalTrack) beginRepublish() Info { + t.mu.Lock() + defer t.mu.Unlock() + t.state = stateRepublishing + return t.info +} + +// completeRepublish reports false if the track was not being republished. +func (t *LocalTrack) completeRepublish(sid livekit.TrackID) bool { + t.mu.Lock() + defer t.mu.Unlock() + if t.state != stateRepublishing { + return false + } + t.state = statePublished + t.info.SID = sid + return true +} + +// localPipeline turns frames into packets ready for the data channel. +type localPipeline struct { + encryptor Encryptor + packetizer *packetizer +} + +func newLocalPipeline(handle trackHandle, encryptor Encryptor) localPipeline { + return localPipeline{encryptor: encryptor, packetizer: newPacketizer(handle, transportMTU)} +} + +func (p *localPipeline) processFrame(frame Frame) (FramePackets, error) { + payload := frame.Payload + extensions := Extensions{UserTimestamp: frame.UserTimestamp} + if p.encryptor != nil { + ciphertext, e2ee, err := p.encryptor.Encrypt(payload) + if err != nil { + return nil, err + } + payload, extensions.E2EE = ciphertext, &e2ee + } + + packets, err := p.packetizer.packetize(payload, extensions) + if err != nil { + return nil, err + } + return marshalPackets(packets) +} + +func marshalPackets(packets []dtp.Packet) (FramePackets, error) { + raw := make(FramePackets, len(packets)) + for i := range packets { + buf, err := packets[i].Marshal() + if err != nil { + return nil, err + } + raw[i] = buf + } + return raw, nil +} diff --git a/internal/datatrack/local_test.go b/internal/datatrack/local_test.go new file mode 100644 index 00000000..e0b76941 --- /dev/null +++ b/internal/datatrack/local_test.go @@ -0,0 +1,401 @@ +// 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" + "math" + "testing" + "time" + + dtp "github.com/livekit/protocol/datatrack" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/utils/pointer" + "github.com/stretchr/testify/require" +) + +// fakeLocalTransport records what the manager sends, in the order it was sent. +type fakeLocalTransport struct { + publishRequests chan *livekit.PublishDataTrackRequest + unpublishRequests chan *livekit.UnpublishDataTrackRequest + frames chan FramePackets +} + +func newFakeLocalTransport() *fakeLocalTransport { + return &fakeLocalTransport{ + publishRequests: make(chan *livekit.PublishDataTrackRequest, 16), + unpublishRequests: make(chan *livekit.UnpublishDataTrackRequest, 16), + frames: make(chan FramePackets, 16), + } +} + +func (f *fakeLocalTransport) SendPublishRequest(_ context.Context, req *livekit.PublishDataTrackRequest) error { + f.publishRequests <- req + return nil +} + +func (f *fakeLocalTransport) SendUnpublishRequest(req *livekit.UnpublishDataTrackRequest) error { + f.unpublishRequests <- req + return nil +} + +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) + go func() { + track, err := m.Publish(ctx, PublishOptions{Name: name}) + result <- publishResult{track: track, err: err} + }() + return result +} + +// accepted is the SFU's answer to a publish request. +func accepted(request *livekit.PublishDataTrackRequest, sid livekit.TrackID) *livekit.PublishDataTrackResponse { + return &livekit.PublishDataTrackResponse{Info: &livekit.DataTrackInfo{ + PubHandle: request.GetPubHandle(), + Sid: string(sid), + Name: request.GetName(), + Encryption: request.GetEncryption(), + }} +} + +func TestLocalPipeline_ProcessFrame(t *testing.T) { + pipeline := newLocalPipeline(0x8811, nil) + + const repeatedByte = 0xab + frame := Frame{ + Payload: bytes.Repeat([]byte{repeatedByte}, 32_000), + UserTimestamp: pointer.To(uint64(0x4411221111118811)), + } + + packets, err := pipeline.processFrame(frame) + require.NoError(t, err) + require.Len(t, packets, 3) + + for _, raw := range packets { + packet, err := parsePacket(raw) + require.NoError(t, err) + require.Nil(t, extensionsOf(&packet.Header).E2EE) + require.NotEmpty(t, packet.Payload) + require.Equal(t, bytes.Repeat([]byte{repeatedByte}, len(packet.Payload)), packet.Payload) + } +} + +func TestLocalManager_Shutdown(t *testing.T) { + m := NewLocalManager(LocalManagerParams{Transport: newFakeLocalTransport()}) + m.Shutdown() + require.Empty(t, m.PublishResponsesForSyncState()) +} + +func TestLocalManager_Publish(t *testing.T) { + const payloadSize, packetCount = 256, 10 + trackName, trackSID := "track", livekit.TrackID("DTR_1234") + + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + result := publishAsync(context.Background(), m, trackName) + + request := expectEvent(t, transport.publishRequests) + require.Equal(t, livekit.Encryption_NONE, request.GetEncryption()) + require.Equal(t, trackName, request.GetName()) + + // SFU accepts publication + m.HandlePublishResponse(accepted(request, trackSID)) + + res := expectEvent(t, result) + require.NoError(t, res.err) + track := res.track + require.False(t, track.Info().UsesE2EE) + require.Equal(t, trackName, track.Info().Name) + require.Equal(t, trackSID, track.Info().SID) + + for range packetCount { + require.NoError(t, track.TryPush(Frame{Payload: bytes.Repeat([]byte{0xfa}, payloadSize)})) + time.Sleep(10 * time.Millisecond) + } + for range packetCount { + packets := expectEvent(t, transport.frames) + packet, err := parsePacket(packets[0]) + require.NoError(t, err) + require.Len(t, packet.Payload, payloadSize) + } + + track.Unpublish() + unpublish := expectEvent(t, transport.unpublishRequests) + require.Equal(t, request.GetPubHandle(), unpublish.GetPubHandle()) +} + +func TestLocalManager_PublishSfuError(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + result := publishAsync(context.Background(), m, "test") + + // SFU rejects publication + request := expectEvent(t, transport.publishRequests) + handled := m.HandleRequestResponse(&livekit.RequestResponse{ + Request: &livekit.RequestResponse_PublishDataTrack{PublishDataTrack: request}, + Reason: livekit.RequestResponse_LIMIT_EXCEEDED, + }) + require.True(t, handled) + + res := expectEvent(t, result) + require.ErrorIs(t, res.err, ErrLimitReached) +} + +func TestLocalManager_PublishHandlesExhausted(t *testing.T) { + m := NewLocalManager(LocalManagerParams{Transport: newFakeLocalTransport()}) + m.handles.value = math.MaxUint16 + + _, err := m.Publish(context.Background(), PublishOptions{Name: "test"}) + require.ErrorIs(t, err, ErrLimitReached) +} + +func TestLocalManager_PublishCancelled(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + ctx, cancel := context.WithCancel(context.Background()) + result := publishAsync(ctx, m, "test") + + request := expectEvent(t, transport.publishRequests) + + // Caller gives up before SFU responds + cancel() + res := expectEvent(t, result) + require.ErrorIs(t, res.err, context.Canceled) + time.Sleep(50 * time.Millisecond) + + // Late SFU response arrives after cancellation + m.HandlePublishResponse(accepted(request, "DTR_1234")) + + // Manager sends unpublish for the orphaned handle + unpublish := expectEvent(t, transport.unpublishRequests) + require.Equal(t, request.GetPubHandle(), unpublish.GetPubHandle()) +} + +func TestLocalManager_PublishAlreadyCancelled(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := m.Publish(ctx, PublishOptions{Name: "test"}) + require.ErrorIs(t, err, context.Canceled) + require.Empty(t, transport.publishRequests) +} + +// blockingTransport never becomes ready: SendPublishRequest waits on ctx like the real transport +// waits on the engine connection. +type blockingTransport struct{ *fakeLocalTransport } + +func (b blockingTransport) SendPublishRequest(ctx context.Context, _ *livekit.PublishDataTrackRequest) error { + <-ctx.Done() + return ctx.Err() +} + +func TestLocalManager_PublishCancelledWhileTransportBlocked(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: blockingTransport{transport}}) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + res := expectEvent(t, publishAsync(ctx, m, "test")) + require.ErrorIs(t, res.err, context.DeadlineExceeded) + + // The abandoned request released its handle and left nothing pending + m.mu.Lock() + require.Empty(t, m.pending) + require.Empty(t, m.active) + m.mu.Unlock() +} + +func TestLocalManager_PublishWithE2EE(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport, Encryptor: prefixingEncryptor{}}) + + result := publishAsync(context.Background(), m, "secure") + + // SFU publish request should indicate e2ee + request := expectEvent(t, transport.publishRequests) + require.Equal(t, livekit.Encryption_GCM, request.GetEncryption()) + + // SFU accepts publication with e2ee + m.HandlePublishResponse(accepted(request, "DTR_1234")) + + res := expectEvent(t, result) + require.NoError(t, res.err) + track := res.track + require.True(t, track.Info().UsesE2EE) + + // Push a frame and verify encryption was applied + require.NoError(t, track.TryPush(Frame{Payload: []byte{1, 2, 3, 4, 5}})) + + 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.NotNil(t, extensionsOf(&packet.Header).E2EE) +} + +func TestLocalManager_RepublishTracks(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + // Publish a track through the full flow + trackName, trackSID := "track", livekit.TrackID("DTR_1234") + + result := publishAsync(context.Background(), m, trackName) + request := expectEvent(t, transport.publishRequests) + m.HandlePublishResponse(accepted(request, trackSID)) + + res := expectEvent(t, result) + require.NoError(t, res.err) + track := res.track + require.Equal(t, trackSID, track.Info().SID) + + // Simulate reconnect + m.RepublishTracks() + time.Sleep(50 * time.Millisecond) + + // TryPush should fail while republishing + require.ErrorIs(t, track.TryPush(Frame{Payload: []byte{0xff}}), ErrQueueFull) + + // SFU re-publishes with a new SID + request = expectEvent(t, transport.publishRequests) + require.Equal(t, uint32(track.handle), request.GetPubHandle()) + require.Equal(t, trackName, request.GetName()) + + newSID := livekit.TrackID("DTR_5678") + m.HandlePublishResponse(accepted(request, newSID)) + time.Sleep(50 * time.Millisecond) + + // SID updated in place, pushes succeed again + require.Equal(t, newSID, track.Info().SID) + require.NoError(t, track.TryPush(Frame{Payload: []byte{0xff}})) +} + +func TestLocalManager_RepublishRejected(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + result := publishAsync(context.Background(), m, "track") + request := expectEvent(t, transport.publishRequests) + m.HandlePublishResponse(accepted(request, "DTR_1234")) + track := expectEvent(t, result).track + + // Simulate reconnect + m.RepublishTracks() + request = expectEvent(t, transport.publishRequests) + + // SFU rejects the republish + handled := m.HandleRequestResponse(&livekit.RequestResponse{ + Request: &livekit.RequestResponse_PublishDataTrack{PublishDataTrack: request}, + Reason: livekit.RequestResponse_LIMIT_EXCEEDED, + }) + require.True(t, handled) + + // Track is unpublished and no longer republished on the next reconnect + expectEvent(t, track.Unpublished()) + require.False(t, track.IsPublished()) + require.ErrorIs(t, track.TryPush(Frame{Payload: []byte{0xff}}), ErrUnpublished) + + m.RepublishTracks() + select { + case request := <-transport.publishRequests: + require.Failf(t, "unexpected republish", "handle %d", request.GetPubHandle()) + case <-time.After(50 * time.Millisecond): + } +} + +func TestLocalManager_QueryPublished(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + // Publish two tracks + for _, name := range []string{"track_a", "track_b"} { + result := publishAsync(context.Background(), m, name) + request := expectEvent(t, transport.publishRequests) + m.HandlePublishResponse(accepted(request, "DTR_1234")) + require.NoError(t, expectEvent(t, result).err) + } + + published := m.PublishResponsesForSyncState() + require.Len(t, published, 2) + + var names []string + for _, response := range published { + names = append(names, response.GetInfo().GetName()) + } + require.Contains(t, names, "track_a") + require.Contains(t, names, "track_b") +} + +func TestLocalManager_ShutdownWithPendingAndActive(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + // Pending publication (no SFU response sent) + pending := publishAsync(context.Background(), m, "pending") + expectEvent(t, transport.publishRequests) + + // Active publication (fully published) + active := publishAsync(context.Background(), m, "active") + request := expectEvent(t, transport.publishRequests) + m.HandlePublishResponse(accepted(request, "DTR_1234")) + + res := expectEvent(t, active) + require.NoError(t, res.err) + activeTrack := res.track + require.True(t, activeTrack.IsPublished()) + + // Shutdown the manager + m.Shutdown() + time.Sleep(50 * time.Millisecond) + + // Pending publish receives disconnected error + require.ErrorIs(t, expectEvent(t, pending).err, ErrDisconnected) + + // Active track is no longer published + require.False(t, activeTrack.IsPublished()) +} diff --git a/datatrack/packet.go b/internal/datatrack/packet.go similarity index 100% rename from datatrack/packet.go rename to internal/datatrack/packet.go diff --git a/datatrack/packet_test.go b/internal/datatrack/packet_test.go similarity index 97% rename from datatrack/packet_test.go rename to internal/datatrack/packet_test.go index aa120dcf..5fca6c06 100644 --- a/datatrack/packet_test.go +++ b/internal/datatrack/packet_test.go @@ -25,7 +25,7 @@ import ( "github.com/stretchr/testify/require" ) -// testHeader is the header of the packet used by the Rust serialization tests, without extensions. +// testHeader is a representative packet header without extensions. func testHeader() dtp.Header { header := dtp.Header{ Version: supportedVersion, @@ -49,7 +49,7 @@ func testExtensions() Extensions { } } -// testPacket is the packet used by the Rust serialization tests. +// testPacket is a representative packet with all extensions. func testPacket() dtp.Packet { header := testHeader() if err := testExtensions().apply(&header); err != nil { diff --git a/datatrack/packetizer.go b/internal/datatrack/packetizer.go similarity index 100% rename from datatrack/packetizer.go rename to internal/datatrack/packetizer.go diff --git a/datatrack/packetizer_test.go b/internal/datatrack/packetizer_test.go similarity index 100% rename from datatrack/packetizer_test.go rename to internal/datatrack/packetizer_test.go diff --git a/datatrack/proto.go b/internal/datatrack/proto.go similarity index 77% rename from datatrack/proto.go rename to internal/datatrack/proto.go index 3b438311..112f19f5 100644 --- a/datatrack/proto.go +++ b/internal/datatrack/proto.go @@ -203,3 +203,73 @@ func extractTrackInfo(participant *livekit.ParticipantInfo) ([]Info, error) { } return infos, nil } + +// publishRequest is what the local participant asks the SFU to publish. +type publishRequest struct { + handle trackHandle + name string + usesE2EE bool + schema *SchemaID + frameEncoding FrameEncoding +} + +func (r publishRequest) toProto() *livekit.PublishDataTrackRequest { + req := &livekit.PublishDataTrackRequest{ + PubHandle: uint32(r.handle), + Name: r.name, + Encryption: livekit.Encryption_NONE, + } + if r.usesE2EE { + req.Encryption = livekit.Encryption_GCM + } + if r.schema != nil { + req.Schema = schemaIDToProto(*r.schema) + } + if r.frameEncoding != nil { + req.FrameEncoding = frameEncodingToProto(r.frameEncoding) + } + return req +} + +func unpublishRequestToProto(handle trackHandle) *livekit.UnpublishDataTrackRequest { + return &livekit.UnpublishDataTrackRequest{PubHandle: uint32(handle)} +} + +// publishResponse is the SFU's answer to a publishRequest: info when accepted, err when rejected. +type publishResponse struct { + handle trackHandle + info Info + err error +} + +// publishRejectionFromRequestResponse extracts a rejected publish request; ok is false when the +// response is not a publish rejection. +func publishRejectionFromRequestResponse(msg *livekit.RequestResponse) (rejection publishResponse, ok bool) { + request, isPublish := msg.GetRequest().(*livekit.RequestResponse_PublishDataTrack) + if !isPublish || msg.GetReason() == livekit.RequestResponse_OK { + return publishResponse{}, false + } + handle, err := handleFromUint32(request.PublishDataTrack.GetPubHandle()) + if err != nil { + return publishResponse{}, false + } + + rejection = publishResponse{handle: handle} + switch msg.GetReason() { + case livekit.RequestResponse_NOT_ALLOWED: + rejection.err = ErrNotAllowed + case livekit.RequestResponse_DUPLICATE_NAME: + rejection.err = ErrDuplicateName + case livekit.RequestResponse_INVALID_NAME: + rejection.err = ErrInvalidName + case livekit.RequestResponse_LIMIT_EXCEEDED: + rejection.err = ErrLimitReached + default: + rejection.err = fmt.Errorf("data track publish rejected (%s): %s", msg.GetReason(), msg.GetMessage()) + return rejection, true + } + if message := msg.GetMessage(); message != "" { + rejection.err = fmt.Errorf("%w: %s", rejection.err, message) + } + return rejection, true +} diff --git a/datatrack/proto_test.go b/internal/datatrack/proto_test.go similarity index 80% rename from datatrack/proto_test.go rename to internal/datatrack/proto_test.go index f1e40cf6..116ab0c3 100644 --- a/datatrack/proto_test.go +++ b/internal/datatrack/proto_test.go @@ -115,3 +115,29 @@ func TestProto_ExtractTrackInfo(t *testing.T) { require.Equal(t, "track1", infos[0].Name) require.Equal(t, livekit.TrackID("DTR_1234"), infos[0].SID) } + +func TestProto_PublishRequestToProto(t *testing.T) { + request := publishRequest{handle: 1, name: "track", usesE2EE: true}.toProto() + require.Equal(t, uint32(1), request.GetPubHandle()) + require.Equal(t, "track", request.GetName()) + require.Equal(t, livekit.Encryption_GCM, request.GetEncryption()) +} + +func TestProto_UnpublishRequestToProto(t *testing.T) { + request := unpublishRequestToProto(1) + require.Equal(t, uint32(1), request.GetPubHandle()) +} + +func TestProto_PublishRejectionFromRequestResponse(t *testing.T) { + response := &livekit.RequestResponse{ + Request: &livekit.RequestResponse_PublishDataTrack{ + PublishDataTrack: &livekit.PublishDataTrackRequest{PubHandle: 1}, + }, + Reason: livekit.RequestResponse_NOT_ALLOWED, + } + + rejection, ok := publishRejectionFromRequestResponse(response) + require.True(t, ok) + require.Equal(t, trackHandle(1), rejection.handle) + require.ErrorIs(t, rejection.err, ErrNotAllowed) +} diff --git a/datatrack/schema.go b/internal/datatrack/schema.go similarity index 100% rename from datatrack/schema.go rename to internal/datatrack/schema.go diff --git a/datatrack/schema_test.go b/internal/datatrack/schema_test.go similarity index 100% rename from datatrack/schema_test.go rename to internal/datatrack/schema_test.go diff --git a/localdatatrack.go b/localdatatrack.go new file mode 100644 index 00000000..c94ac405 --- /dev/null +++ b/localdatatrack.go @@ -0,0 +1,70 @@ +// 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 ( + "context" + "errors" + "time" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/server-sdk-go/v2/datatrack" + dtinternal "github.com/livekit/server-sdk-go/v2/internal/datatrack" +) + +const dataTrackPublishTimeout = 10 * time.Second + +// localDataTrackTransport connects the local data track manager to the engine. +type localDataTrackTransport struct { + engine *RTCEngine +} + +func (t localDataTrackTransport) SendPublishRequest(ctx context.Context, req *livekit.PublishDataTrackRequest) error { + if err := t.engine.ensurePublisherConnected(ctx, true); err != nil { + return err + } + return t.engine.SendPublishDataTrack(req) +} + +func (t localDataTrackTransport) SendUnpublishRequest(req *livekit.UnpublishDataTrackRequest) error { + return t.engine.SendUnpublishDataTrack(req) +} + +func (t localDataTrackTransport) SendFrame(frame dtinternal.FramePackets) { + t.engine.sendDataTrackFrame(frame) +} + +// PublishDataTrack publishes a data track and waits for the server to accept it. A 10 second +// deadline applies when ctx has none. The name must be unique among the participant's data tracks. +func (p *LocalParticipant) PublishDataTrack(ctx context.Context, name string, opts ...datatrack.PublishOption) (*datatrack.LocalTrack, error) { + options := datatrack.PublishOptions{Name: name} + for _, opt := range opts { + opt(&options) + } + + _, hasDeadline := ctx.Deadline() + ctx, cancel := withDefaultTimeout(ctx, dataTrackPublishTimeout) + defer cancel() + + track, err := p.dataTracks.Publish(ctx, options) + if err != nil { + if !hasDeadline && errors.Is(err, context.DeadlineExceeded) { + return nil, datatrack.ErrPublishTimeout + } + return nil, err + } + p.log.Infow("published data track", "name", name, "trackID", track.Info().SID) + return track, nil +} diff --git a/localdatatrack_test.go b/localdatatrack_test.go new file mode 100644 index 00000000..95bcdacd --- /dev/null +++ b/localdatatrack_test.go @@ -0,0 +1,39 @@ +// 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 ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// The room never connects, so the publish request waits on the engine. The caller's deadline must +// end that wait rather than the 15 s connect timeout. +func TestPublishDataTrackDeadlineInterruptsConnectionWait(t *testing.T) { + room := NewRoom(&RoomCallback{}) + t.Cleanup(room.engine.Close) + + const deadline = 50 * time.Millisecond + start := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), deadline) + defer cancel() + + _, err := room.LocalParticipant.PublishDataTrack(ctx, "test") + require.ErrorIs(t, err, context.DeadlineExceeded) + require.WithinRange(t, time.Now(), start.Add(deadline), start.Add(time.Second)) +} diff --git a/localparticipant.go b/localparticipant.go index fb651aa0..8f4a37e9 100644 --- a/localparticipant.go +++ b/localparticipant.go @@ -30,6 +30,7 @@ import ( "github.com/livekit/protocol/livekit" protoLogger "github.com/livekit/protocol/logger" + "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) const ( @@ -39,6 +40,7 @@ const ( type LocalParticipant struct { baseParticipant engine *RTCEngine + dataTracks *datatrack.LocalManager subscriptionPermission *livekit.SubscriptionPermission serverInfo *livekit.ServerInfo @@ -47,13 +49,15 @@ type LocalParticipant struct { } func newLocalParticipant(engine *RTCEngine, roomcallback *RoomCallback, serverInfo *livekit.ServerInfo, log protoLogger.Logger) *LocalParticipant { - return &LocalParticipant{ + p := &LocalParticipant{ baseParticipant: *newBaseParticipant(roomcallback, log.WithValues("isLocal", true)), engine: engine, serverInfo: serverInfo, rpcPendingAcks: &sync.Map{}, rpcPendingResponses: &sync.Map{}, } + p.dataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: engine}, Logger: p.log}) + return p } func (p *LocalParticipant) prepareTrackPublication(track webrtc.TrackLocal, opts *TrackPublicationOptions, pubOpts ...LocalTrackPublishOption) (*LocalTrackPublication, *livekit.AddTrackRequest, error) { @@ -1206,4 +1210,5 @@ func (p *LocalParticipant) SetLogger(log protoLogger.Logger) { pub.SetLogger(p.log) return true }) + p.dataTracks.SetLogger(p.log) } diff --git a/room.go b/room.go index fb0930e6..acbef166 100644 --- a/room.go +++ b/room.go @@ -778,6 +778,7 @@ func (r *Room) sendSyncState() { Subscribe: !sendUnsub, }, PublishTracks: publishedTracks, + PublishDataTracks: r.LocalParticipant.dataTracks.PublishResponsesForSyncState(), DataChannels: dataChannels, TrackSidsDisabled: trackSidsDisabled, // MIGRATION-TODO DatachannelReceiveStates @@ -787,6 +788,7 @@ func (r *Room) sendSyncState() { func (r *Room) cleanup() { r.engine.Close() r.LocalParticipant.closeTracks() + r.LocalParticipant.dataTracks.Shutdown() r.setSid("", true) r.byteStreamHandlers.Clear() @@ -1106,6 +1108,7 @@ func (r *Room) OnRestarted( r.OnParticipantUpdate(otherParticipants) r.LocalParticipant.republishTracks() + r.LocalParticipant.dataTracks.RepublishTracks() } func (r *Room) OnRestartConnected() { @@ -1401,6 +1404,9 @@ func (r *Room) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.Medi // OnRequestResponse handles responses to signal requests that no caller is waiting on. func (r *Room) OnRequestResponse(res *livekit.RequestResponse) { + if r.LocalParticipant.dataTracks.HandleRequestResponse(res) { + return + } if res.GetReason() != livekit.RequestResponse_OK { r.log.Warnw( "signal request failed", nil, @@ -1412,9 +1418,11 @@ func (r *Room) OnRequestResponse(res *livekit.RequestResponse) { } func (r *Room) OnPublishDataTrackResponse(publishDataTrackResponse *livekit.PublishDataTrackResponse) { + r.LocalParticipant.dataTracks.HandlePublishResponse(publishDataTrackResponse) } func (r *Room) OnUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit.UnpublishDataTrackResponse) { + r.LocalParticipant.dataTracks.HandleUnpublishResponse(unpublishDataTrackResponse) } func (r *Room) OnDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) {