Data track subscribing

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
 	}

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")
 )

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       SID
+	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
+}

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, SID("DTR_1234"), updates["publisher"][0].SID)
+	require.Empty(t, updates["leaving"])
+}