diff --git a/datatracksender.go b/datatracksender.go new file mode 100644 index 00000000..495a0153 --- /dev/null +++ b/datatracksender.go @@ -0,0 +1,135 @@ +// 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" + + protoLogger "github.com/livekit/protocol/logger" + "github.com/pion/webrtc/v4" +) + +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 { + log protoLogger.Logger + + lock sync.Mutex + dc *webrtc.DataChannel + frame dataTrackFramePackets + notify chan struct{} + done chan struct{} +} + +func newDataTrackSender(log protoLogger.Logger) *dataTrackSender { + s := &dataTrackSender{ + log: log, + notify: make(chan struct{}, 1), + done: make(chan struct{}), + } + go s.run() + return s +} + +// setDataChannel points the sender at the channel it should write to. +func (s *dataTrackSender) setDataChannel(dc *webrtc.DataChannel) { + s.lock.Lock() + s.dc = dc + s.lock.Unlock() + + s.wake() +} + +func (s *dataTrackSender) setLogger(log protoLogger.Logger) { + s.log = log +} + +func (s *dataTrackSender) send(frame dataTrackFramePackets) { + if dropped := s.push(frame); dropped != nil { + s.log.Debugw("dropping data track frame", "numPackets", len(dropped)) + } +} + +func (s *dataTrackSender) wake() { + select { + case s.notify <- struct{}{}: + default: + } +} + +func (s *dataTrackSender) stop() { + close(s.done) +} + +func (s *dataTrackSender) push(frame dataTrackFramePackets) (dropped dataTrackFramePackets) { + if len(frame) == 0 { + return nil + } + + s.lock.Lock() + dropped, s.frame = s.frame, frame + s.lock.Unlock() + + s.wake() + return dropped +} + +func (s *dataTrackSender) pop() dataTrackFramePackets { + s.lock.Lock() + defer s.lock.Unlock() + + frame := s.frame + s.frame = nil + return frame +} + +func (s *dataTrackSender) run() { + var ( + dc *webrtc.DataChannel + inFlight dataTrackFramePackets + ) + for { + select { + case <-s.done: + return + case <-s.notify: + } + + s.lock.Lock() + current := s.dc + s.lock.Unlock() + if current != dc { + // A partially sent frame cannot be completed on a new channel. + dc, inFlight = current, nil + } + + for dc != nil && dc.ReadyState() == webrtc.DataChannelStateOpen && dc.BufferedAmount() <= dataTrackBufferedAmountLowThreshold { + if len(inFlight) == 0 { + if inFlight = s.pop(); inFlight == nil { + break + } + } + if err := dc.Send(inFlight[0]); err != nil { + s.log.Debugw("could not send data track packet", "error", err) + } + inFlight = inFlight[1:] + } + } +} diff --git a/datatracksender_test.go b/datatracksender_test.go new file mode 100644 index 00000000..a76ae0e3 --- /dev/null +++ b/datatracksender_test.go @@ -0,0 +1,55 @@ +// 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" + + "github.com/stretchr/testify/require" +) + +func testFrame(marker byte, packets int) dataTrackFramePackets { + frame := make(dataTrackFramePackets, packets) + for i := range frame { + frame[i] = []byte{marker, byte(i)} + } + return frame +} + +func TestDataTrackSenderQueue(t *testing.T) { + s := newDataTrackSender(logger) + t.Cleanup(s.stop) + + require.Nil(t, s.push(nil)) + require.Nil(t, s.pop()) + + multi := testFrame(0xaa, 13) + require.Nil(t, s.push(multi)) + require.Equal(t, multi, s.pop()) + + older, newer := testFrame(0x01, 4), testFrame(0x02, 3) + require.Nil(t, s.push(older)) + require.Equal(t, older, s.push(newer)) + require.Equal(t, newer, s.pop()) + require.Nil(t, s.pop()) +} + +func TestSendDataTrackFrameKeepsFreshest(t *testing.T) { + engine := newTestEngine(t) + + engine.sendDataTrackFrame(testFrame(0x01, 2)) + engine.sendDataTrackFrame(testFrame(0x02, 2)) + require.Equal(t, testFrame(0x02, 2), engine.dataTrackSender.pop()) +} diff --git a/engine.go b/engine.go index 6356e687..62bcf93d 100644 --- a/engine.go +++ b/engine.go @@ -80,6 +80,7 @@ type engineHandler interface { OnPublishDataTrackResponse(publishDataTrackResponse *livekit.PublishDataTrackResponse) OnUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit.UnpublishDataTrackResponse) OnDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) + OnDataTrackPacket(data []byte) } // ------------------------------------------- @@ -92,8 +93,9 @@ var ( // ------------------------------------------- const ( - reliableDataChannelName = "_reliable" - lossyDataChannelName = "_lossy" + reliableDataChannelName = "_reliable" + lossyDataChannelName = "_lossy" + dataTrackDataChannelName = "_data_track" maxReconnectCount = 10 initialReconnectInterval = 300 * time.Millisecond @@ -127,9 +129,13 @@ type RTCEngine struct { lossyDC *webrtc.DataChannel reliableDCSub *webrtc.DataChannel lossyDCSub *webrtc.DataChannel + dataTrackDC *webrtc.DataChannel + dataTrackDCSub *webrtc.DataChannel reliableMsgLock sync.Mutex reliableMsgSeq uint32 + dataTrackSender *dataTrackSender + trackPublishedListenersLock sync.Mutex trackPublishedListeners map[string]chan *livekit.TrackPublishedResponse @@ -170,6 +176,7 @@ func NewRTCEngine( Logger: e.log, Processor: e, }) + e.dataTrackSender = newDataTrackSender(e.log) e.configureSignalling(useSinglePeerConnection) return e @@ -202,6 +209,7 @@ func (e *RTCEngine) configureSignalling(useSinglePeerConnection bool) { // SetLogger overrides default logger. func (e *RTCEngine) SetLogger(l protoLogger.Logger) { e.log = l + e.dataTrackSender.setLogger(l) e.connectionManager.setLogger(l) e.signalling.SetLogger(l) e.signalHandler.SetLogger(l) @@ -369,6 +377,7 @@ func (e *RTCEngine) Close() { e.connectionManager.setClosed() e.abortPendingRequests() + e.dataTrackSender.stop() e.pclock.Lock() e.pendingPublisherOffer = webrtc.SessionDescription{} @@ -557,6 +566,20 @@ func (e *RTCEngine) createPublisherPCLocked(configuration webrtc.Configuration) return err } e.reliableDC.OnMessage(e.handleDataPacket) + + e.dataTrackDC, err = e.publisher.pc.CreateDataChannel(dataTrackDataChannelName, &webrtc.DataChannelInit{ + Ordered: &falseVal, + MaxRetransmits: new(uint16), + }) + if err != nil { + e.dclock.Unlock() + return err + } + e.dataTrackDC.OnMessage(e.handleDataTrackPacket) + e.dataTrackDC.SetBufferedAmountLowThreshold(dataTrackBufferedAmountLowThreshold) + e.dataTrackDC.OnBufferedAmountLow(e.dataTrackSender.wake) + e.dataTrackDC.OnOpen(e.dataTrackSender.wake) + e.dataTrackSender.setDataChannel(e.dataTrackDC) e.dclock.Unlock() return nil @@ -643,6 +666,10 @@ func (e *RTCEngine) createSubscriberPCLocked(configuration webrtc.Configuration) e.reliableDCSub = c } else if c.Label() == lossyDataChannelName { e.lossyDCSub = c + } else if c.Label() == dataTrackDataChannelName { + e.dataTrackDCSub = c + c.OnMessage(e.handleDataTrackPacket) + return } else { return } @@ -750,7 +777,9 @@ func (e *RTCEngine) ensurePublisherConnected(ensureDataReady bool) error { func (e *RTCEngine) dataPubChannelReady() bool { e.dclock.RLock() defer e.dclock.RUnlock() - return e.reliableDC.ReadyState() == webrtc.DataChannelStateOpen && e.lossyDC.ReadyState() == webrtc.DataChannelStateOpen + return e.reliableDC.ReadyState() == webrtc.DataChannelStateOpen && + e.lossyDC.ReadyState() == webrtc.DataChannelStateOpen && + e.dataTrackDC.ReadyState() == webrtc.DataChannelStateOpen } func (e *RTCEngine) RegisterTrackPublishedListener(cid string, c chan *livekit.TrackPublishedResponse) { @@ -909,6 +938,13 @@ func (e *RTCEngine) handleDataPacket(msg webrtc.DataChannelMessage) { } } +func (e *RTCEngine) handleDataTrackPacket(msg webrtc.DataChannelMessage) { + if msg.IsString { + return + } + e.engineHandler.OnDataTrackPacket(msg.Data) +} + func (e *RTCEngine) readDataPacket(msg webrtc.DataChannelMessage) (*livekit.DataPacket, error) { dataPacket := &livekit.DataPacket{} if msg.IsString { @@ -1809,3 +1845,7 @@ func waitUntilConnected(d time.Duration, test func() bool) error { } } } + +func (e *RTCEngine) sendDataTrackFrame(frame dataTrackFramePackets) { + e.dataTrackSender.send(frame) +} diff --git a/room.go b/room.go index ca5cc6d0..4cddd5e5 100644 --- a/room.go +++ b/room.go @@ -1415,6 +1415,9 @@ func (r *Room) OnUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit. func (r *Room) OnDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) { } +func (r *Room) OnDataTrackPacket(data []byte) { +} + func (r *Room) OnStreamHeader(streamHeader *livekit.DataStream_Header, participantIdentity string) { switch header := streamHeader.ContentHeader.(type) { case *livekit.DataStream_Header_TextHeader: