From fbb1f9fc80d55c70c3b92dd370d1a95b06a71e6d Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 6 Jul 2022 17:27:32 +0530 Subject: [PATCH] Move subscribe/unsubscribe queue to participant. (#813) * Move subscribe/unsubscribe queue to participant. As subscribe/unsubscribe operation can come from both local media track or remote media track, participant needs to have it. * Remove comment * Stop reneg timer on close * address comments --- pkg/rtc/mediatrack.go | 2 + pkg/rtc/mediatrackreceiver.go | 86 ++++++-- pkg/rtc/mediatracksubscriptions.go | 187 ++---------------- pkg/rtc/participant.go | 141 +++++++++++-- pkg/rtc/subscribedtrack.go | 5 + pkg/rtc/transport.go | 9 + pkg/rtc/types/interfaces.go | 7 + .../typesfakes/fake_local_media_track.go | 65 ++++++ .../typesfakes/fake_local_participant.go | 162 +++++++++++++++ pkg/rtc/types/typesfakes/fake_media_track.go | 65 ++++++ .../types/typesfakes/fake_subscribed_track.go | 65 ++++++ 11 files changed, 600 insertions(+), 194 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 9426dd17e..19b1d9494 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -38,6 +38,7 @@ type MediaTrackParams struct { SdpCid string ParticipantID livekit.ParticipantID ParticipantIdentity livekit.ParticipantIdentity + ParticipantVersion uint32 // channel to send RTCP packets to the source RTCPChan chan []rtcp.Packet BufferFactory *buffer.Factory @@ -61,6 +62,7 @@ func NewMediaTrack(params MediaTrackParams) *MediaTrack { MediaTrack: t, ParticipantID: params.ParticipantID, ParticipantIdentity: params.ParticipantIdentity, + ParticipantVersion: params.ParticipantVersion, BufferFactory: params.BufferFactory, ReceiverConfig: params.ReceiverConfig, SubscriberConfig: params.SubscriberConfig, diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index aff5ed43a..d77e8bdb8 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -38,6 +38,20 @@ func (r *simulcastReceiver) Priority() int { return r.priority } +type MediaTrackReceiverParams struct { + TrackInfo *livekit.TrackInfo + MediaTrack types.MediaTrack + ParticipantID livekit.ParticipantID + ParticipantIdentity livekit.ParticipantIdentity + ParticipantVersion uint32 + BufferFactory *buffer.Factory + ReceiverConfig ReceiverConfig + SubscriberConfig DirectionConfig + VideoConfig config.VideoConfig + Telemetry telemetry.TelemetryService + Logger logger.Logger +} + type MediaTrackReceiver struct { params MediaTrackReceiverParams muted atomic.Bool @@ -62,19 +76,6 @@ type MediaTrackReceiver struct { *MediaTrackSubscriptions } -type MediaTrackReceiverParams struct { - TrackInfo *livekit.TrackInfo - MediaTrack types.MediaTrack - ParticipantID livekit.ParticipantID - ParticipantIdentity livekit.ParticipantIdentity - BufferFactory *buffer.Factory - ReceiverConfig ReceiverConfig - SubscriberConfig DirectionConfig - VideoConfig config.VideoConfig - Telemetry telemetry.TelemetryService - Logger logger.Logger -} - func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver { t := &MediaTrackReceiver{ params: params, @@ -92,6 +93,9 @@ func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver Logger: params.Logger, }) t.MediaTrackSubscriptions.OnDownTrackCreated(t.onDownTrackCreated) + t.MediaTrackSubscriptions.OnSubscriptionOperationComplete(func(sub types.LocalParticipant) { + go sub.ClearInProgressAndProcessSubscriptionRequestsQueue(t.ID()) + }) if t.trackInfo.Muted { t.SetMuted(true) @@ -305,6 +309,10 @@ func (t *MediaTrackReceiver) PublisherIdentity() livekit.ParticipantIdentity { return t.params.ParticipantIdentity } +func (t *MediaTrackReceiver) PublisherVersion() uint32 { + return t.params.ParticipantVersion +} + func (t *MediaTrackReceiver) IsSimulcast() bool { return t.simulcasted.Load() } @@ -349,6 +357,12 @@ func (t *MediaTrackReceiver) AddOnClose(f func()) { // AddSubscriber subscribes sub to current mediaTrack func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) error { + trackID := t.ID() + sub.EnqueueSubscribeTrack(trackID, t.addSubscriber) + return nil +} + +func (t *MediaTrackReceiver) addSubscriber(sub types.LocalParticipant) error { t.lock.RLock() receivers := t.receiversShadow potentialCodecs := make([]webrtc.RTPCodecParameters, len(t.potentialCodecs)) @@ -390,6 +404,52 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) error { return nil } +// RemoveSubscriber removes participant from subscription +// stop all forwarders to the client +func (t *MediaTrackReceiver) RemoveSubscriber(subscriberID livekit.ParticipantID, willBeResumed bool) { + subTrack := t.getSubscribedTrack(subscriberID) + if subTrack == nil { + return + } + + sub := subTrack.Subscriber() + trackID := subTrack.ID() + sub.EnqueueUnsubscribeTrack(trackID, willBeResumed, t.MediaTrackSubscriptions.RemoveSubscriber) +} + +func (t *MediaTrackReceiver) RemoveAllSubscribers(willBeResumed bool) { + t.params.Logger.Debugw("removing all subscribers") + for _, subscriberID := range t.MediaTrackSubscriptions.GetAllSubscribers() { + t.RemoveSubscriber(subscriberID, willBeResumed) + } +} + +func (t *MediaTrackReceiver) RevokeDisallowedSubscribers(allowedSubscriberIdentities []livekit.ParticipantIdentity) []livekit.ParticipantIdentity { + var revokedSubscriberIdentities []livekit.ParticipantIdentity + + // LK-TODO: large number of subscribers needs to be solved for this loop + for _, subTrack := range t.MediaTrackSubscriptions.getAllSubscribedTracks() { + found := false + for _, allowedIdentity := range allowedSubscriberIdentities { + if subTrack.SubscriberIdentity() == allowedIdentity { + found = true + break + } + } + + if !found { + t.params.Logger.Infow("revoking subscription", + "subscriber", subTrack.SubscriberIdentity(), + "subscriberID", subTrack.SubscriberID(), + ) + t.RemoveSubscriber(subTrack.SubscriberID(), false) + revokedSubscriberIdentities = append(revokedSubscriberIdentities, subTrack.SubscriberIdentity()) + } + } + + return revokedSubscriberIdentities +} + func (t *MediaTrackReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) { t.lock.Lock() t.trackInfo = proto.Clone(ti).(*livekit.TrackInfo) diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 41aa7ebd2..76547dd34 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -34,30 +34,12 @@ var ( errNotFound = errors.New("not found") ) -type SubscribeRequestType int - -const ( - SubscribeRequestTypeRemove SubscribeRequestType = iota - SubscribeRequestTypeAdd -) - -type SubscribeRequest struct { - requestType SubscribeRequestType - sub types.LocalParticipant - wr *WrappedReceiver - willBeResumed bool -} - // MediaTrackSubscriptions manages subscriptions of a media track type MediaTrackSubscriptions struct { params MediaTrackSubscriptionsParams subscribedTracksMu sync.RWMutex subscribedTracks map[livekit.ParticipantID]types.SubscribedTrack - inProgress map[livekit.ParticipantID]bool - requestsQueue map[livekit.ParticipantID][]SubscribeRequest - - onNoSubscribers func() // quality level enable/disable maxQualityLock sync.RWMutex @@ -70,7 +52,8 @@ type MediaTrackSubscriptions struct { qualityNotifyOpQueue *utils.OpsQueue - onDownTrackCreated func(downTrack *sfu.DownTrack) + onDownTrackCreated func(downTrack *sfu.DownTrack) + onSubscriptionOperationComplete func(sub types.LocalParticipant) } type MediaTrackSubscriptionsParams struct { @@ -90,8 +73,6 @@ func NewMediaTrackSubscriptions(params MediaTrackSubscriptionsParams) *MediaTrac t := &MediaTrackSubscriptions{ params: params, subscribedTracks: make(map[livekit.ParticipantID]types.SubscribedTrack), - inProgress: make(map[livekit.ParticipantID]bool), - requestsQueue: make(map[livekit.ParticipantID][]SubscribeRequest), maxSubscriberQuality: make(map[livekit.ParticipantID]*types.SubscribedCodecQuality), maxSubscriberNodeQuality: make(map[livekit.NodeID][]types.SubscribedCodecQuality), maxSubscribedQuality: make(map[string]livekit.VideoQuality), @@ -119,14 +100,14 @@ func (t *MediaTrackSubscriptions) Close() { t.qualityNotifyOpQueue.Stop() } -func (t *MediaTrackSubscriptions) OnNoSubscribers(f func()) { - t.onNoSubscribers = f -} - func (t *MediaTrackSubscriptions) OnDownTrackCreated(f func(downTrack *sfu.DownTrack)) { t.onDownTrackCreated = f } +func (t *MediaTrackSubscriptions) OnSubscriptionOperationComplete(f func(sub types.LocalParticipant)) { + t.onSubscriptionOperationComplete = f +} + func (t *MediaTrackSubscriptions) SetMuted(muted bool) { // update quality based on subscription if unmuting. // This will queue up the current state, but subscriber @@ -155,64 +136,8 @@ func (t *MediaTrackSubscriptions) AddCodec(mime string) { t.subscribedTracksMu.Unlock() } -func (t *MediaTrackSubscriptions) processRequestsQueue(subscriberID livekit.ParticipantID) { - t.subscribedTracksMu.Lock() - if t.inProgress[subscriberID] || len(t.requestsQueue[subscriberID]) == 0 { - t.subscribedTracksMu.Unlock() - return - } - - request := t.requestsQueue[subscriberID][0] - t.requestsQueue[subscriberID] = t.requestsQueue[subscriberID][1:] - if len(t.requestsQueue[subscriberID]) == 0 { - delete(t.requestsQueue, subscriberID) - } - - t.inProgress[subscriberID] = true - t.subscribedTracksMu.Unlock() - - switch request.requestType { - case SubscribeRequestTypeAdd: - err := t.addSubscriber(request.sub, request.wr) - if err != nil { - if err != errAlreadySubscribed { - t.params.Logger.Errorw("error adding subscriber", err, "subscriberID", subscriberID) - } - - // process pending request even if adding errors out - go t.clearInProgressAndProcessRequestsQueue(subscriberID) - } - - case SubscribeRequestTypeRemove: - err := t.removeSubscriber(subscriberID, request.willBeResumed) - if err != nil { - go t.clearInProgressAndProcessRequestsQueue(subscriberID) - } - - default: - t.params.Logger.Warnw("unknown request type", nil) - - // let the queue move forward - go t.clearInProgressAndProcessRequestsQueue(subscriberID) - } -} - // AddSubscriber subscribes sub to current mediaTrack func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *WrappedReceiver) error { - subscriberID := sub.ID() - t.subscribedTracksMu.Lock() - t.requestsQueue[subscriberID] = append(t.requestsQueue[subscriberID], SubscribeRequest{ - requestType: SubscribeRequestTypeAdd, - sub: sub, - wr: wr, - }) - t.subscribedTracksMu.Unlock() - - t.processRequestsQueue(subscriberID) - return nil -} - -func (t *MediaTrackSubscriptions) addSubscriber(sub types.LocalParticipant, wr *WrappedReceiver) error { trackID := t.params.MediaTrack.ID() subscriberID := sub.ID() @@ -254,6 +179,7 @@ func (t *MediaTrackSubscriptions) addSubscriber(sub types.LocalParticipant, wr * subTrack := NewSubscribedTrack(SubscribedTrackParams{ PublisherID: t.params.MediaTrack.PublisherID(), PublisherIdentity: t.params.MediaTrack.PublisherIdentity(), + PublisherVersion: t.params.MediaTrack.PublisherVersion(), Subscriber: sub, MediaTrack: t.params.MediaTrack, DownTrack: downTrack, @@ -396,7 +322,9 @@ func (t *MediaTrackSubscriptions) addSubscriber(sub types.LocalParticipant, wr * sub.Negotiate(false) } - t.clearInProgressAndProcessRequestsQueue(subscriberID) + if t.onSubscriptionOperationComplete != nil { + t.onSubscriptionOperationComplete(sub) + } }() t.params.Telemetry.TrackSubscribed( @@ -413,51 +341,17 @@ func (t *MediaTrackSubscriptions) addSubscriber(sub types.LocalParticipant, wr * // RemoveSubscriber removes participant from subscription // stop all forwarders to the client -func (t *MediaTrackSubscriptions) RemoveSubscriber(subscriberID livekit.ParticipantID, willBeResumed bool) { - t.subscribedTracksMu.Lock() - t.requestsQueue[subscriberID] = append(t.requestsQueue[subscriberID], SubscribeRequest{ - requestType: SubscribeRequestTypeRemove, - willBeResumed: willBeResumed, - }) - t.subscribedTracksMu.Unlock() - - t.processRequestsQueue(subscriberID) -} - -func (t *MediaTrackSubscriptions) removeSubscriber(subscriberID livekit.ParticipantID, willBeResumed bool) error { - t.params.Logger.Debugw("removing subscriber", "subscriberID", subscriberID, "willBeResumed", willBeResumed) +func (t *MediaTrackSubscriptions) RemoveSubscriber(subscriberID livekit.ParticipantID, willBeResumed bool) error { subTrack := t.getSubscribedTrack(subscriberID) if subTrack == nil { return errNotFound } + t.params.Logger.Debugw("removing subscriber", "subscriberID", subscriberID, "willBeResumed", willBeResumed) t.closeSubscribedTrack(subTrack, willBeResumed) return nil } -func (t *MediaTrackSubscriptions) RemoveAllSubscribers(willBeResumed bool) { - t.params.Logger.Debugw("removing all subscribers") - - var subIDs []livekit.ParticipantID - t.subscribedTracksMu.Lock() - for _, subTrack := range t.getAllSubscribedTracksLocked() { - subscriberID := subTrack.SubscriberID() - t.requestsQueue[subscriberID] = append(t.requestsQueue[subscriberID], SubscribeRequest{ - requestType: SubscribeRequestTypeRemove, - willBeResumed: willBeResumed, - }) - - subIDs = append(subIDs, subscriberID) - } - t.subscribedTracksMu.Unlock() - - for _, subID := range subIDs { - t.processRequestsQueue(subID) - } - - t.maybeNotifyNoSubscribers() -} - func (t *MediaTrackSubscriptions) closeSubscribedTrack(subTrack types.SubscribedTrack, willBeResumed bool) { dt := subTrack.DownTrack() if dt == nil { @@ -483,32 +377,6 @@ func (t *MediaTrackSubscriptions) ResyncAllSubscribers() { } } -func (t *MediaTrackSubscriptions) RevokeDisallowedSubscribers(allowedSubscriberIdentities []livekit.ParticipantIdentity) []livekit.ParticipantIdentity { - var revokedSubscriberIdentities []livekit.ParticipantIdentity - - // LK-TODO: large number of subscribers needs to be solved for this loop - for _, subTrack := range t.getAllSubscribedTracks() { - found := false - for _, allowedIdentity := range allowedSubscriberIdentities { - if subTrack.SubscriberIdentity() == allowedIdentity { - found = true - break - } - } - - if !found { - t.params.Logger.Infow("revoking subscription", - "subscriber", subTrack.SubscriberIdentity(), - "subscriberID", subTrack.SubscriberID(), - ) - t.RemoveSubscriber(subTrack.SubscriberID(), false) - revokedSubscriberIdentities = append(revokedSubscriberIdentities, subTrack.SubscriberIdentity()) - } - } - - return revokedSubscriberIdentities -} - func (t *MediaTrackSubscriptions) GetAllSubscribers() []livekit.ParticipantID { t.subscribedTracksMu.RLock() defer t.subscribedTracksMu.RUnlock() @@ -850,26 +718,18 @@ func (t *MediaTrackSubscriptions) stopMaxQualityTimer() { } } -func (t *MediaTrackSubscriptions) maybeNotifyNoSubscribers() { - if t.onNoSubscribers == nil { - return - } - - t.subscribedTracksMu.RLock() - empty := len(t.subscribedTracks) == 0 && len(t.inProgress) == 0 && len(t.requestsQueue) == 0 - t.subscribedTracksMu.RUnlock() - - if empty { - t.onNoSubscribers() - } -} - func (t *MediaTrackSubscriptions) downTrackClosed( sub types.LocalParticipant, subTrack types.SubscribedTrack, willBeResumed bool, sender *webrtc.RTPSender, ) { + defer func() { + if t.onSubscriptionOperationComplete != nil { + t.onSubscriptionOperationComplete(sub) + } + }() + subscriberID := sub.ID() t.subscribedTracksMu.Lock() delete(t.subscribedTracks, subscriberID) @@ -914,15 +774,4 @@ func (t *MediaTrackSubscriptions) downTrackClosed( if !willBeResumed { sub.Negotiate(false) } - - t.clearInProgressAndProcessRequestsQueue(subscriberID) - t.maybeNotifyNoSubscribers() -} - -func (t *MediaTrackSubscriptions) clearInProgressAndProcessRequestsQueue(subscriberID livekit.ParticipantID) { - t.subscribedTracksMu.Lock() - delete(t.inProgress, subscriberID) - t.subscribedTracksMu.Unlock() - - t.processRequestsQueue(subscriberID) } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 68baad360..03a9c1267 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -49,6 +49,20 @@ type downTrackState struct { forwarder sfu.ForwarderState } +type SubscribeRequestType int + +const ( + SubscribeRequestTypeRemove SubscribeRequestType = iota + SubscribeRequestTypeAdd +) + +type SubscribeRequest struct { + requestType SubscribeRequestType + willBeResumed bool + addCb func(sub types.LocalParticipant) error + removeCb func(subscriberID livekit.ParticipantID, willBeResumed bool) error +} + type ParticipantParams struct { Identity livekit.ParticipantIdentity Name livekit.ParticipantName @@ -144,6 +158,10 @@ type ParticipantImpl struct { iceConfig types.IceConfig cachedDownTracks map[livekit.TrackID]*downTrackState + + subscriptionInProgress map[livekit.TrackID]bool + subscriptionRequestsQueue map[livekit.TrackID][]SubscribeRequest + trackPublisherVersion map[livekit.TrackID]uint32 } func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { @@ -157,16 +175,19 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { return nil, ErrMissingGrants } p := &ParticipantImpl{ - params: params, - rtcpCh: make(chan []rtcp.Packet, 100), - pendingTracks: make(map[string]*pendingTrackInfo), - subscribedTracks: make(map[livekit.TrackID]types.SubscribedTrack), - subscribedTracksSettings: make(map[livekit.TrackID]*livekit.UpdateTrackSettings), - disallowedSubscriptions: make(map[livekit.TrackID]livekit.ParticipantID), - subscribedTo: make(map[livekit.ParticipantID]struct{}), - connectedAt: time.Now(), - rttUpdatedAt: time.Now(), - cachedDownTracks: make(map[livekit.TrackID]*downTrackState), + params: params, + rtcpCh: make(chan []rtcp.Packet, 100), + pendingTracks: make(map[string]*pendingTrackInfo), + subscribedTracks: make(map[livekit.TrackID]types.SubscribedTrack), + subscribedTracksSettings: make(map[livekit.TrackID]*livekit.UpdateTrackSettings), + disallowedSubscriptions: make(map[livekit.TrackID]livekit.ParticipantID), + subscribedTo: make(map[livekit.ParticipantID]struct{}), + connectedAt: time.Now(), + rttUpdatedAt: time.Now(), + cachedDownTracks: make(map[livekit.TrackID]*downTrackState), + subscriptionInProgress: make(map[livekit.TrackID]bool), + subscriptionRequestsQueue: make(map[livekit.TrackID][]SubscribeRequest), + trackPublisherVersion: make(map[livekit.TrackID]uint32), } p.version.Store(params.InitialVersion) p.migrateState.Store(types.MigrateStateInit) @@ -954,8 +975,17 @@ func (p *ParticipantImpl) AddSubscribedTrack(subTrack types.SubscribedTrack) { "publisherIdentity", subTrack.PublisherIdentity(), "trackID", subTrack.ID()) p.lock.Lock() + if v, ok := p.trackPublisherVersion[subTrack.ID()]; ok && v > subTrack.PublisherVersion() { + p.lock.Unlock() + p.params.Logger.Infow("ignoring add subscribedTrack from older version", "current", v, "requesting", subTrack.PublisherVersion()) + return + } + p.trackPublisherVersion[subTrack.ID()] = subTrack.PublisherVersion() + onSubscribedTo := p.onSubscribedTo + p.subscribedTracks[subTrack.ID()] = subTrack + settings := p.subscribedTracksSettings[subTrack.ID()] p.lock.Unlock() @@ -987,10 +1017,16 @@ func (p *ParticipantImpl) RemoveSubscribedTrack(subTrack types.SubscribedTrack) "publisherIdentity", subTrack.PublisherIdentity(), "trackID", subTrack.ID(), "kind", subTrack.DownTrack().Kind()) - p.subscriber.RemoveTrack(subTrack) - p.lock.Lock() + if v, ok := p.trackPublisherVersion[subTrack.ID()]; ok && v > subTrack.PublisherVersion() { + p.lock.Unlock() + p.params.Logger.Infow("ignoring remove subscribedTrack from older version", "current", v, "requesting", subTrack.PublisherVersion()) + return + } + p.trackPublisherVersion[subTrack.ID()] = subTrack.PublisherVersion() + delete(p.subscribedTracks, subTrack.ID()) + // remove from subscribed map numRemaining := 0 for _, st := range p.subscribedTracks { @@ -1010,6 +1046,8 @@ func (p *ParticipantImpl) RemoveSubscribedTrack(subTrack types.SubscribedTrack) } p.lock.Unlock() + p.subscriber.RemoveTrack(subTrack) + if numRemaining == 0 { // // When a participant leaves OR @@ -1659,6 +1697,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei SdpCid: track.ID(), ParticipantID: p.params.SID, ParticipantIdentity: p.params.Identity, + ParticipantVersion: p.version.Load(), RTCPChan: p.rtcpCh, BufferFactory: p.params.Config.BufferFactory, ReceiverConfig: p.params.Config.Receiver, @@ -1723,6 +1762,7 @@ func (p *ParticipantImpl) addMigrateMutedTrack(cid string, t *livekit.TrackInfo) SdpCid: cid, ParticipantID: p.params.SID, ParticipantIdentity: p.params.Identity, + ParticipantVersion: p.version.Load(), RTCPChan: p.rtcpCh, BufferFactory: p.params.Config.BufferFactory, ReceiverConfig: p.params.Config.Receiver, @@ -2070,3 +2110,80 @@ func (p *ParticipantImpl) handleNegotiationFailed() { }) p.closeSignalConnection() } + +func (p *ParticipantImpl) EnqueueSubscribeTrack(trackID livekit.TrackID, f func(sub types.LocalParticipant) error) { + p.params.Logger.Infow("queueing subscribe", "trackID", trackID) + + p.lock.Lock() + p.subscriptionRequestsQueue[trackID] = append(p.subscriptionRequestsQueue[trackID], SubscribeRequest{ + requestType: SubscribeRequestTypeAdd, + addCb: f, + }) + p.lock.Unlock() + + go p.ProcessSubscriptionRequestsQueue(trackID) +} + +func (p *ParticipantImpl) EnqueueUnsubscribeTrack(trackID livekit.TrackID, willBeResumed bool, f func(subscriberID livekit.ParticipantID, willBeResumed bool) error) { + p.params.Logger.Infow("queueing unsubscribe", "trackID", trackID) + + p.lock.Lock() + p.subscriptionRequestsQueue[trackID] = append(p.subscriptionRequestsQueue[trackID], SubscribeRequest{ + requestType: SubscribeRequestTypeRemove, + willBeResumed: willBeResumed, + removeCb: f, + }) + p.lock.Unlock() + + go p.ProcessSubscriptionRequestsQueue(trackID) +} + +func (p *ParticipantImpl) ProcessSubscriptionRequestsQueue(trackID livekit.TrackID) { + p.lock.Lock() + if p.subscriptionInProgress[trackID] || len(p.subscriptionRequestsQueue[trackID]) == 0 { + p.lock.Unlock() + return + } + + request := p.subscriptionRequestsQueue[trackID][0] + p.subscriptionRequestsQueue[trackID] = p.subscriptionRequestsQueue[trackID][1:] + if len(p.subscriptionRequestsQueue[trackID]) == 0 { + delete(p.subscriptionRequestsQueue, trackID) + } + + p.subscriptionInProgress[trackID] = true + p.lock.Unlock() + + switch request.requestType { + case SubscribeRequestTypeAdd: + err := request.addCb(p) + if err != nil { + if err != errAlreadySubscribed { + p.params.Logger.Errorw("error adding subscriber", err, "trackID", trackID) + } + + // process pending request even if adding errors out + go p.ClearInProgressAndProcessSubscriptionRequestsQueue(trackID) + } + + case SubscribeRequestTypeRemove: + err := request.removeCb(p.ID(), request.willBeResumed) + if err != nil { + go p.ClearInProgressAndProcessSubscriptionRequestsQueue(trackID) + } + + default: + p.params.Logger.Warnw("unknown request type", nil) + + // let the queue move forward + go p.ClearInProgressAndProcessSubscriptionRequestsQueue(trackID) + } +} + +func (p *ParticipantImpl) ClearInProgressAndProcessSubscriptionRequestsQueue(trackID livekit.TrackID) { + p.lock.Lock() + delete(p.subscriptionInProgress, trackID) + p.lock.Unlock() + + p.ProcessSubscriptionRequestsQueue(trackID) +} diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 22a3b613d..1367a2931 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -21,6 +21,7 @@ const ( type SubscribedTrackParams struct { PublisherID livekit.ParticipantID PublisherIdentity livekit.ParticipantIdentity + PublisherVersion uint32 Subscriber types.LocalParticipant MediaTrack types.MediaTrack DownTrack *sfu.DownTrack @@ -80,6 +81,10 @@ func (t *SubscribedTrack) PublisherIdentity() livekit.ParticipantIdentity { return t.params.PublisherIdentity } +func (t *SubscribedTrack) PublisherVersion() uint32 { + return t.params.PublisherVersion +} + func (t *SubscribedTrack) SubscriberID() livekit.ParticipantID { return t.params.Subscriber.ID() } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 102858e38..eb68e1751 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -267,6 +267,13 @@ func (t *PCTransport) IsEstablished() bool { } func (t *PCTransport) Close() { + t.lock.Lock() + if t.signalStateCheckTimer != nil { + t.signalStateCheckTimer.Stop() + t.signalStateCheckTimer = nil + } + t.lock.Unlock() + if t.streamAllocator != nil { t.streamAllocator.Stop() } @@ -288,6 +295,7 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error { if t.signalStateCheckTimer != nil { t.signalStateCheckTimer.Stop() + t.signalStateCheckTimer = nil } for _, c := range t.pendingCandidates { @@ -434,6 +442,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { negotiateVersion := t.negotiateCounter.Inc() if t.signalStateCheckTimer != nil { t.signalStateCheckTimer.Stop() + t.signalStateCheckTimer = nil } t.signalStateCheckTimer = time.AfterFunc(negotiationFailedTimout, func() { t.lock.RLock() diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index a90b481ad..7fc3a7321 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -284,6 +284,11 @@ type LocalParticipant interface { CacheDownTrack(trackID livekit.TrackID, rtpTransceiver *webrtc.RTPTransceiver, forwarderState sfu.ForwarderState) UncacheDownTrack(rtpTransceiver *webrtc.RTPTransceiver) GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.RTPTransceiver, sfu.ForwarderState) + + EnqueueSubscribeTrack(trackID livekit.TrackID, f func(sub LocalParticipant) error) + EnqueueUnsubscribeTrack(trackID livekit.TrackID, willBeResumed bool, f func(subscriberID livekit.ParticipantID, willBeResumed bool) error) + ProcessSubscriptionRequestsQueue(trackID livekit.TrackID) + ClearInProgressAndProcessSubscriptionRequestsQueue(trackID livekit.TrackID) } // Room is a container of participants, and can provide room-level actions @@ -312,6 +317,7 @@ type MediaTrack interface { PublisherID() livekit.ParticipantID PublisherIdentity() livekit.ParticipantIdentity + PublisherVersion() uint32 IsMuted() bool SetMuted(muted bool) @@ -362,6 +368,7 @@ type SubscribedTrack interface { ID() livekit.TrackID PublisherID() livekit.ParticipantID PublisherIdentity() livekit.ParticipantIdentity + PublisherVersion() uint32 SubscriberID() livekit.ParticipantID SubscriberIdentity() livekit.ParticipantIdentity Subscriber() LocalParticipant diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index b36152d7d..04a1999de 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -184,6 +184,16 @@ type FakeLocalMediaTrack struct { publisherIdentityReturnsOnCall map[int]struct { result1 livekit.ParticipantIdentity } + PublisherVersionStub func() uint32 + publisherVersionMutex sync.RWMutex + publisherVersionArgsForCall []struct { + } + publisherVersionReturns struct { + result1 uint32 + } + publisherVersionReturnsOnCall map[int]struct { + result1 uint32 + } ReceiversStub func() []sfu.TrackReceiver receiversMutex sync.RWMutex receiversArgsForCall []struct { @@ -1203,6 +1213,59 @@ func (fake *FakeLocalMediaTrack) PublisherIdentityReturnsOnCall(i int, result1 l }{result1} } +func (fake *FakeLocalMediaTrack) PublisherVersion() uint32 { + fake.publisherVersionMutex.Lock() + ret, specificReturn := fake.publisherVersionReturnsOnCall[len(fake.publisherVersionArgsForCall)] + fake.publisherVersionArgsForCall = append(fake.publisherVersionArgsForCall, struct { + }{}) + stub := fake.PublisherVersionStub + fakeReturns := fake.publisherVersionReturns + fake.recordInvocation("PublisherVersion", []interface{}{}) + fake.publisherVersionMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalMediaTrack) PublisherVersionCallCount() int { + fake.publisherVersionMutex.RLock() + defer fake.publisherVersionMutex.RUnlock() + return len(fake.publisherVersionArgsForCall) +} + +func (fake *FakeLocalMediaTrack) PublisherVersionCalls(stub func() uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = stub +} + +func (fake *FakeLocalMediaTrack) PublisherVersionReturns(result1 uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = nil + fake.publisherVersionReturns = struct { + result1 uint32 + }{result1} +} + +func (fake *FakeLocalMediaTrack) PublisherVersionReturnsOnCall(i int, result1 uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = nil + if fake.publisherVersionReturnsOnCall == nil { + fake.publisherVersionReturnsOnCall = make(map[int]struct { + result1 uint32 + }) + } + fake.publisherVersionReturnsOnCall[i] = struct { + result1 uint32 + }{result1} +} + func (fake *FakeLocalMediaTrack) Receivers() []sfu.TrackReceiver { fake.receiversMutex.Lock() ret, specificReturn := fake.receiversReturnsOnCall[len(fake.receiversArgsForCall)] @@ -1710,6 +1773,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.publisherIDMutex.RUnlock() fake.publisherIdentityMutex.RLock() defer fake.publisherIdentityMutex.RUnlock() + fake.publisherVersionMutex.RLock() + defer fake.publisherVersionMutex.RUnlock() fake.receiversMutex.RLock() defer fake.receiversMutex.RUnlock() fake.removeAllSubscribersMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index cc0590016..7d2aa6d5a 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -98,6 +98,11 @@ type FakeLocalParticipant struct { claimGrantsReturnsOnCall map[int]struct { result1 *auth.ClaimGrants } + ClearInProgressAndProcessSubscriptionRequestsQueueStub func(livekit.TrackID) + clearInProgressAndProcessSubscriptionRequestsQueueMutex sync.RWMutex + clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall []struct { + arg1 livekit.TrackID + } CloseStub func(bool, types.ParticipantCloseReason) error closeMutex sync.RWMutex closeArgsForCall []struct { @@ -130,6 +135,19 @@ type FakeLocalParticipant struct { debugInfoReturnsOnCall map[int]struct { result1 map[string]interface{} } + EnqueueSubscribeTrackStub func(livekit.TrackID, func(sub types.LocalParticipant) error) + enqueueSubscribeTrackMutex sync.RWMutex + enqueueSubscribeTrackArgsForCall []struct { + arg1 livekit.TrackID + arg2 func(sub types.LocalParticipant) error + } + EnqueueUnsubscribeTrackStub func(livekit.TrackID, bool, func(subscriberID livekit.ParticipantID, willBeResumed bool) error) + enqueueUnsubscribeTrackMutex sync.RWMutex + enqueueUnsubscribeTrackArgsForCall []struct { + arg1 livekit.TrackID + arg2 bool + arg3 func(subscriberID livekit.ParticipantID, willBeResumed bool) error + } GetAdaptiveStreamStub func() bool getAdaptiveStreamMutex sync.RWMutex getAdaptiveStreamArgsForCall []struct { @@ -397,6 +415,11 @@ type FakeLocalParticipant struct { onTrackUpdatedArgsForCall []struct { arg1 func(types.LocalParticipant, types.MediaTrack) } + ProcessSubscriptionRequestsQueueStub func(livekit.TrackID) + processSubscriptionRequestsQueueMutex sync.RWMutex + processSubscriptionRequestsQueueArgsForCall []struct { + arg1 livekit.TrackID + } ProtocolVersionStub func() types.ProtocolVersion protocolVersionMutex sync.RWMutex protocolVersionArgsForCall []struct { @@ -1123,6 +1146,38 @@ func (fake *FakeLocalParticipant) ClaimGrantsReturnsOnCall(i int, result1 *auth. }{result1} } +func (fake *FakeLocalParticipant) ClearInProgressAndProcessSubscriptionRequestsQueue(arg1 livekit.TrackID) { + fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.Lock() + fake.clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall = append(fake.clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall, struct { + arg1 livekit.TrackID + }{arg1}) + stub := fake.ClearInProgressAndProcessSubscriptionRequestsQueueStub + fake.recordInvocation("ClearInProgressAndProcessSubscriptionRequestsQueue", []interface{}{arg1}) + fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.Unlock() + if stub != nil { + fake.ClearInProgressAndProcessSubscriptionRequestsQueueStub(arg1) + } +} + +func (fake *FakeLocalParticipant) ClearInProgressAndProcessSubscriptionRequestsQueueCallCount() int { + fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RLock() + defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RUnlock() + return len(fake.clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall) +} + +func (fake *FakeLocalParticipant) ClearInProgressAndProcessSubscriptionRequestsQueueCalls(stub func(livekit.TrackID)) { + fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.Lock() + defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.Unlock() + fake.ClearInProgressAndProcessSubscriptionRequestsQueueStub = stub +} + +func (fake *FakeLocalParticipant) ClearInProgressAndProcessSubscriptionRequestsQueueArgsForCall(i int) livekit.TrackID { + fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RLock() + defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RUnlock() + argsForCall := fake.clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) Close(arg1 bool, arg2 types.ParticipantCloseReason) error { fake.closeMutex.Lock() ret, specificReturn := fake.closeReturnsOnCall[len(fake.closeArgsForCall)] @@ -1291,6 +1346,73 @@ func (fake *FakeLocalParticipant) DebugInfoReturnsOnCall(i int, result1 map[stri }{result1} } +func (fake *FakeLocalParticipant) EnqueueSubscribeTrack(arg1 livekit.TrackID, arg2 func(sub types.LocalParticipant) error) { + fake.enqueueSubscribeTrackMutex.Lock() + fake.enqueueSubscribeTrackArgsForCall = append(fake.enqueueSubscribeTrackArgsForCall, struct { + arg1 livekit.TrackID + arg2 func(sub types.LocalParticipant) error + }{arg1, arg2}) + stub := fake.EnqueueSubscribeTrackStub + fake.recordInvocation("EnqueueSubscribeTrack", []interface{}{arg1, arg2}) + fake.enqueueSubscribeTrackMutex.Unlock() + if stub != nil { + fake.EnqueueSubscribeTrackStub(arg1, arg2) + } +} + +func (fake *FakeLocalParticipant) EnqueueSubscribeTrackCallCount() int { + fake.enqueueSubscribeTrackMutex.RLock() + defer fake.enqueueSubscribeTrackMutex.RUnlock() + return len(fake.enqueueSubscribeTrackArgsForCall) +} + +func (fake *FakeLocalParticipant) EnqueueSubscribeTrackCalls(stub func(livekit.TrackID, func(sub types.LocalParticipant) error)) { + fake.enqueueSubscribeTrackMutex.Lock() + defer fake.enqueueSubscribeTrackMutex.Unlock() + fake.EnqueueSubscribeTrackStub = stub +} + +func (fake *FakeLocalParticipant) EnqueueSubscribeTrackArgsForCall(i int) (livekit.TrackID, func(sub types.LocalParticipant) error) { + fake.enqueueSubscribeTrackMutex.RLock() + defer fake.enqueueSubscribeTrackMutex.RUnlock() + argsForCall := fake.enqueueSubscribeTrackArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeLocalParticipant) EnqueueUnsubscribeTrack(arg1 livekit.TrackID, arg2 bool, arg3 func(subscriberID livekit.ParticipantID, willBeResumed bool) error) { + fake.enqueueUnsubscribeTrackMutex.Lock() + fake.enqueueUnsubscribeTrackArgsForCall = append(fake.enqueueUnsubscribeTrackArgsForCall, struct { + arg1 livekit.TrackID + arg2 bool + arg3 func(subscriberID livekit.ParticipantID, willBeResumed bool) error + }{arg1, arg2, arg3}) + stub := fake.EnqueueUnsubscribeTrackStub + fake.recordInvocation("EnqueueUnsubscribeTrack", []interface{}{arg1, arg2, arg3}) + fake.enqueueUnsubscribeTrackMutex.Unlock() + if stub != nil { + fake.EnqueueUnsubscribeTrackStub(arg1, arg2, arg3) + } +} + +func (fake *FakeLocalParticipant) EnqueueUnsubscribeTrackCallCount() int { + fake.enqueueUnsubscribeTrackMutex.RLock() + defer fake.enqueueUnsubscribeTrackMutex.RUnlock() + return len(fake.enqueueUnsubscribeTrackArgsForCall) +} + +func (fake *FakeLocalParticipant) EnqueueUnsubscribeTrackCalls(stub func(livekit.TrackID, bool, func(subscriberID livekit.ParticipantID, willBeResumed bool) error)) { + fake.enqueueUnsubscribeTrackMutex.Lock() + defer fake.enqueueUnsubscribeTrackMutex.Unlock() + fake.EnqueueUnsubscribeTrackStub = stub +} + +func (fake *FakeLocalParticipant) EnqueueUnsubscribeTrackArgsForCall(i int) (livekit.TrackID, bool, func(subscriberID livekit.ParticipantID, willBeResumed bool) error) { + fake.enqueueUnsubscribeTrackMutex.RLock() + defer fake.enqueueUnsubscribeTrackMutex.RUnlock() + argsForCall := fake.enqueueUnsubscribeTrackArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 +} + func (fake *FakeLocalParticipant) GetAdaptiveStream() bool { fake.getAdaptiveStreamMutex.Lock() ret, specificReturn := fake.getAdaptiveStreamReturnsOnCall[len(fake.getAdaptiveStreamArgsForCall)] @@ -2749,6 +2871,38 @@ func (fake *FakeLocalParticipant) OnTrackUpdatedArgsForCall(i int) func(types.Lo return argsForCall.arg1 } +func (fake *FakeLocalParticipant) ProcessSubscriptionRequestsQueue(arg1 livekit.TrackID) { + fake.processSubscriptionRequestsQueueMutex.Lock() + fake.processSubscriptionRequestsQueueArgsForCall = append(fake.processSubscriptionRequestsQueueArgsForCall, struct { + arg1 livekit.TrackID + }{arg1}) + stub := fake.ProcessSubscriptionRequestsQueueStub + fake.recordInvocation("ProcessSubscriptionRequestsQueue", []interface{}{arg1}) + fake.processSubscriptionRequestsQueueMutex.Unlock() + if stub != nil { + fake.ProcessSubscriptionRequestsQueueStub(arg1) + } +} + +func (fake *FakeLocalParticipant) ProcessSubscriptionRequestsQueueCallCount() int { + fake.processSubscriptionRequestsQueueMutex.RLock() + defer fake.processSubscriptionRequestsQueueMutex.RUnlock() + return len(fake.processSubscriptionRequestsQueueArgsForCall) +} + +func (fake *FakeLocalParticipant) ProcessSubscriptionRequestsQueueCalls(stub func(livekit.TrackID)) { + fake.processSubscriptionRequestsQueueMutex.Lock() + defer fake.processSubscriptionRequestsQueueMutex.Unlock() + fake.ProcessSubscriptionRequestsQueueStub = stub +} + +func (fake *FakeLocalParticipant) ProcessSubscriptionRequestsQueueArgsForCall(i int) livekit.TrackID { + fake.processSubscriptionRequestsQueueMutex.RLock() + defer fake.processSubscriptionRequestsQueueMutex.RUnlock() + argsForCall := fake.processSubscriptionRequestsQueueArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) ProtocolVersion() types.ProtocolVersion { fake.protocolVersionMutex.Lock() ret, specificReturn := fake.protocolVersionReturnsOnCall[len(fake.protocolVersionArgsForCall)] @@ -4331,12 +4485,18 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.canSubscribeMutex.RUnlock() fake.claimGrantsMutex.RLock() defer fake.claimGrantsMutex.RUnlock() + fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RLock() + defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RUnlock() fake.closeMutex.RLock() defer fake.closeMutex.RUnlock() fake.connectedAtMutex.RLock() defer fake.connectedAtMutex.RUnlock() fake.debugInfoMutex.RLock() defer fake.debugInfoMutex.RUnlock() + fake.enqueueSubscribeTrackMutex.RLock() + defer fake.enqueueSubscribeTrackMutex.RUnlock() + fake.enqueueUnsubscribeTrackMutex.RLock() + defer fake.enqueueUnsubscribeTrackMutex.RUnlock() fake.getAdaptiveStreamMutex.RLock() defer fake.getAdaptiveStreamMutex.RUnlock() fake.getAudioLevelMutex.RLock() @@ -4397,6 +4557,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.onTrackPublishedMutex.RUnlock() fake.onTrackUpdatedMutex.RLock() defer fake.onTrackUpdatedMutex.RUnlock() + fake.processSubscriptionRequestsQueueMutex.RLock() + defer fake.processSubscriptionRequestsQueueMutex.RUnlock() fake.protocolVersionMutex.RLock() defer fake.protocolVersionMutex.RUnlock() fake.removeSubscribedTrackMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index a3b2c3d87..2593f2786 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -151,6 +151,16 @@ type FakeMediaTrack struct { publisherIdentityReturnsOnCall map[int]struct { result1 livekit.ParticipantIdentity } + PublisherVersionStub func() uint32 + publisherVersionMutex sync.RWMutex + publisherVersionArgsForCall []struct { + } + publisherVersionReturns struct { + result1 uint32 + } + publisherVersionReturnsOnCall map[int]struct { + result1 uint32 + } ReceiversStub func() []sfu.TrackReceiver receiversMutex sync.RWMutex receiversArgsForCall []struct { @@ -985,6 +995,59 @@ func (fake *FakeMediaTrack) PublisherIdentityReturnsOnCall(i int, result1 liveki }{result1} } +func (fake *FakeMediaTrack) PublisherVersion() uint32 { + fake.publisherVersionMutex.Lock() + ret, specificReturn := fake.publisherVersionReturnsOnCall[len(fake.publisherVersionArgsForCall)] + fake.publisherVersionArgsForCall = append(fake.publisherVersionArgsForCall, struct { + }{}) + stub := fake.PublisherVersionStub + fakeReturns := fake.publisherVersionReturns + fake.recordInvocation("PublisherVersion", []interface{}{}) + fake.publisherVersionMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) PublisherVersionCallCount() int { + fake.publisherVersionMutex.RLock() + defer fake.publisherVersionMutex.RUnlock() + return len(fake.publisherVersionArgsForCall) +} + +func (fake *FakeMediaTrack) PublisherVersionCalls(stub func() uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = stub +} + +func (fake *FakeMediaTrack) PublisherVersionReturns(result1 uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = nil + fake.publisherVersionReturns = struct { + result1 uint32 + }{result1} +} + +func (fake *FakeMediaTrack) PublisherVersionReturnsOnCall(i int, result1 uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = nil + if fake.publisherVersionReturnsOnCall == nil { + fake.publisherVersionReturnsOnCall = make(map[int]struct { + result1 uint32 + }) + } + fake.publisherVersionReturnsOnCall[i] = struct { + result1 uint32 + }{result1} +} + func (fake *FakeMediaTrack) Receivers() []sfu.TrackReceiver { fake.receiversMutex.Lock() ret, specificReturn := fake.receiversReturnsOnCall[len(fake.receiversArgsForCall)] @@ -1401,6 +1464,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.publisherIDMutex.RUnlock() fake.publisherIdentityMutex.RLock() defer fake.publisherIdentityMutex.RUnlock() + fake.publisherVersionMutex.RLock() + defer fake.publisherVersionMutex.RUnlock() fake.receiversMutex.RLock() defer fake.receiversMutex.RUnlock() fake.removeAllSubscribersMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_subscribed_track.go b/pkg/rtc/types/typesfakes/fake_subscribed_track.go index 2d0dc8d5e..db72d6798 100644 --- a/pkg/rtc/types/typesfakes/fake_subscribed_track.go +++ b/pkg/rtc/types/typesfakes/fake_subscribed_track.go @@ -75,6 +75,16 @@ type FakeSubscribedTrack struct { publisherIdentityReturnsOnCall map[int]struct { result1 livekit.ParticipantIdentity } + PublisherVersionStub func() uint32 + publisherVersionMutex sync.RWMutex + publisherVersionArgsForCall []struct { + } + publisherVersionReturns struct { + result1 uint32 + } + publisherVersionReturnsOnCall map[int]struct { + result1 uint32 + } SetPublisherMutedStub func(bool) setPublisherMutedMutex sync.RWMutex setPublisherMutedArgsForCall []struct { @@ -473,6 +483,59 @@ func (fake *FakeSubscribedTrack) PublisherIdentityReturnsOnCall(i int, result1 l }{result1} } +func (fake *FakeSubscribedTrack) PublisherVersion() uint32 { + fake.publisherVersionMutex.Lock() + ret, specificReturn := fake.publisherVersionReturnsOnCall[len(fake.publisherVersionArgsForCall)] + fake.publisherVersionArgsForCall = append(fake.publisherVersionArgsForCall, struct { + }{}) + stub := fake.PublisherVersionStub + fakeReturns := fake.publisherVersionReturns + fake.recordInvocation("PublisherVersion", []interface{}{}) + fake.publisherVersionMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSubscribedTrack) PublisherVersionCallCount() int { + fake.publisherVersionMutex.RLock() + defer fake.publisherVersionMutex.RUnlock() + return len(fake.publisherVersionArgsForCall) +} + +func (fake *FakeSubscribedTrack) PublisherVersionCalls(stub func() uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = stub +} + +func (fake *FakeSubscribedTrack) PublisherVersionReturns(result1 uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = nil + fake.publisherVersionReturns = struct { + result1 uint32 + }{result1} +} + +func (fake *FakeSubscribedTrack) PublisherVersionReturnsOnCall(i int, result1 uint32) { + fake.publisherVersionMutex.Lock() + defer fake.publisherVersionMutex.Unlock() + fake.PublisherVersionStub = nil + if fake.publisherVersionReturnsOnCall == nil { + fake.publisherVersionReturnsOnCall = make(map[int]struct { + result1 uint32 + }) + } + fake.publisherVersionReturnsOnCall[i] = struct { + result1 uint32 + }{result1} +} + func (fake *FakeSubscribedTrack) SetPublisherMuted(arg1 bool) { fake.setPublisherMutedMutex.Lock() fake.setPublisherMutedArgsForCall = append(fake.setPublisherMutedArgsForCall, struct { @@ -737,6 +800,8 @@ func (fake *FakeSubscribedTrack) Invocations() map[string][][]interface{} { defer fake.publisherIDMutex.RUnlock() fake.publisherIdentityMutex.RLock() defer fake.publisherIdentityMutex.RUnlock() + fake.publisherVersionMutex.RLock() + defer fake.publisherVersionMutex.RUnlock() fake.setPublisherMutedMutex.RLock() defer fake.setPublisherMutedMutex.RUnlock() fake.subscriberMutex.RLock()