diff --git a/go.mod b/go.mod index 4f80fdf53..79609685d 100644 --- a/go.mod +++ b/go.mod @@ -14,7 +14,7 @@ require ( github.com/google/wire v0.5.0 github.com/gorilla/websocket v1.4.2 github.com/hashicorp/golang-lru v0.5.4 - github.com/livekit/protocol v0.11.8-0.20220103045453-c441eb5f03c8 + github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 diff --git a/go.sum b/go.sum index a33092c72..6499baf25 100644 --- a/go.sum +++ b/go.sum @@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA= github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= -github.com/livekit/protocol v0.11.8-0.20220103045453-c441eb5f03c8 h1:eo4OOUKLgNguaHEmcItTfk2IzVC2s2r6/WXqFfj5HjE= -github.com/livekit/protocol v0.11.8-0.20220103045453-c441eb5f03c8/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= +github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed h1:6vxJ62pwuhXtEjqvsANTIoEcTgHR9laMa9tR3Xr0fAM= +github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls= github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0= diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index c9ac7491f..883c78f4a 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -8,6 +8,7 @@ import ( "sync/atomic" "time" + "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "github.com/livekit/protocol/livekit" @@ -15,10 +16,8 @@ import ( "github.com/livekit/protocol/utils" "github.com/pion/rtcp" "github.com/pion/webrtc/v3" - "github.com/pion/webrtc/v3/pkg/rtcerr" "github.com/livekit/livekit-server/pkg/config" - "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/twcc" @@ -29,7 +28,6 @@ const ( lostUpdateDelta = time.Second connectionQualityUpdateInterval = 5 * time.Second layerSelectionTolerance = 0.9 - initialQualityUpdateWait = 10 * time.Second ) // MediaTrack represents a WebRTC track that needs to be forwarded @@ -46,13 +44,13 @@ type MediaTrack struct { lock sync.RWMutex - // map of target participantID -> types.SubscribedTrack - subscribedTracks sync.Map // participantID => types.SubscribedTrack - twcc *twcc.Responder - audioLevel *AudioLevel - receiver sfu.Receiver - lastPLI time.Time - layerDimensions sync.Map // quality => *livekit.VideoLayer + twcc *twcc.Responder + + audioLevelMu sync.RWMutex + audioLevel *AudioLevel + + receiver sfu.Receiver + layerDimensions sync.Map // livekit.VideoQuality => *livekit.VideoLayer // track audio fraction lost statsLock sync.Mutex @@ -65,15 +63,9 @@ type MediaTrack struct { done chan struct{} - // quality level enable/disable - maxQualityLock sync.RWMutex - maxSubscriberQuality map[livekit.ParticipantID]livekit.VideoQuality - maxSubscribedQuality livekit.VideoQuality - allSubscribersMuted bool - onSubscribedMaxQualityChange func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality) error - maxQualityTimer *time.Timer - onClose []func() + + *MediaTrackSubscriptions } type MediaTrackParams struct { @@ -95,15 +87,23 @@ type MediaTrackParams struct { func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTrack { t := &MediaTrack{ - params: params, - ssrc: track.SSRC(), - streamID: track.StreamID(), - codec: track.Codec(), - connectionStats: connectionquality.NewConnectionStats(), - done: make(chan struct{}), - maxSubscriberQuality: make(map[livekit.ParticipantID]livekit.VideoQuality), + params: params, + ssrc: track.SSRC(), + streamID: track.StreamID(), + codec: track.Codec(), + connectionStats: connectionquality.NewConnectionStats(), + done: make(chan struct{}), } + t.MediaTrackSubscriptions = NewMediaTrackSubscriptions(MediaTrackSubscriptionsParams{ + MediaTrack: t, + BufferFactory: params.BufferFactory, + ReceiverConfig: params.ReceiverConfig, + SubscriberConfig: params.SubscriberConfig, + Telemetry: params.Telemetry, + Logger: ¶ms.Logger, + }) + if params.TrackInfo.Muted { t.SetMuted(true) } @@ -139,6 +139,14 @@ func (t *MediaTrack) Source() livekit.TrackSource { return t.params.TrackInfo.Source } +func (t *MediaTrack) PublisherID() livekit.ParticipantID { + return t.params.ParticipantID +} + +func (t *MediaTrack) PublisherIdentity() livekit.ParticipantIdentity { + return t.params.ParticipantIdentity +} + func (t *MediaTrack) IsSimulcast() bool { return t.simulcasted.Get() } @@ -160,18 +168,7 @@ func (t *MediaTrack) SetMuted(muted bool) { } t.lock.RUnlock() - // mute all subscribed tracks - t.subscribedTracks.Range(func(_, value interface{}) bool { - if st, ok := value.(types.SubscribedTrack); ok { - st.SetPublisherMuted(muted) - } - return true - }) - - // update quality based on subscription if unmuting - if !muted { - t.updateQualityChange() - } + t.MediaTrackSubscriptions.SetMuted(muted) } func (t *MediaTrack) AddOnClose(f func()) { @@ -181,11 +178,6 @@ func (t *MediaTrack) AddOnClose(f func()) { t.onClose = append(t.onClose, f) } -func (t *MediaTrack) IsSubscriber(subID livekit.ParticipantID) bool { - _, ok := t.subscribedTracks.Load(subID) - return ok -} - func (t *MediaTrack) PublishLossPercentage() uint32 { return FixedPointToPercent(uint8(atomic.LoadUint32(&t.currentUpFracLost))) } @@ -195,19 +187,11 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error { t.lock.Lock() defer t.lock.Unlock() - subscriberID := sub.ID() - - // don't subscribe to the same track multiple times - if _, ok := t.subscribedTracks.Load(subscriberID); ok { - return nil - } - if t.receiver == nil { // cannot add, no receiver return errors.New("cannot subscribe without a receiver in place") } - codec := t.receiver.Codec() // using DownTrack from ion-sfu streamId := string(t.params.ParticipantID) if sub.ProtocolVersion().SupportsPackedStreamId() { @@ -216,163 +200,27 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error { streamId = PackStreamID(t.params.ParticipantID, t.ID()) } - receiver := NewWrappedReceiver(t.receiver, t.ID(), streamId) - - var rtcpFeedback []webrtc.RTCPFeedback - switch t.Kind() { - case livekit.TrackType_AUDIO: - rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Audio - case livekit.TrackType_VIDEO: - rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Video - } - downTrack, err := sfu.NewDownTrack(webrtc.RTPCodecCapability{ - MimeType: codec.MimeType, - ClockRate: codec.ClockRate, - Channels: codec.Channels, - SDPFmtpLine: codec.SDPFmtpLine, - RTCPFeedback: rtcpFeedback, - }, receiver, t.params.BufferFactory, subscriberID, t.params.ReceiverConfig.PacketBufferSize) + downTrack, err := t.MediaTrackSubscriptions.AddSubscriber(sub, t.receiver.Codec(), NewWrappedReceiver(t.receiver, t.ID(), streamId)) if err != nil { return err } - subTrack := NewSubscribedTrack(SubscribedTrackParams{ - PublisherID: t.params.ParticipantID, - PublisherIdentity: t.params.ParticipantIdentity, - SubscriberID: subscriberID, - MediaTrack: t, - DownTrack: downTrack, - }) - var transceiver *webrtc.RTPTransceiver - var sender *webrtc.RTPSender - if sub.ProtocolVersion().SupportsTransceiverReuse() { - // - // AddTrack will create a new transceiver or re-use an unused one - // if the attributes match. This prevents SDP from bloating - // because of dormant transceivers building up. - // - sender, err = sub.SubscriberPC().AddTrack(downTrack) - if err != nil { - return err + if downTrack != nil { + if t.Kind() == livekit.TrackType_AUDIO { + downTrack.AddReceiverReportListener(t.handleMaxLossFeedback) } - // as there is no way to get transceiver from sender, search - for _, tr := range sub.SubscriberPC().GetTransceivers() { - if tr.Sender() == sender { - transceiver = tr - break - } - } - if transceiver == nil { - // cannot add, no transceiver - return errors.New("cannot subscribe without a transceiver in place") - } - } else { - transceiver, err = sub.SubscriberPC().AddTransceiverFromTrack(downTrack, webrtc.RTPTransceiverInit{ - Direction: webrtc.RTPTransceiverDirectionSendonly, - }) - if err != nil { - return err - } - - sender = transceiver.Sender() - if sender == nil { - // cannot add, no sender - return errors.New("cannot subscribe without a sender in place") - } + t.receiver.AddDownTrack(downTrack) } - - sendParameters := sender.GetParameters() - downTrack.SetRTPHeaderExtensions(sendParameters.HeaderExtensions) - - downTrack.SetTransceiver(transceiver) - // when outtrack is bound, start loop to send reports - downTrack.OnBind(func() { - go subTrack.Bound() - go t.sendDownTrackBindingReports(sub) - }) - downTrack.OnPacketSent(func(_ *sfu.DownTrack, size int) { - t.params.Telemetry.OnDownstreamPacket(subscriberID, t.ID(), size) - }) - downTrack.OnPaddingSent(func(_ *sfu.DownTrack, size int) { - t.params.Telemetry.OnDownstreamPacket(subscriberID, t.ID(), size) - }) - downTrack.OnRTCP(func(pkts []rtcp.Packet) { - t.params.Telemetry.HandleRTCP(livekit.StreamType_DOWNSTREAM, subscriberID, t.ID(), pkts) - }) - - downTrack.OnCloseHandler(func() { - go func() { - t.subscribedTracks.Delete(subscriberID) - t.params.Telemetry.TrackUnsubscribed(context.Background(), subscriberID, t.ToProto()) - - // ignore if the subscribing sub is not connected - if sub.SubscriberPC().ConnectionState() == webrtc.PeerConnectionStateClosed { - return - } - - // if the source has been terminated, we'll need to terminate all of the subscribedtracks - // however, if the dest sub has disconnected, then we can skip - if sender == nil { - return - } - t.params.Logger.Debugw("removing peerconnection track", - "track", t.ID(), - "subscriber", sub.Identity(), - "subscriberID", subscriberID, - "kind", t.Kind(), - ) - if err := sub.SubscriberPC().RemoveTrack(sender); err != nil { - if err == webrtc.ErrConnectionClosed { - // sub closing, can skip removing subscribedtracks - return - } - if _, ok := err.(*rtcerr.InvalidStateError); !ok { - // most of these are safe to ignore, since the track state might have already - // been set to Inactive - t.params.Logger.Debugw("could not remove remoteTrack from forwarder", - "error", err, - "subscriber", sub.Identity(), - "subscriberID", subscriberID, - ) - } - } - - t.NotifySubscriberMute(subscriberID) - sub.RemoveSubscribedTrack(subTrack) - sub.Negotiate() - }() - }) - if t.Kind() == livekit.TrackType_AUDIO { - downTrack.AddReceiverReportListener(t.handleMaxLossFeedback) - } - - t.subscribedTracks.Store(subscriberID, subTrack) - subTrack.SetPublisherMuted(t.IsMuted()) - - t.receiver.AddDownTrack(downTrack) - // since sub will lock, run it in a goroutine to avoid deadlocks - go func() { - t.NotifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_HIGH) // start with HIGH, let subscription change it later - sub.AddSubscribedTrack(subTrack) - sub.Negotiate() - }() - - t.params.Telemetry.TrackSubscribed(context.Background(), subscriberID, t.ToProto()) return nil } func (t *MediaTrack) NumUpTracks() (uint32, uint32) { numExpected := atomic.LoadUint32(&t.numUpTracks) - t.maxQualityLock.RLock() - maxSubscribed := uint32(0) - if !t.allSubscribersMuted { - maxSubscribed = uint32(SpatialLayerForQuality(t.maxSubscribedQuality) + 1) - } - t.maxQualityLock.RUnlock() - if maxSubscribed < numExpected { - numExpected = maxSubscribed + numSubscribedLayers := t.numSubscribedLayers() + if numSubscribedLayers < numExpected { + numExpected = numSubscribedLayers } t.lock.RLock() @@ -399,10 +247,15 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra buff.OnFeedback(t.handlePublisherFeedback) if t.Kind() == livekit.TrackType_AUDIO { + t.audioLevelMu.Lock() t.audioLevel = NewAudioLevel(t.params.AudioConfig.ActiveLevel, t.params.AudioConfig.MinPercentile) buff.OnAudioLevel(func(level uint8, duration uint32) { + t.audioLevelMu.RLock() + defer t.audioLevelMu.RUnlock() + t.audioLevel.Observe(level, duration) }) + t.audioLevelMu.Unlock() } else if t.Kind() == livekit.TrackType_VIDEO { if twcc != nil { buff.OnTransportWideCC(func(sn uint16, timeNS int64, marker bool) { @@ -471,57 +324,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra }) } -// RemoveSubscriber removes participant from subscription -// stop all forwarders to the client -func (t *MediaTrack) RemoveSubscriber(participantID livekit.ParticipantID) { - subTrack := t.getSubscribedTrack(participantID) - if subTrack != nil { - go subTrack.DownTrack().Close() - } -} - -func (t *MediaTrack) RemoveAllSubscribers() { - t.params.Logger.Debugw("removing all subscribers", "track", t.ID()) - t.lock.Lock() - defer t.lock.Unlock() - t.subscribedTracks.Range(func(_, val interface{}) bool { - if subTrack, ok := val.(types.SubscribedTrack); ok { - go subTrack.DownTrack().Close() - } - return true - }) - t.subscribedTracks = sync.Map{} -} - -func (t *MediaTrack) RevokeDisallowedSubscribers(allowedSubscriberIDs []livekit.ParticipantID) []livekit.ParticipantID { - t.lock.Lock() - defer t.lock.Unlock() - - var revokedSubscriberIDs []livekit.ParticipantID - // LK-TODO: large number of subscribers needs to be solved for this loop - t.subscribedTracks.Range(func(key interface{}, val interface{}) bool { - if subID, ok := key.(livekit.ParticipantID); ok { - found := false - for _, allowedID := range allowedSubscriberIDs { - if subID == allowedID { - found = true - break - } - } - - if !found { - if subTrack, ok := val.(types.SubscribedTrack); ok { - go subTrack.DownTrack().Close() - revokedSubscriberIDs = append(revokedSubscriberIDs, subID) - } - } - } - return true - }) - - return revokedSubscriberIDs -} - func (t *MediaTrack) ToProto() *livekit.TrackInfo { info := t.params.TrackInfo info.Muted = t.IsMuted() @@ -538,16 +340,23 @@ func (t *MediaTrack) ToProto() *livekit.TrackInfo { return info } +func (t *MediaTrack) GetAudioLevel() (level uint8, active bool) { + t.audioLevelMu.RLock() + defer t.audioLevelMu.RUnlock() + + if t.audioLevel == nil { + return silentAudioLevel, false + } + return t.audioLevel.GetLevel() +} + func (t *MediaTrack) UpdateVideoLayers(layers []*livekit.VideoLayer) { for _, layer := range layers { t.layerDimensions.Store(layer.Quality, layer) } - t.subscribedTracks.Range(func(_, val interface{}) bool { - if st, ok := val.(types.SubscribedTrack); ok { - st.UpdateVideoLayer() - } - return true - }) + + t.MediaTrackSubscriptions.UpdateVideoLayers() + // TODO: this might need to trigger a participant update for clients to pick up dimension change } @@ -596,53 +405,6 @@ func (t *MediaTrack) GetQualityForDimension(width, height uint32) livekit.VideoQ return quality } -func (t *MediaTrack) getSubscribedTrack(subscriberID livekit.ParticipantID) types.SubscribedTrack { - if val, ok := t.subscribedTracks.Load(subscriberID); ok { - if st, ok := val.(types.SubscribedTrack); ok { - return st - } - } - return nil -} - -// TODO: send for all downtracks from the source participant -// https://tools.ietf.org/html/rfc7941 -func (t *MediaTrack) sendDownTrackBindingReports(sub types.Participant) { - var sd []rtcp.SourceDescriptionChunk - - subTrack := t.getSubscribedTrack(sub.ID()) - if subTrack == nil { - return - } - - chunks := subTrack.DownTrack().CreateSourceDescriptionChunks() - if chunks == nil { - return - } - sd = append(sd, chunks...) - - pkts := []rtcp.Packet{ - &rtcp.SourceDescription{Chunks: sd}, - } - - go func() { - defer RecoverSilent() - batch := pkts - i := 0 - for { - if err := sub.SubscriberPC().WriteRTCP(batch); err != nil { - t.params.Logger.Errorw("could not write RTCP", err) - return - } - if i > 5 { - return - } - i++ - time.Sleep(20 * time.Millisecond) - } - }() -} - func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) { var maxLost uint8 var hasReport bool @@ -714,17 +476,34 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) { // handles max loss for audio packets func (t *MediaTrack) handleMaxLossFeedback(_ *sfu.DownTrack, report *rtcp.ReceiverReport) { - var ( - shouldUpdate bool - maxLost uint8 - ) t.statsLock.Lock() for _, rr := range report.Reports { if t.maxDownFracLost < rr.FractionLost { t.maxDownFracLost = rr.FractionLost } } + t.statsLock.Unlock() + t.maybeUpdateLoss() +} + +func (t *MediaTrack) NotifySubscriberNodeMediaLoss(_nodeID string, fractionalLoss uint8) { + t.statsLock.Lock() + if t.maxDownFracLost < fractionalLoss { + t.maxDownFracLost = fractionalLoss + } + t.statsLock.Unlock() + + t.maybeUpdateLoss() +} + +func (t *MediaTrack) maybeUpdateLoss() { + var ( + shouldUpdate bool + maxLost uint8 + ) + + t.statsLock.Lock() now := time.Now() if now.Sub(t.maxDownFracLostTs) > lostUpdateDelta { shouldUpdate = true @@ -748,24 +527,16 @@ func (t *MediaTrack) DebugInfo() map[string]interface{} { "PubMuted": t.muted.Get(), } - subscribedTrackInfo := make([]map[string]interface{}, 0) - t.subscribedTracks.Range(func(_, val interface{}) bool { - if track, ok := val.(*SubscribedTrack); ok { - dt := track.DownTrack().DebugInfo() - dt["PubMuted"] = track.pubMuted.Get() - dt["SubMuted"] = track.subMuted.Get() - subscribedTrackInfo = append(subscribedTrackInfo, dt) - } - return true - }) - info["DownTracks"] = subscribedTrackInfo + info["DownTracks"] = t.MediaTrackSubscriptions.DebugInfo() + t.lock.RLock() if t.receiver != nil { receiverInfo := t.receiver.DebugInfo() for k, v := range receiverInfo { info[k] = v } } + t.lock.RUnlock() return info } @@ -816,132 +587,17 @@ func (t *MediaTrack) calculateVideoScore() { } func (t *MediaTrack) OnSubscribedMaxQualityChange(f func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality) error) { - t.onSubscribedMaxQualityChange = f -} - -func (t *MediaTrack) NotifySubscriberMute(subscriberID livekit.ParticipantID) { - if t.Kind() != livekit.TrackType_VIDEO { - return - } - - t.maxQualityLock.Lock() - _, ok := t.maxSubscriberQuality[subscriberID] - if !ok { - t.maxQualityLock.Unlock() - return - } - - delete(t.maxSubscriberQuality, subscriberID) - t.maxQualityLock.Unlock() - - t.updateQualityChange() -} - -func (t *MediaTrack) NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) { - if t.Kind() != livekit.TrackType_VIDEO { - return - } - - t.maxQualityLock.Lock() - maxQuality, ok := t.maxSubscriberQuality[subscriberID] - if ok && maxQuality == quality { - t.maxQualityLock.Unlock() - return - } - - t.maxSubscriberQuality[subscriberID] = quality - t.maxQualityLock.Unlock() - - t.updateQualityChange() -} - -func (t *MediaTrack) startMaxQualityTimer() { - t.maxQualityLock.Lock() - defer t.maxQualityLock.Unlock() - - if t.Kind() != livekit.TrackType_VIDEO { - return - } - - t.maxQualityTimer = time.AfterFunc(initialQualityUpdateWait, func() { - t.stopMaxQualityTimer() - t.updateQualityChange() - }) -} - -func (t *MediaTrack) stopMaxQualityTimer() { - t.maxQualityLock.Lock() - defer t.maxQualityLock.Unlock() - - if t.maxQualityTimer != nil { - t.maxQualityTimer.Stop() - t.maxQualityTimer = nil - } -} - -func (t *MediaTrack) updateQualityChange() { - if t.Kind() != livekit.TrackType_VIDEO || t.IsMuted() { - return - } - - var subscribedQualities []*livekit.SubscribedQuality - - t.maxQualityLock.Lock() - allSubscribersMuted := false - maxSubscribedQuality := livekit.VideoQuality_LOW - if len(t.maxSubscriberQuality) == 0 { - allSubscribersMuted = true - } else { - for _, subQuality := range t.maxSubscriberQuality { - if subQuality > maxSubscribedQuality { - maxSubscribedQuality = subQuality - } + t.MediaTrackSubscriptions.OnSubscribedMaxQualityChange(func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) { + if f != nil && !t.IsMuted() { + _ = f(t.ID(), subscribedQualities) } - } - notifyMaxExpected := false - maxExpectedSpatialLayer := int32(-1) - if allSubscribersMuted { - if !t.allSubscribersMuted { - notifyMaxExpected = true - maxExpectedSpatialLayer = sfu.InvalidLayerSpatial - - t.allSubscribersMuted = true - - subscribedQualities = []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: false}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, - {Quality: livekit.VideoQuality_HIGH, Enabled: false}, - } - } - } else { - if t.allSubscribersMuted || maxSubscribedQuality != t.maxSubscribedQuality { - t.allSubscribersMuted = false - notifyMaxExpected = true - maxExpectedSpatialLayer = SpatialLayerForQuality(maxSubscribedQuality) - t.maxSubscribedQuality = maxSubscribedQuality - - for q := livekit.VideoQuality_LOW; q <= livekit.VideoQuality_HIGH; q++ { - subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{ - Quality: q, - Enabled: q <= t.maxSubscribedQuality, - }) - } - } - } - t.maxQualityLock.Unlock() - - if notifyMaxExpected { t.lock.RLock() if t.receiver != nil { - t.receiver.SetMaxExpectedSpatialLayer(maxExpectedSpatialLayer) + t.receiver.SetMaxExpectedSpatialLayer(SpatialLayerForQuality(maxSubscribedQuality)) } t.lock.RUnlock() - } - - if len(subscribedQualities) != 0 && t.onSubscribedMaxQualityChange != nil { - _ = t.onSubscribedMaxQualityChange(t.ID(), subscribedQualities) - } + }) } //--------------------------- @@ -952,7 +608,11 @@ func SpatialLayerForQuality(quality livekit.VideoQuality) int32 { return 0 case livekit.VideoQuality_MEDIUM: return 1 - default: + case livekit.VideoQuality_HIGH: return 2 + case livekit.VideoQuality_OFF: + return -1 + default: + return -1 } } diff --git a/pkg/rtc/mediatrack_test.go b/pkg/rtc/mediatrack_test.go index cf3daa6c7..f0265ef5e 100644 --- a/pkg/rtc/mediatrack_test.go +++ b/pkg/rtc/mediatrack_test.go @@ -137,7 +137,7 @@ func TestSubscribedMaxQuality(t *testing.T) { }) // mute all subscribers - mt.NotifySubscriberMute("s1") + mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF) expectedSubscribedQualities := []*livekit.SubscribedQuality{ &livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false}, @@ -216,7 +216,7 @@ func TestSubscribedMaxQuality(t *testing.T) { require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) // muting "s2" only should not disable all qualities - mt.NotifySubscriberMute("s2") + mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_OFF) expectedSubscribedQualities = []*livekit.SubscribedQuality{ &livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true}, @@ -227,7 +227,7 @@ func TestSubscribedMaxQuality(t *testing.T) { require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) // muting "s1" also should disable all qualities - mt.NotifySubscriberMute("s1") + mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF) expectedSubscribedQualities = []*livekit.SubscribedQuality{ &livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false}, diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go new file mode 100644 index 000000000..e159ea394 --- /dev/null +++ b/pkg/rtc/mediatracksubscriptions.go @@ -0,0 +1,543 @@ +package rtc + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/pion/rtcp" + "github.com/pion/webrtc/v3" + "github.com/pion/webrtc/v3/pkg/rtcerr" + + "github.com/livekit/livekit-server/pkg/rtc/types" + "github.com/livekit/livekit-server/pkg/sfu" + "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/livekit-server/pkg/telemetry" +) + +const ( + initialQualityUpdateWait = 10 * time.Second +) + +// MediaTrackSubscriptions manages subscriptions of a media track +type MediaTrackSubscriptions struct { + params MediaTrackSubscriptionsParams + + subscribedTracksMu sync.RWMutex + subscribedTracks map[livekit.ParticipantID]types.SubscribedTrack // participantID => types.SubscribedTrack + + onNoSubscribers func() + + // quality level enable/disable + maxQualityLock sync.RWMutex + maxSubscriberQuality map[livekit.ParticipantID]livekit.VideoQuality + maxSubscriberNodeQuality map[string]livekit.VideoQuality // nodeID => livekit.VideoQuality + maxSubscribedQuality livekit.VideoQuality + onSubscribedMaxQualityChange func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) + maxQualityTimer *time.Timer +} + +type MediaTrackSubscriptionsParams struct { + MediaTrack types.MediaTrack + + BufferFactory *buffer.Factory + ReceiverConfig ReceiverConfig + SubscriberConfig DirectionConfig + + Telemetry telemetry.TelemetryService + + Logger *logger.Logger +} + +func NewMediaTrackSubscriptions(params MediaTrackSubscriptionsParams) *MediaTrackSubscriptions { + t := &MediaTrackSubscriptions{ + params: params, + subscribedTracks: make(map[livekit.ParticipantID]types.SubscribedTrack), + maxSubscriberQuality: make(map[livekit.ParticipantID]livekit.VideoQuality), + maxSubscriberNodeQuality: make(map[string]livekit.VideoQuality), + } + + return t +} + +func (t *MediaTrackSubscriptions) OnNoSubscribers(f func()) { + t.onNoSubscribers = f +} + +func (t *MediaTrackSubscriptions) SetMuted(muted bool) { + t.subscribedTracksMu.RLock() + subscribedTracks := t.subscribedTracks + t.subscribedTracksMu.RUnlock() + + // mute all subscribed tracks + for _, st := range subscribedTracks { + st.SetPublisherMuted(muted) + } + + // update quality based on subscription if unmuting + if !muted { + t.updateQualityChange() + } +} + +func (t *MediaTrackSubscriptions) IsSubscriber(subID livekit.ParticipantID) bool { + t.subscribedTracksMu.RLock() + defer t.subscribedTracksMu.RUnlock() + + _, ok := t.subscribedTracks[subID] + return ok +} + +// AddSubscriber subscribes sub to current mediaTrack +func (t *MediaTrackSubscriptions) AddSubscriber(sub types.Participant, codec webrtc.RTPCodecCapability, wr WrappedReceiver) (*sfu.DownTrack, error) { + subscriberID := sub.ID() + + t.subscribedTracksMu.Lock() + defer t.subscribedTracksMu.Unlock() + + // don't subscribe to the same track multiple times + if _, ok := t.subscribedTracks[subscriberID]; ok { + return nil, nil + } + + var rtcpFeedback []webrtc.RTCPFeedback + switch t.params.MediaTrack.Kind() { + case livekit.TrackType_AUDIO: + rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Audio + case livekit.TrackType_VIDEO: + rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Video + } + downTrack, err := sfu.NewDownTrack(webrtc.RTPCodecCapability{ + MimeType: codec.MimeType, + ClockRate: codec.ClockRate, + Channels: codec.Channels, + SDPFmtpLine: codec.SDPFmtpLine, + RTCPFeedback: rtcpFeedback, + }, wr, t.params.BufferFactory, subscriberID, t.params.ReceiverConfig.PacketBufferSize) + if err != nil { + return nil, err + } + subTrack := NewSubscribedTrack(SubscribedTrackParams{ + PublisherID: t.params.MediaTrack.PublisherID(), + PublisherIdentity: t.params.MediaTrack.PublisherIdentity(), + SubscriberID: subscriberID, + MediaTrack: t.params.MediaTrack, + DownTrack: downTrack, + }) + + var transceiver *webrtc.RTPTransceiver + var sender *webrtc.RTPSender + if sub.ProtocolVersion().SupportsTransceiverReuse() { + // + // AddTrack will create a new transceiver or re-use an unused one + // if the attributes match. This prevents SDP from bloating + // because of dormant transceivers building up. + // + sender, err = sub.SubscriberPC().AddTrack(downTrack) + if err != nil { + return nil, err + } + + // as there is no way to get transceiver from sender, search + for _, tr := range sub.SubscriberPC().GetTransceivers() { + if tr.Sender() == sender { + transceiver = tr + break + } + } + if transceiver == nil { + // cannot add, no transceiver + return nil, errors.New("cannot subscribe without a transceiver in place") + } + } else { + transceiver, err = sub.SubscriberPC().AddTransceiverFromTrack(downTrack, webrtc.RTPTransceiverInit{ + Direction: webrtc.RTPTransceiverDirectionSendonly, + }) + if err != nil { + return nil, err + } + + sender = transceiver.Sender() + if sender == nil { + // cannot add, no sender + return nil, errors.New("cannot subscribe without a sender in place") + } + } + + sendParameters := sender.GetParameters() + downTrack.SetRTPHeaderExtensions(sendParameters.HeaderExtensions) + + downTrack.SetTransceiver(transceiver) + // when outtrack is bound, start loop to send reports + downTrack.OnBind(func() { + go subTrack.Bound() + go t.sendDownTrackBindingReports(sub) + }) + downTrack.OnPacketSent(func(_ *sfu.DownTrack, size int) { + if t.params.Telemetry != nil { + t.params.Telemetry.OnDownstreamPacket(subscriberID, t.params.MediaTrack.ID(), size) + } + }) + downTrack.OnPaddingSent(func(_ *sfu.DownTrack, size int) { + if t.params.Telemetry != nil { + t.params.Telemetry.OnDownstreamPacket(subscriberID, t.params.MediaTrack.ID(), size) + } + }) + downTrack.OnRTCP(func(pkts []rtcp.Packet) { + if t.params.Telemetry != nil { + t.params.Telemetry.HandleRTCP(livekit.StreamType_DOWNSTREAM, subscriberID, t.params.MediaTrack.ID(), pkts) + } + }) + + downTrack.OnCloseHandler(func() { + go func() { + t.subscribedTracksMu.Lock() + delete(t.subscribedTracks, subscriberID) + t.subscribedTracksMu.Unlock() + + t.maybeNotifyNoSubscribers() + if t.params.Telemetry != nil { + t.params.Telemetry.TrackUnsubscribed(context.Background(), subscriberID, t.params.MediaTrack.ToProto()) + } + + // ignore if the subscribing sub is not connected + if sub.SubscriberPC().ConnectionState() == webrtc.PeerConnectionStateClosed { + return + } + + // if the source has been terminated, we'll need to terminate all of the subscribedtracks + // however, if the dest sub has disconnected, then we can skip + if sender == nil { + return + } + if t.params.Logger != nil { + t.params.Logger.Debugw("removing peerconnection track", + "track", t.params.MediaTrack.ID(), + "subscriber", sub.Identity(), + "subscriberID", subscriberID, + "kind", t.params.MediaTrack.Kind(), + ) + } + if err := sub.SubscriberPC().RemoveTrack(sender); err != nil { + if err == webrtc.ErrConnectionClosed { + // sub closing, can skip removing subscribedtracks + return + } + if _, ok := err.(*rtcerr.InvalidStateError); !ok { + // most of these are safe to ignore, since the track state might have already + // been set to Inactive + if t.params.Logger != nil { + t.params.Logger.Debugw("could not remove remoteTrack from forwarder", + "error", err, + "subscriber", sub.Identity(), + "subscriberID", subscriberID, + ) + } + } + } + + t.NotifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_OFF) + sub.RemoveSubscribedTrack(subTrack) + sub.Negotiate() + }() + }) + + t.subscribedTracks[subscriberID] = subTrack + subTrack.SetPublisherMuted(t.params.MediaTrack.IsMuted()) + + // since sub will lock, run it in a goroutine to avoid deadlocks + go func() { + t.NotifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_HIGH) // start with HIGH, let subscription change it later + sub.AddSubscribedTrack(subTrack) + sub.Negotiate() + }() + + if t.params.Telemetry != nil { + t.params.Telemetry.TrackSubscribed(context.Background(), subscriberID, t.params.MediaTrack.ToProto()) + } + return downTrack, nil +} + +// RemoveSubscriber removes participant from subscription +// stop all forwarders to the client +func (t *MediaTrackSubscriptions) RemoveSubscriber(participantID livekit.ParticipantID) { + subTrack := t.getSubscribedTrack(participantID) + if subTrack != nil { + go subTrack.DownTrack().Close() + } +} + +func (t *MediaTrackSubscriptions) RemoveAllSubscribers() { + if t.params.Logger != nil { + t.params.Logger.Debugw("removing all subscribers", "track", t.params.MediaTrack.ID()) + } + + t.subscribedTracksMu.RLock() + subscribedTracks := t.subscribedTracks + t.subscribedTracksMu.RUnlock() + + for _, subTrack := range subscribedTracks { + go subTrack.DownTrack().Close() + } +} + +func (t *MediaTrackSubscriptions) RevokeDisallowedSubscribers(allowedSubscriberIDs []livekit.ParticipantID) []livekit.ParticipantID { + var revokedSubscriberIDs []livekit.ParticipantID + + t.subscribedTracksMu.RLock() + subscribedTracks := t.subscribedTracks + t.subscribedTracksMu.RUnlock() + + // LK-TODO: large number of subscribers needs to be solved for this loop + for subID, subTrack := range subscribedTracks { + found := false + for _, allowedID := range allowedSubscriberIDs { + if subID == allowedID { + found = true + break + } + } + + if !found { + go subTrack.DownTrack().Close() + revokedSubscriberIDs = append(revokedSubscriberIDs, subID) + } + } + + return revokedSubscriberIDs +} + +func (t *MediaTrackSubscriptions) UpdateVideoLayers() { + t.subscribedTracksMu.RLock() + subscribedTracks := t.subscribedTracks + t.subscribedTracksMu.RUnlock() + + for _, st := range subscribedTracks { + st.UpdateVideoLayer() + } +} + +func (t *MediaTrackSubscriptions) getSubscribedTrack(subscriberID livekit.ParticipantID) types.SubscribedTrack { + t.subscribedTracksMu.RLock() + defer t.subscribedTracksMu.RUnlock() + + return t.subscribedTracks[subscriberID] +} + +// TODO: send for all downtracks from the source participant +// https://tools.ietf.org/html/rfc7941 +func (t *MediaTrackSubscriptions) sendDownTrackBindingReports(sub types.Participant) { + var sd []rtcp.SourceDescriptionChunk + + subTrack := t.getSubscribedTrack(sub.ID()) + if subTrack == nil { + return + } + + chunks := subTrack.DownTrack().CreateSourceDescriptionChunks() + if chunks == nil { + return + } + sd = append(sd, chunks...) + + pkts := []rtcp.Packet{ + &rtcp.SourceDescription{Chunks: sd}, + } + + go func() { + defer RecoverSilent() + batch := pkts + i := 0 + for { + if err := sub.SubscriberPC().WriteRTCP(batch); err != nil { + if t.params.Logger != nil { + t.params.Logger.Errorw("could not write RTCP", err) + } + return + } + if i > 5 { + return + } + i++ + time.Sleep(20 * time.Millisecond) + } + }() +} + +func (t *MediaTrackSubscriptions) DebugInfo() []map[string]interface{} { + t.subscribedTracksMu.RLock() + subscribedTracks := t.subscribedTracks + t.subscribedTracksMu.RUnlock() + + subscribedTrackInfo := make([]map[string]interface{}, 0) + for _, val := range subscribedTracks { + if st, ok := val.(*SubscribedTrack); ok { + dt := st.DownTrack().DebugInfo() + dt["PubMuted"] = st.pubMuted.Get() + dt["SubMuted"] = st.subMuted.Get() + subscribedTrackInfo = append(subscribedTrackInfo, dt) + } + } + + return subscribedTrackInfo +} + +func (t *MediaTrackSubscriptions) OnSubscribedMaxQualityChange(f func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality)) { + t.onSubscribedMaxQualityChange = f +} + +func (t *MediaTrackSubscriptions) NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) { + if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { + return + } + + t.maxQualityLock.Lock() + if quality == livekit.VideoQuality_OFF { + _, ok := t.maxSubscriberQuality[subscriberID] + if !ok { + t.maxQualityLock.Unlock() + return + } + + delete(t.maxSubscriberQuality, subscriberID) + } else { + maxQuality, ok := t.maxSubscriberQuality[subscriberID] + if ok && maxQuality == quality { + t.maxQualityLock.Unlock() + return + } + + t.maxSubscriberQuality[subscriberID] = quality + } + t.maxQualityLock.Unlock() + + t.updateQualityChange() +} + +func (t *MediaTrackSubscriptions) NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality) { + if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { + return + } + + t.maxQualityLock.Lock() + if quality == livekit.VideoQuality_OFF { + _, ok := t.maxSubscriberNodeQuality[nodeID] + if !ok { + t.maxQualityLock.Unlock() + return + } + + delete(t.maxSubscriberNodeQuality, nodeID) + } else { + maxQuality, ok := t.maxSubscriberNodeQuality[nodeID] + if ok && maxQuality == quality { + t.maxQualityLock.Unlock() + return + } + + t.maxSubscriberNodeQuality[nodeID] = quality + } + t.maxQualityLock.Unlock() + + t.updateQualityChange() +} + +func (t *MediaTrackSubscriptions) startMaxQualityTimer() { + t.maxQualityLock.Lock() + defer t.maxQualityLock.Unlock() + + if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { + return + } + + t.maxQualityTimer = time.AfterFunc(initialQualityUpdateWait, func() { + t.stopMaxQualityTimer() + t.updateQualityChange() + }) +} + +func (t *MediaTrackSubscriptions) stopMaxQualityTimer() { + t.maxQualityLock.Lock() + defer t.maxQualityLock.Unlock() + + if t.maxQualityTimer != nil { + t.maxQualityTimer.Stop() + t.maxQualityTimer = nil + } +} + +func (t *MediaTrackSubscriptions) updateQualityChange() { + if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { + return + } + + t.maxQualityLock.Lock() + maxSubscribedQuality := livekit.VideoQuality_OFF + for _, subQuality := range t.maxSubscriberQuality { + if maxSubscribedQuality == livekit.VideoQuality_OFF || subQuality > maxSubscribedQuality { + maxSubscribedQuality = subQuality + } + } + + for _, subQuality := range t.maxSubscriberNodeQuality { + if maxSubscribedQuality == livekit.VideoQuality_OFF || subQuality > maxSubscribedQuality { + maxSubscribedQuality = subQuality + } + } + + if maxSubscribedQuality == t.maxSubscribedQuality { + t.maxQualityLock.Unlock() + return + } + + t.maxSubscribedQuality = maxSubscribedQuality + + var subscribedQualities []*livekit.SubscribedQuality + if t.maxSubscribedQuality == livekit.VideoQuality_OFF { + subscribedQualities = []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: false}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + } + } else { + for q := livekit.VideoQuality_LOW; q <= livekit.VideoQuality_HIGH; q++ { + subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{ + Quality: q, + Enabled: q <= t.maxSubscribedQuality, + }) + } + } + t.maxQualityLock.Unlock() + + if t.onSubscribedMaxQualityChange != nil { + t.onSubscribedMaxQualityChange(subscribedQualities, maxSubscribedQuality) + } +} + +func (t *MediaTrackSubscriptions) numSubscribedLayers() uint32 { + t.maxQualityLock.RLock() + numSubscribedLayers := uint32(0) + if t.maxSubscribedQuality != livekit.VideoQuality_OFF { + numSubscribedLayers = uint32(SpatialLayerForQuality(t.maxSubscribedQuality) + 1) + } + t.maxQualityLock.RUnlock() + + return numSubscribedLayers +} + +func (t *MediaTrackSubscriptions) maybeNotifyNoSubscribers() { + if t.onNoSubscribers == nil { + return + } + + t.subscribedTracksMu.RLock() + empty := len(t.subscribedTracks) == 0 + t.subscribedTracksMu.RUnlock() + + if empty { + t.onNoSubscribers() + } +} diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index d962740c3..9c7b697d9 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -789,6 +789,14 @@ func (p *ParticipantImpl) SubscriptionPermissionUpdate(publisherID livekit.Parti } } +func (p *ParticipantImpl) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error { + return p.uptrackManager.UpdateSubscribedQuality(nodeID, trackID, maxQuality) +} + +func (p *ParticipantImpl) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error { + return p.uptrackManager.UpdateMediaLoss(nodeID, trackID, fractionalLoss) +} + func (p *ParticipantImpl) setupUptrackManager() { p.uptrackManager = NewUptrackManager(UptrackManagerParams{ Identity: p.params.Identity, diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 90a7b2199..7a6ef25ea 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -99,7 +99,7 @@ func (t *SubscribedTrack) UpdateVideoLayer() { return } if t.subMuted.Get() { - t.MediaTrack().NotifySubscriberMute(t.params.SubscriberID) + t.MediaTrack().NotifySubscriberMaxQuality(t.params.SubscriberID, livekit.VideoQuality_OFF) return } settings, ok := t.settings.Load().(*livekit.UpdateTrackSettings) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index a6405f692..a4dac0e1e 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -93,6 +93,10 @@ type Participant interface { UpdateSubscriptionPermissions(permissions *livekit.UpdateSubscriptionPermissions, resolver func(participantID livekit.ParticipantID) Participant) error SubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool) + UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error + + UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error + DebugInfo() map[string]interface{} } @@ -116,6 +120,11 @@ type MediaTrack interface { Source() livekit.TrackSource IsSimulcast() bool + PublisherID() livekit.ParticipantID + PublisherIdentity() livekit.ParticipantIdentity + + ToProto() *livekit.TrackInfo + // subscribers AddSubscriber(participant Participant) error RemoveSubscriber(participantID livekit.ParticipantID) @@ -126,8 +135,10 @@ type MediaTrack interface { // returns quality information that's appropriate for width & height GetQualityForDimension(width, height uint32) livekit.VideoQuality - NotifySubscriberMute(subscriberID livekit.ParticipantID) NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) + NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality) + + NotifySubscriberNodeMediaLoss(nodeID string, fractionalLoss uint8) } // PublishedTrack is the main interface representing a track published to the room diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 128a750f9..f818f7f0c 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -99,10 +99,37 @@ type FakeMediaTrack struct { arg1 livekit.ParticipantID arg2 livekit.VideoQuality } - NotifySubscriberMuteStub func(livekit.ParticipantID) - notifySubscriberMuteMutex sync.RWMutex - notifySubscriberMuteArgsForCall []struct { - arg1 livekit.ParticipantID + NotifySubscriberNodeMaxQualityStub func(string, livekit.VideoQuality) + notifySubscriberNodeMaxQualityMutex sync.RWMutex + notifySubscriberNodeMaxQualityArgsForCall []struct { + arg1 string + arg2 livekit.VideoQuality + } + NotifySubscriberNodeMediaLossStub func(string, uint8) + notifySubscriberNodeMediaLossMutex sync.RWMutex + notifySubscriberNodeMediaLossArgsForCall []struct { + arg1 string + arg2 uint8 + } + PublisherIDStub func() livekit.ParticipantID + publisherIDMutex sync.RWMutex + publisherIDArgsForCall []struct { + } + publisherIDReturns struct { + result1 livekit.ParticipantID + } + publisherIDReturnsOnCall map[int]struct { + result1 livekit.ParticipantID + } + PublisherIdentityStub func() livekit.ParticipantIdentity + publisherIdentityMutex sync.RWMutex + publisherIdentityArgsForCall []struct { + } + publisherIdentityReturns struct { + result1 livekit.ParticipantIdentity + } + publisherIdentityReturnsOnCall map[int]struct { + result1 livekit.ParticipantIdentity } RemoveAllSubscribersStub func() removeAllSubscribersMutex sync.RWMutex @@ -139,6 +166,16 @@ type FakeMediaTrack struct { sourceReturnsOnCall map[int]struct { result1 livekit.TrackSource } + ToProtoStub func() *livekit.TrackInfo + toProtoMutex sync.RWMutex + toProtoArgsForCall []struct { + } + toProtoReturns struct { + result1 *livekit.TrackInfo + } + toProtoReturnsOnCall map[int]struct { + result1 *livekit.TrackInfo + } UpdateVideoLayersStub func([]*livekit.VideoLayer) updateVideoLayersMutex sync.RWMutex updateVideoLayersArgsForCall []struct { @@ -630,36 +667,176 @@ func (fake *FakeMediaTrack) NotifySubscriberMaxQualityArgsForCall(i int) (liveki return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeMediaTrack) NotifySubscriberMute(arg1 livekit.ParticipantID) { - fake.notifySubscriberMuteMutex.Lock() - fake.notifySubscriberMuteArgsForCall = append(fake.notifySubscriberMuteArgsForCall, struct { - arg1 livekit.ParticipantID - }{arg1}) - stub := fake.NotifySubscriberMuteStub - fake.recordInvocation("NotifySubscriberMute", []interface{}{arg1}) - fake.notifySubscriberMuteMutex.Unlock() +func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQuality(arg1 string, arg2 livekit.VideoQuality) { + fake.notifySubscriberNodeMaxQualityMutex.Lock() + fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct { + arg1 string + arg2 livekit.VideoQuality + }{arg1, arg2}) + stub := fake.NotifySubscriberNodeMaxQualityStub + fake.recordInvocation("NotifySubscriberNodeMaxQuality", []interface{}{arg1, arg2}) + fake.notifySubscriberNodeMaxQualityMutex.Unlock() if stub != nil { - fake.NotifySubscriberMuteStub(arg1) + fake.NotifySubscriberNodeMaxQualityStub(arg1, arg2) } } -func (fake *FakeMediaTrack) NotifySubscriberMuteCallCount() int { - fake.notifySubscriberMuteMutex.RLock() - defer fake.notifySubscriberMuteMutex.RUnlock() - return len(fake.notifySubscriberMuteArgsForCall) +func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCallCount() int { + fake.notifySubscriberNodeMaxQualityMutex.RLock() + defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() + return len(fake.notifySubscriberNodeMaxQualityArgsForCall) } -func (fake *FakeMediaTrack) NotifySubscriberMuteCalls(stub func(livekit.ParticipantID)) { - fake.notifySubscriberMuteMutex.Lock() - defer fake.notifySubscriberMuteMutex.Unlock() - fake.NotifySubscriberMuteStub = stub +func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(string, livekit.VideoQuality)) { + fake.notifySubscriberNodeMaxQualityMutex.Lock() + defer fake.notifySubscriberNodeMaxQualityMutex.Unlock() + fake.NotifySubscriberNodeMaxQualityStub = stub } -func (fake *FakeMediaTrack) NotifySubscriberMuteArgsForCall(i int) livekit.ParticipantID { - fake.notifySubscriberMuteMutex.RLock() - defer fake.notifySubscriberMuteMutex.RUnlock() - argsForCall := fake.notifySubscriberMuteArgsForCall[i] - return argsForCall.arg1 +func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (string, livekit.VideoQuality) { + fake.notifySubscriberNodeMaxQualityMutex.RLock() + defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() + argsForCall := fake.notifySubscriberNodeMaxQualityArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLoss(arg1 string, arg2 uint8) { + fake.notifySubscriberNodeMediaLossMutex.Lock() + fake.notifySubscriberNodeMediaLossArgsForCall = append(fake.notifySubscriberNodeMediaLossArgsForCall, struct { + arg1 string + arg2 uint8 + }{arg1, arg2}) + stub := fake.NotifySubscriberNodeMediaLossStub + fake.recordInvocation("NotifySubscriberNodeMediaLoss", []interface{}{arg1, arg2}) + fake.notifySubscriberNodeMediaLossMutex.Unlock() + if stub != nil { + fake.NotifySubscriberNodeMediaLossStub(arg1, arg2) + } +} + +func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCallCount() int { + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() + return len(fake.notifySubscriberNodeMediaLossArgsForCall) +} + +func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCalls(stub func(string, uint8)) { + fake.notifySubscriberNodeMediaLossMutex.Lock() + defer fake.notifySubscriberNodeMediaLossMutex.Unlock() + fake.NotifySubscriberNodeMediaLossStub = stub +} + +func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (string, uint8) { + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() + argsForCall := fake.notifySubscriberNodeMediaLossArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeMediaTrack) PublisherID() livekit.ParticipantID { + fake.publisherIDMutex.Lock() + ret, specificReturn := fake.publisherIDReturnsOnCall[len(fake.publisherIDArgsForCall)] + fake.publisherIDArgsForCall = append(fake.publisherIDArgsForCall, struct { + }{}) + stub := fake.PublisherIDStub + fakeReturns := fake.publisherIDReturns + fake.recordInvocation("PublisherID", []interface{}{}) + fake.publisherIDMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) PublisherIDCallCount() int { + fake.publisherIDMutex.RLock() + defer fake.publisherIDMutex.RUnlock() + return len(fake.publisherIDArgsForCall) +} + +func (fake *FakeMediaTrack) PublisherIDCalls(stub func() livekit.ParticipantID) { + fake.publisherIDMutex.Lock() + defer fake.publisherIDMutex.Unlock() + fake.PublisherIDStub = stub +} + +func (fake *FakeMediaTrack) PublisherIDReturns(result1 livekit.ParticipantID) { + fake.publisherIDMutex.Lock() + defer fake.publisherIDMutex.Unlock() + fake.PublisherIDStub = nil + fake.publisherIDReturns = struct { + result1 livekit.ParticipantID + }{result1} +} + +func (fake *FakeMediaTrack) PublisherIDReturnsOnCall(i int, result1 livekit.ParticipantID) { + fake.publisherIDMutex.Lock() + defer fake.publisherIDMutex.Unlock() + fake.PublisherIDStub = nil + if fake.publisherIDReturnsOnCall == nil { + fake.publisherIDReturnsOnCall = make(map[int]struct { + result1 livekit.ParticipantID + }) + } + fake.publisherIDReturnsOnCall[i] = struct { + result1 livekit.ParticipantID + }{result1} +} + +func (fake *FakeMediaTrack) PublisherIdentity() livekit.ParticipantIdentity { + fake.publisherIdentityMutex.Lock() + ret, specificReturn := fake.publisherIdentityReturnsOnCall[len(fake.publisherIdentityArgsForCall)] + fake.publisherIdentityArgsForCall = append(fake.publisherIdentityArgsForCall, struct { + }{}) + stub := fake.PublisherIdentityStub + fakeReturns := fake.publisherIdentityReturns + fake.recordInvocation("PublisherIdentity", []interface{}{}) + fake.publisherIdentityMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) PublisherIdentityCallCount() int { + fake.publisherIdentityMutex.RLock() + defer fake.publisherIdentityMutex.RUnlock() + return len(fake.publisherIdentityArgsForCall) +} + +func (fake *FakeMediaTrack) PublisherIdentityCalls(stub func() livekit.ParticipantIdentity) { + fake.publisherIdentityMutex.Lock() + defer fake.publisherIdentityMutex.Unlock() + fake.PublisherIdentityStub = stub +} + +func (fake *FakeMediaTrack) PublisherIdentityReturns(result1 livekit.ParticipantIdentity) { + fake.publisherIdentityMutex.Lock() + defer fake.publisherIdentityMutex.Unlock() + fake.PublisherIdentityStub = nil + fake.publisherIdentityReturns = struct { + result1 livekit.ParticipantIdentity + }{result1} +} + +func (fake *FakeMediaTrack) PublisherIdentityReturnsOnCall(i int, result1 livekit.ParticipantIdentity) { + fake.publisherIdentityMutex.Lock() + defer fake.publisherIdentityMutex.Unlock() + fake.PublisherIdentityStub = nil + if fake.publisherIdentityReturnsOnCall == nil { + fake.publisherIdentityReturnsOnCall = make(map[int]struct { + result1 livekit.ParticipantIdentity + }) + } + fake.publisherIdentityReturnsOnCall[i] = struct { + result1 livekit.ParticipantIdentity + }{result1} } func (fake *FakeMediaTrack) RemoveAllSubscribers() { @@ -869,6 +1046,59 @@ func (fake *FakeMediaTrack) SourceReturnsOnCall(i int, result1 livekit.TrackSour }{result1} } +func (fake *FakeMediaTrack) ToProto() *livekit.TrackInfo { + fake.toProtoMutex.Lock() + ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)] + fake.toProtoArgsForCall = append(fake.toProtoArgsForCall, struct { + }{}) + stub := fake.ToProtoStub + fakeReturns := fake.toProtoReturns + fake.recordInvocation("ToProto", []interface{}{}) + fake.toProtoMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) ToProtoCallCount() int { + fake.toProtoMutex.RLock() + defer fake.toProtoMutex.RUnlock() + return len(fake.toProtoArgsForCall) +} + +func (fake *FakeMediaTrack) ToProtoCalls(stub func() *livekit.TrackInfo) { + fake.toProtoMutex.Lock() + defer fake.toProtoMutex.Unlock() + fake.ToProtoStub = stub +} + +func (fake *FakeMediaTrack) ToProtoReturns(result1 *livekit.TrackInfo) { + fake.toProtoMutex.Lock() + defer fake.toProtoMutex.Unlock() + fake.ToProtoStub = nil + fake.toProtoReturns = struct { + result1 *livekit.TrackInfo + }{result1} +} + +func (fake *FakeMediaTrack) ToProtoReturnsOnCall(i int, result1 *livekit.TrackInfo) { + fake.toProtoMutex.Lock() + defer fake.toProtoMutex.Unlock() + fake.ToProtoStub = nil + if fake.toProtoReturnsOnCall == nil { + fake.toProtoReturnsOnCall = make(map[int]struct { + result1 *livekit.TrackInfo + }) + } + fake.toProtoReturnsOnCall[i] = struct { + result1 *livekit.TrackInfo + }{result1} +} + func (fake *FakeMediaTrack) UpdateVideoLayers(arg1 []*livekit.VideoLayer) { var arg1Copy []*livekit.VideoLayer if arg1 != nil { @@ -927,8 +1157,14 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.nameMutex.RUnlock() fake.notifySubscriberMaxQualityMutex.RLock() defer fake.notifySubscriberMaxQualityMutex.RUnlock() - fake.notifySubscriberMuteMutex.RLock() - defer fake.notifySubscriberMuteMutex.RUnlock() + fake.notifySubscriberNodeMaxQualityMutex.RLock() + defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() + fake.publisherIDMutex.RLock() + defer fake.publisherIDMutex.RUnlock() + fake.publisherIdentityMutex.RLock() + defer fake.publisherIdentityMutex.RUnlock() fake.removeAllSubscribersMutex.RLock() defer fake.removeAllSubscribersMutex.RUnlock() fake.removeSubscriberMutex.RLock() @@ -939,6 +1175,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.setMutedMutex.RUnlock() fake.sourceMutex.RLock() defer fake.sourceMutex.RUnlock() + fake.toProtoMutex.RLock() + defer fake.toProtoMutex.RUnlock() fake.updateVideoLayersMutex.RLock() defer fake.updateVideoLayersMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 2ff751ac3..c5dea5305 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -494,6 +494,32 @@ type FakeParticipant struct { toProtoReturnsOnCall map[int]struct { result1 *livekit.ParticipantInfo } + UpdateMediaLossStub func(string, livekit.TrackID, uint32) error + updateMediaLossMutex sync.RWMutex + updateMediaLossArgsForCall []struct { + arg1 string + arg2 livekit.TrackID + arg3 uint32 + } + updateMediaLossReturns struct { + result1 error + } + updateMediaLossReturnsOnCall map[int]struct { + result1 error + } + UpdateSubscribedQualityStub func(string, livekit.TrackID, livekit.VideoQuality) error + updateSubscribedQualityMutex sync.RWMutex + updateSubscribedQualityArgsForCall []struct { + arg1 string + arg2 livekit.TrackID + arg3 livekit.VideoQuality + } + updateSubscribedQualityReturns struct { + result1 error + } + updateSubscribedQualityReturnsOnCall map[int]struct { + result1 error + } UpdateSubscriptionPermissionsStub func(*livekit.UpdateSubscriptionPermissions, func(participantID livekit.ParticipantID) types.Participant) error updateSubscriptionPermissionsMutex sync.RWMutex updateSubscriptionPermissionsArgsForCall []struct { @@ -3142,6 +3168,132 @@ func (fake *FakeParticipant) ToProtoReturnsOnCall(i int, result1 *livekit.Partic }{result1} } +func (fake *FakeParticipant) UpdateMediaLoss(arg1 string, arg2 livekit.TrackID, arg3 uint32) error { + fake.updateMediaLossMutex.Lock() + ret, specificReturn := fake.updateMediaLossReturnsOnCall[len(fake.updateMediaLossArgsForCall)] + fake.updateMediaLossArgsForCall = append(fake.updateMediaLossArgsForCall, struct { + arg1 string + arg2 livekit.TrackID + arg3 uint32 + }{arg1, arg2, arg3}) + stub := fake.UpdateMediaLossStub + fakeReturns := fake.updateMediaLossReturns + fake.recordInvocation("UpdateMediaLoss", []interface{}{arg1, arg2, arg3}) + fake.updateMediaLossMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) UpdateMediaLossCallCount() int { + fake.updateMediaLossMutex.RLock() + defer fake.updateMediaLossMutex.RUnlock() + return len(fake.updateMediaLossArgsForCall) +} + +func (fake *FakeParticipant) UpdateMediaLossCalls(stub func(string, livekit.TrackID, uint32) error) { + fake.updateMediaLossMutex.Lock() + defer fake.updateMediaLossMutex.Unlock() + fake.UpdateMediaLossStub = stub +} + +func (fake *FakeParticipant) UpdateMediaLossArgsForCall(i int) (string, livekit.TrackID, uint32) { + fake.updateMediaLossMutex.RLock() + defer fake.updateMediaLossMutex.RUnlock() + argsForCall := fake.updateMediaLossArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 +} + +func (fake *FakeParticipant) UpdateMediaLossReturns(result1 error) { + fake.updateMediaLossMutex.Lock() + defer fake.updateMediaLossMutex.Unlock() + fake.UpdateMediaLossStub = nil + fake.updateMediaLossReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeParticipant) UpdateMediaLossReturnsOnCall(i int, result1 error) { + fake.updateMediaLossMutex.Lock() + defer fake.updateMediaLossMutex.Unlock() + fake.UpdateMediaLossStub = nil + if fake.updateMediaLossReturnsOnCall == nil { + fake.updateMediaLossReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.updateMediaLossReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeParticipant) UpdateSubscribedQuality(arg1 string, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error { + fake.updateSubscribedQualityMutex.Lock() + ret, specificReturn := fake.updateSubscribedQualityReturnsOnCall[len(fake.updateSubscribedQualityArgsForCall)] + fake.updateSubscribedQualityArgsForCall = append(fake.updateSubscribedQualityArgsForCall, struct { + arg1 string + arg2 livekit.TrackID + arg3 livekit.VideoQuality + }{arg1, arg2, arg3}) + stub := fake.UpdateSubscribedQualityStub + fakeReturns := fake.updateSubscribedQualityReturns + fake.recordInvocation("UpdateSubscribedQuality", []interface{}{arg1, arg2, arg3}) + fake.updateSubscribedQualityMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) UpdateSubscribedQualityCallCount() int { + fake.updateSubscribedQualityMutex.RLock() + defer fake.updateSubscribedQualityMutex.RUnlock() + return len(fake.updateSubscribedQualityArgsForCall) +} + +func (fake *FakeParticipant) UpdateSubscribedQualityCalls(stub func(string, livekit.TrackID, livekit.VideoQuality) error) { + fake.updateSubscribedQualityMutex.Lock() + defer fake.updateSubscribedQualityMutex.Unlock() + fake.UpdateSubscribedQualityStub = stub +} + +func (fake *FakeParticipant) UpdateSubscribedQualityArgsForCall(i int) (string, livekit.TrackID, livekit.VideoQuality) { + fake.updateSubscribedQualityMutex.RLock() + defer fake.updateSubscribedQualityMutex.RUnlock() + argsForCall := fake.updateSubscribedQualityArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 +} + +func (fake *FakeParticipant) UpdateSubscribedQualityReturns(result1 error) { + fake.updateSubscribedQualityMutex.Lock() + defer fake.updateSubscribedQualityMutex.Unlock() + fake.UpdateSubscribedQualityStub = nil + fake.updateSubscribedQualityReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeParticipant) UpdateSubscribedQualityReturnsOnCall(i int, result1 error) { + fake.updateSubscribedQualityMutex.Lock() + defer fake.updateSubscribedQualityMutex.Unlock() + fake.UpdateSubscribedQualityStub = nil + if fake.updateSubscribedQualityReturnsOnCall == nil { + fake.updateSubscribedQualityReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.updateSubscribedQualityReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeParticipant) UpdateSubscriptionPermissions(arg1 *livekit.UpdateSubscriptionPermissions, arg2 func(participantID livekit.ParticipantID) types.Participant) error { fake.updateSubscriptionPermissionsMutex.Lock() ret, specificReturn := fake.updateSubscriptionPermissionsReturnsOnCall[len(fake.updateSubscriptionPermissionsArgsForCall)] @@ -3315,6 +3467,10 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.subscriptionPermissionUpdateMutex.RUnlock() fake.toProtoMutex.RLock() defer fake.toProtoMutex.RUnlock() + fake.updateMediaLossMutex.RLock() + defer fake.updateMediaLossMutex.RUnlock() + fake.updateSubscribedQualityMutex.RLock() + defer fake.updateSubscribedQualityMutex.RUnlock() fake.updateSubscriptionPermissionsMutex.RLock() defer fake.updateSubscriptionPermissionsMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} diff --git a/pkg/rtc/types/typesfakes/fake_published_track.go b/pkg/rtc/types/typesfakes/fake_published_track.go index 1cd1ab58c..c29239aa8 100644 --- a/pkg/rtc/types/typesfakes/fake_published_track.go +++ b/pkg/rtc/types/typesfakes/fake_published_track.go @@ -115,10 +115,17 @@ type FakePublishedTrack struct { arg1 livekit.ParticipantID arg2 livekit.VideoQuality } - NotifySubscriberMuteStub func(livekit.ParticipantID) - notifySubscriberMuteMutex sync.RWMutex - notifySubscriberMuteArgsForCall []struct { - arg1 livekit.ParticipantID + NotifySubscriberNodeMaxQualityStub func(string, livekit.VideoQuality) + notifySubscriberNodeMaxQualityMutex sync.RWMutex + notifySubscriberNodeMaxQualityArgsForCall []struct { + arg1 string + arg2 livekit.VideoQuality + } + NotifySubscriberNodeMediaLossStub func(string, uint8) + notifySubscriberNodeMediaLossMutex sync.RWMutex + notifySubscriberNodeMediaLossArgsForCall []struct { + arg1 string + arg2 uint8 } NumUpTracksStub func() (uint32, uint32) numUpTracksMutex sync.RWMutex @@ -142,6 +149,26 @@ type FakePublishedTrack struct { publishLossPercentageReturnsOnCall map[int]struct { result1 uint32 } + PublisherIDStub func() livekit.ParticipantID + publisherIDMutex sync.RWMutex + publisherIDArgsForCall []struct { + } + publisherIDReturns struct { + result1 livekit.ParticipantID + } + publisherIDReturnsOnCall map[int]struct { + result1 livekit.ParticipantID + } + PublisherIdentityStub func() livekit.ParticipantIdentity + publisherIdentityMutex sync.RWMutex + publisherIdentityArgsForCall []struct { + } + publisherIdentityReturns struct { + result1 livekit.ParticipantIdentity + } + publisherIdentityReturnsOnCall map[int]struct { + result1 livekit.ParticipantIdentity + } ReceiverStub func() sfu.TrackReceiver receiverMutex sync.RWMutex receiverArgsForCall []struct { @@ -793,36 +820,70 @@ func (fake *FakePublishedTrack) NotifySubscriberMaxQualityArgsForCall(i int) (li return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakePublishedTrack) NotifySubscriberMute(arg1 livekit.ParticipantID) { - fake.notifySubscriberMuteMutex.Lock() - fake.notifySubscriberMuteArgsForCall = append(fake.notifySubscriberMuteArgsForCall, struct { - arg1 livekit.ParticipantID - }{arg1}) - stub := fake.NotifySubscriberMuteStub - fake.recordInvocation("NotifySubscriberMute", []interface{}{arg1}) - fake.notifySubscriberMuteMutex.Unlock() +func (fake *FakePublishedTrack) NotifySubscriberNodeMaxQuality(arg1 string, arg2 livekit.VideoQuality) { + fake.notifySubscriberNodeMaxQualityMutex.Lock() + fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct { + arg1 string + arg2 livekit.VideoQuality + }{arg1, arg2}) + stub := fake.NotifySubscriberNodeMaxQualityStub + fake.recordInvocation("NotifySubscriberNodeMaxQuality", []interface{}{arg1, arg2}) + fake.notifySubscriberNodeMaxQualityMutex.Unlock() if stub != nil { - fake.NotifySubscriberMuteStub(arg1) + fake.NotifySubscriberNodeMaxQualityStub(arg1, arg2) } } -func (fake *FakePublishedTrack) NotifySubscriberMuteCallCount() int { - fake.notifySubscriberMuteMutex.RLock() - defer fake.notifySubscriberMuteMutex.RUnlock() - return len(fake.notifySubscriberMuteArgsForCall) +func (fake *FakePublishedTrack) NotifySubscriberNodeMaxQualityCallCount() int { + fake.notifySubscriberNodeMaxQualityMutex.RLock() + defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() + return len(fake.notifySubscriberNodeMaxQualityArgsForCall) } -func (fake *FakePublishedTrack) NotifySubscriberMuteCalls(stub func(livekit.ParticipantID)) { - fake.notifySubscriberMuteMutex.Lock() - defer fake.notifySubscriberMuteMutex.Unlock() - fake.NotifySubscriberMuteStub = stub +func (fake *FakePublishedTrack) NotifySubscriberNodeMaxQualityCalls(stub func(string, livekit.VideoQuality)) { + fake.notifySubscriberNodeMaxQualityMutex.Lock() + defer fake.notifySubscriberNodeMaxQualityMutex.Unlock() + fake.NotifySubscriberNodeMaxQualityStub = stub } -func (fake *FakePublishedTrack) NotifySubscriberMuteArgsForCall(i int) livekit.ParticipantID { - fake.notifySubscriberMuteMutex.RLock() - defer fake.notifySubscriberMuteMutex.RUnlock() - argsForCall := fake.notifySubscriberMuteArgsForCall[i] - return argsForCall.arg1 +func (fake *FakePublishedTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (string, livekit.VideoQuality) { + fake.notifySubscriberNodeMaxQualityMutex.RLock() + defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() + argsForCall := fake.notifySubscriberNodeMaxQualityArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakePublishedTrack) NotifySubscriberNodeMediaLoss(arg1 string, arg2 uint8) { + fake.notifySubscriberNodeMediaLossMutex.Lock() + fake.notifySubscriberNodeMediaLossArgsForCall = append(fake.notifySubscriberNodeMediaLossArgsForCall, struct { + arg1 string + arg2 uint8 + }{arg1, arg2}) + stub := fake.NotifySubscriberNodeMediaLossStub + fake.recordInvocation("NotifySubscriberNodeMediaLoss", []interface{}{arg1, arg2}) + fake.notifySubscriberNodeMediaLossMutex.Unlock() + if stub != nil { + fake.NotifySubscriberNodeMediaLossStub(arg1, arg2) + } +} + +func (fake *FakePublishedTrack) NotifySubscriberNodeMediaLossCallCount() int { + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() + return len(fake.notifySubscriberNodeMediaLossArgsForCall) +} + +func (fake *FakePublishedTrack) NotifySubscriberNodeMediaLossCalls(stub func(string, uint8)) { + fake.notifySubscriberNodeMediaLossMutex.Lock() + defer fake.notifySubscriberNodeMediaLossMutex.Unlock() + fake.NotifySubscriberNodeMediaLossStub = stub +} + +func (fake *FakePublishedTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (string, uint8) { + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() + argsForCall := fake.notifySubscriberNodeMediaLossArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakePublishedTrack) NumUpTracks() (uint32, uint32) { @@ -934,6 +995,112 @@ func (fake *FakePublishedTrack) PublishLossPercentageReturnsOnCall(i int, result }{result1} } +func (fake *FakePublishedTrack) PublisherID() livekit.ParticipantID { + fake.publisherIDMutex.Lock() + ret, specificReturn := fake.publisherIDReturnsOnCall[len(fake.publisherIDArgsForCall)] + fake.publisherIDArgsForCall = append(fake.publisherIDArgsForCall, struct { + }{}) + stub := fake.PublisherIDStub + fakeReturns := fake.publisherIDReturns + fake.recordInvocation("PublisherID", []interface{}{}) + fake.publisherIDMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakePublishedTrack) PublisherIDCallCount() int { + fake.publisherIDMutex.RLock() + defer fake.publisherIDMutex.RUnlock() + return len(fake.publisherIDArgsForCall) +} + +func (fake *FakePublishedTrack) PublisherIDCalls(stub func() livekit.ParticipantID) { + fake.publisherIDMutex.Lock() + defer fake.publisherIDMutex.Unlock() + fake.PublisherIDStub = stub +} + +func (fake *FakePublishedTrack) PublisherIDReturns(result1 livekit.ParticipantID) { + fake.publisherIDMutex.Lock() + defer fake.publisherIDMutex.Unlock() + fake.PublisherIDStub = nil + fake.publisherIDReturns = struct { + result1 livekit.ParticipantID + }{result1} +} + +func (fake *FakePublishedTrack) PublisherIDReturnsOnCall(i int, result1 livekit.ParticipantID) { + fake.publisherIDMutex.Lock() + defer fake.publisherIDMutex.Unlock() + fake.PublisherIDStub = nil + if fake.publisherIDReturnsOnCall == nil { + fake.publisherIDReturnsOnCall = make(map[int]struct { + result1 livekit.ParticipantID + }) + } + fake.publisherIDReturnsOnCall[i] = struct { + result1 livekit.ParticipantID + }{result1} +} + +func (fake *FakePublishedTrack) PublisherIdentity() livekit.ParticipantIdentity { + fake.publisherIdentityMutex.Lock() + ret, specificReturn := fake.publisherIdentityReturnsOnCall[len(fake.publisherIdentityArgsForCall)] + fake.publisherIdentityArgsForCall = append(fake.publisherIdentityArgsForCall, struct { + }{}) + stub := fake.PublisherIdentityStub + fakeReturns := fake.publisherIdentityReturns + fake.recordInvocation("PublisherIdentity", []interface{}{}) + fake.publisherIdentityMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakePublishedTrack) PublisherIdentityCallCount() int { + fake.publisherIdentityMutex.RLock() + defer fake.publisherIdentityMutex.RUnlock() + return len(fake.publisherIdentityArgsForCall) +} + +func (fake *FakePublishedTrack) PublisherIdentityCalls(stub func() livekit.ParticipantIdentity) { + fake.publisherIdentityMutex.Lock() + defer fake.publisherIdentityMutex.Unlock() + fake.PublisherIdentityStub = stub +} + +func (fake *FakePublishedTrack) PublisherIdentityReturns(result1 livekit.ParticipantIdentity) { + fake.publisherIdentityMutex.Lock() + defer fake.publisherIdentityMutex.Unlock() + fake.PublisherIdentityStub = nil + fake.publisherIdentityReturns = struct { + result1 livekit.ParticipantIdentity + }{result1} +} + +func (fake *FakePublishedTrack) PublisherIdentityReturnsOnCall(i int, result1 livekit.ParticipantIdentity) { + fake.publisherIdentityMutex.Lock() + defer fake.publisherIdentityMutex.Unlock() + fake.PublisherIdentityStub = nil + if fake.publisherIdentityReturnsOnCall == nil { + fake.publisherIdentityReturnsOnCall = make(map[int]struct { + result1 livekit.ParticipantIdentity + }) + } + fake.publisherIdentityReturnsOnCall[i] = struct { + result1 livekit.ParticipantIdentity + }{result1} +} + func (fake *FakePublishedTrack) Receiver() sfu.TrackReceiver { fake.receiverMutex.Lock() ret, specificReturn := fake.receiverReturnsOnCall[len(fake.receiverArgsForCall)] @@ -1415,12 +1582,18 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} { defer fake.nameMutex.RUnlock() fake.notifySubscriberMaxQualityMutex.RLock() defer fake.notifySubscriberMaxQualityMutex.RUnlock() - fake.notifySubscriberMuteMutex.RLock() - defer fake.notifySubscriberMuteMutex.RUnlock() + fake.notifySubscriberNodeMaxQualityMutex.RLock() + defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() fake.numUpTracksMutex.RLock() defer fake.numUpTracksMutex.RUnlock() fake.publishLossPercentageMutex.RLock() defer fake.publishLossPercentageMutex.RUnlock() + fake.publisherIDMutex.RLock() + defer fake.publisherIDMutex.RUnlock() + fake.publisherIdentityMutex.RLock() + defer fake.publisherIdentityMutex.RUnlock() fake.receiverMutex.RLock() defer fake.receiverMutex.RUnlock() fake.removeAllSubscribersMutex.RLock() diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 73990c6db..3207ed303 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -1,6 +1,7 @@ package rtc import ( + "errors" "sync" "github.com/livekit/protocol/livekit" @@ -232,10 +233,7 @@ func (u *UptrackManager) GetAudioLevel() (level uint8, active bool) { level = silentAudioLevel for _, pt := range u.publishedTracks { if mt, ok := pt.(*MediaTrack); ok { - if mt.audioLevel == nil { - continue - } - tl, ta := mt.audioLevel.GetLevel() + tl, ta := mt.GetAudioLevel() if ta { active = true if tl < level { @@ -314,6 +312,40 @@ func (u *UptrackManager) UpdateSubscriptionPermissions( return nil } +func (u *UptrackManager) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error { + u.lock.RLock() + defer u.lock.RUnlock() + + track := u.getPublishedTrack(trackID) + if track == nil { + u.params.Logger.Warnw("could not find track", nil, "trackID", trackID) + return errors.New("could not find track") + } + + if mt, ok := track.(*MediaTrack); ok { + mt.NotifySubscriberNodeMaxQuality(nodeID, maxQuality) + } + + return nil +} + +func (u *UptrackManager) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error { + u.lock.RLock() + defer u.lock.RUnlock() + + track := u.getPublishedTrack(trackID) + if track == nil { + u.params.Logger.Warnw("could not find track", nil, "trackID", trackID) + return errors.New("could not find track") + } + + if mt, ok := track.(*MediaTrack); ok { + mt.NotifySubscriberNodeMediaLoss(nodeID, uint8(fractionalLoss)) + } + + return nil +} + // when a new remoteTrack is created, creates a Track and adds it to room func (u *UptrackManager) MediaTrackReceived(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) { var newTrack bool