Send demand updates

pkg/rtc/datatrack.go

@@ -40,6 +40,8 @@ type DataTrackParams struct {
 	ParticipantID       func() livekit.ParticipantID
 	ParticipantIdentity livekit.ParticipantIdentity
 	BytesTrackStats     *BytesTrackStats
+	// invoked when the number of subscribers changes on a dynacasted track
+	OnSubscriberCountChanged func(subscriberCount uint32)
 }
 
 type subscribedDataTrack struct {
@@ -115,9 +117,9 @@ func (d *DataTrack) Name() string {
 
 func (d *DataTrack) AddSubscriber(sub types.LocalParticipant) (types.DataDownTrack, error) {
 	d.lock.Lock()
-	defer d.lock.Unlock()
 
 	if _, ok := d.subscribedTracks[sub.ID()]; ok {
+		d.lock.Unlock()
 		return nil, errAlreadySubscribed
 	}
 
@@ -139,6 +141,7 @@ func (d *DataTrack) AddSubscriber(sub types.LocalParticipant) (types.DataDownTra
 		BytesTrackStats:  bytesStats,
 	})
 	if err != nil {
+		d.lock.Unlock()
 		bytesStats.Stop()
 		return nil, err
 	}
@@ -147,6 +150,10 @@ func (d *DataTrack) AddSubscriber(sub types.LocalParticipant) (types.DataDownTra
 		subscriber:    sub,
 		dataDownTrack: dataDownTrack,
 	}
+	subscriberCount := len(d.subscribedTracks)
+	d.lock.Unlock()
+
+	d.notifySubscriberCountChanged(subscriberCount)
 	return dataDownTrack, nil
 }
 
@@ -154,13 +161,22 @@ func (d *DataTrack) RemoveSubscriber(subID livekit.ParticipantID) {
 	d.lock.Lock()
 	subscribedTrack, ok := d.subscribedTracks[subID]
 	delete(d.subscribedTracks, subID)
+	subscriberCount := len(d.subscribedTracks)
 	d.lock.Unlock()
 
 	if ok {
 		subscribedTrack.dataDownTrack.Close()
+		d.notifySubscriberCountChanged(subscriberCount)
 	}
 }
 
+func (d *DataTrack) notifySubscriberCountChanged(subscriberCount int) {
+	if !d.dti.IsDynacasted || d.params.OnSubscriberCountChanged == nil {
+		return
+	}
+	d.params.OnSubscriberCountChanged(uint32(subscriberCount))
+}
+
 func (d *DataTrack) IsSubscriber(subID livekit.ParticipantID) bool {
 	d.lock.Lock()
 	defer d.lock.Unlock()

pkg/rtc/participant.go

@@ -1403,6 +1403,9 @@ func (p *ParticipantImpl) SetMigrateInfo(
 					p.params.TelemetryListener,
 					p.params.Reporter,
 				),
+				OnSubscriberCountChanged: func(subscriberCount uint32) {
+					p.sendDataTrackDemandUpdate(dti.PubHandle, subscriberCount)
+				},
 			},
 			dti,
 		)

pkg/rtc/participant_data_track.go

@@ -124,6 +124,9 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData
 				p.params.TelemetryListener,
 				p.params.Reporter,
 			),
+			OnSubscriberCountChanged: func(subscriberCount uint32) {
+				p.sendDataTrackDemandUpdate(dti.PubHandle, subscriberCount)
+			},
 		},
 		dti,
 	)

pkg/rtc/participant_signal.go

@@ -369,6 +369,13 @@ func (p *ParticipantImpl) SendDataTrackSubscriberHandles(handles map[uint32]*liv
 	}))
 }
 
+func (p *ParticipantImpl) sendDataTrackDemandUpdate(pubHandle uint32, subscriberCount uint32) error {
+	return p.signaller.WriteMessage(p.signalling.SignalDataTrackDemandUpdate(&livekit.DataTrackDemandUpdate{
+		PubHandle:       pubHandle,
+		SubscriberCount: subscriberCount,
+	}))
+}
+
 func (p *ParticipantImpl) sendStoreDataBlobResponse(requestId uint32, key *livekit.DataBlobKey) error {
 	return p.signaller.WriteMessage(p.signalling.SignalStoreDataBlobResponse(&livekit.StoreDataBlobResponse{
 		RequestId: requestId,

pkg/rtc/signalling/interfaces.go

@@ -62,6 +62,7 @@ type ParticipantSignalling interface {
 	SignalPublishDataTrackResponse(publishDataTrackResponse *livekit.PublishDataTrackResponse) proto.Message
 	SignalUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit.UnpublishDataTrackResponse) proto.Message
 	SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) proto.Message
+	SignalDataTrackDemandUpdate(dataTrackDemandUpdate *livekit.DataTrackDemandUpdate) proto.Message
 	SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message
 	SignalGetDataBlobResponse(getDataBlobResponse *livekit.GetDataBlobResponse) proto.Message
 }

pkg/rtc/signalling/signalling.go

@@ -259,6 +259,14 @@ func (s *signalling) SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles
 	}
 }
 
+func (s *signalling) SignalDataTrackDemandUpdate(dataTrackDemandUpdate *livekit.DataTrackDemandUpdate) proto.Message {
+	return &livekit.SignalResponse{
+		Message: &livekit.SignalResponse_DataTrackDemandUpdate{
+			DataTrackDemandUpdate: dataTrackDemandUpdate,
+		},
+	}
+}
+
 func (s *signalling) SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message {
 	return &livekit.SignalResponse{
 		Message: &livekit.SignalResponse_StoreDataBlobResponse{

pkg/rtc/signalling/signallingunimplemented.go

@@ -128,6 +128,10 @@ func (u *signallingUnimplemented) SignalDataTrackSubscriberHandles(dataTrackSubs
 	return nil
 }
 
+func (u *signallingUnimplemented) SignalDataTrackDemandUpdate(dataTrackDemandUpdate *livekit.DataTrackDemandUpdate) proto.Message {
+	return nil
+}
+
 func (u *signallingUnimplemented) SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message {
 	return nil
 }