From 793b383a5292b23df0a1e79fcd989b0af404b8fa Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Thu, 8 May 2025 12:58:24 +0800 Subject: [PATCH] Add Moving participant to another room (#3648) * Add Moving participant to another room it is implemented in cloud only since the destination room can exist in different node with the source room * Update pkg/service/errors.go Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * rename * test panic * fake LocalParticipantHelper * revert delete line --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- pkg/rtc/mediatrack.go | 8 +- pkg/rtc/mediatrackreceiver.go | 9 +- pkg/rtc/mediatracksubscriptions.go | 5 +- pkg/rtc/participant.go | 123 ++++-- pkg/rtc/participant_internal_test.go | 1 + pkg/rtc/participant_signal.go | 14 +- pkg/rtc/subscriptionmanager.go | 47 +- pkg/rtc/transport.go | 2 - pkg/rtc/transport_test.go | 34 +- pkg/rtc/transportmanager.go | 6 - pkg/rtc/types/interfaces.go | 26 +- pkg/rtc/types/protocol_version.go | 4 + .../typesfakes/fake_local_media_track.go | 66 +++ .../typesfakes/fake_local_participant.go | 243 +++++++++++ .../fake_local_participant_helper.go | 406 ++++++++++++++++++ pkg/rtc/types/typesfakes/fake_media_track.go | 66 +++ pkg/service/auth.go | 2 +- pkg/service/errors.go | 2 +- pkg/service/roommanager.go | 40 +- pkg/service/roomservice.go | 21 +- pkg/service/rtcservice.go | 9 +- 21 files changed, 1021 insertions(+), 113 deletions(-) create mode 100644 pkg/rtc/types/typesfakes/fake_local_participant_helper.go diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 4be5466c3..2b5a93fc5 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -63,7 +63,7 @@ type MediaTrack struct { type MediaTrackParams struct { SignalCid string SdpCid string - ParticipantID livekit.ParticipantID + ParticipantID func() livekit.ParticipantID ParticipantIdentity livekit.ParticipantIdentity ParticipantVersion uint32 BufferFactory *buffer.Factory @@ -433,7 +433,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe buff.OnFinalRtpStats(func(stats *livekit.RTPStats) { t.params.Telemetry.TrackPublishRTPStats( context.Background(), - t.params.ParticipantID, + t.params.ParticipantID(), t.ID(), mimeType, int(layer), @@ -502,3 +502,7 @@ func (t *MediaTrack) enableRegression() bool { return t.backupCodecPolicy == livekit.BackupCodecPolicy_REGRESSION || (t.backupCodecPolicy == livekit.BackupCodecPolicy_PREFER_REGRESSION && t.params.ShouldRegressCodec()) } + +func (t *MediaTrack) Logger() logger.Logger { + return t.params.Logger +} diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index f94add559..9648c118a 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -121,7 +121,7 @@ func (r *simulcastReceiver) IsRegressed() bool { type MediaTrackReceiverParams struct { MediaTrack types.MediaTrack IsRelayed bool - ParticipantID livekit.ParticipantID + ParticipantID func() livekit.ParticipantID ParticipantIdentity livekit.ParticipantIdentity ParticipantVersion uint32 ReceiverConfig ReceiverConfig @@ -480,7 +480,7 @@ func (t *MediaTrackReceiver) Stream() string { } func (t *MediaTrackReceiver) PublisherID() livekit.ParticipantID { - return t.params.ParticipantID + return t.params.ParticipantID() } func (t *MediaTrackReceiver) PublisherIdentity() livekit.ParticipantIdentity { @@ -594,6 +594,7 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) (types.Su Logger: tLogger, DisableRed: t.TrackInfo().GetDisableRed() || !t.params.AudioConfig.ActiveREDEncoding, }) + subID := sub.ID() subTrack, err := t.MediaTrackSubscriptions.AddSubscriber(sub, wr) // media track could have been closed while adding subscription @@ -607,8 +608,8 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) (types.Su t.lock.RUnlock() if remove { - t.params.Logger.Debugw("removing susbcriber on a not-open track", "subscriberID", sub.ID(), "isExpectedToResume", isExpectedToResume) - _ = t.MediaTrackSubscriptions.RemoveSubscriber(sub.ID(), isExpectedToResume) + t.params.Logger.Debugw("removing susbcriber on a not-open track", "subscriberID", subID, "isExpectedToResume", isExpectedToResume) + _ = t.MediaTrackSubscriptions.RemoveSubscriber(subID, isExpectedToResume) return nil, ErrNotOpen } diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 63d0f175c..1e8f73cdd 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -333,7 +333,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * downTrack.SetTransceiver(transceiver) downTrack.OnCloseHandler(func(isExpectedToResume bool) { - t.downTrackClosed(sub, subTrack, isExpectedToResume) + t.downTrackClosed(subscriberID, sub, subTrack, isExpectedToResume) }) t.subscribedTracksMu.Lock() @@ -443,6 +443,7 @@ func (t *MediaTrackSubscriptions) DebugInfo() []map[string]interface{} { } func (t *MediaTrackSubscriptions) downTrackClosed( + subscriberID livekit.ParticipantID, sub types.LocalParticipant, subTrack types.SubscribedTrack, isExpectedToResume bool, @@ -460,7 +461,7 @@ func (t *MediaTrackSubscriptions) downTrackClosed( go func() { t.subscribedTracksMu.Lock() - delete(t.subscribedTracks, sub.ID()) + delete(t.subscribedTracks, subscriberID) t.subscribedTracksMu.Unlock() subTrack.Close(isExpectedToResume) }() diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 3c7cc8d9e..f4dceeb8c 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -129,6 +129,7 @@ type ParticipantParams struct { PublishEnabledCodecs []*livekit.Codec SubscribeEnabledCodecs []*livekit.Codec Logger logger.Logger + LoggerResolver logger.DeferredFieldResolver SimTracks map[uint32]SimulcastTrackInfo Grants *auth.ClaimGrants InitialVersion uint32 @@ -142,16 +143,12 @@ type ParticipantParams struct { TCPFallbackRTTThreshold int AllowUDPUnstableFallback bool TURNSEnabled bool - GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo - GetRegionSettings func(ip string) *livekit.RegionSettings - GetSubscriberForwarderState func(p types.LocalParticipant) (map[livekit.TrackID]*livekit.RTPForwarderState, error) - ShouldRegressCodec func() bool + ParticipantHelper types.LocalParticipantHelper DisableSupervisor bool ReconnectOnPublicationError bool ReconnectOnSubscriptionError bool ReconnectOnDataChannelError bool VersionGenerator utils.TimedVersionGenerator - TrackResolver types.MediaTrackResolver DisableDynacast bool SubscriberAllowPause bool SubscriptionLimitAudio int32 @@ -176,6 +173,9 @@ type ParticipantImpl struct { params ParticipantParams + participantHelper atomic.Value // types.LocalParticipantHelper + id atomic.Value // types.ParticipantID + isClosed atomic.Bool closeReason atomic.Value // types.ParticipantCloseReason @@ -294,15 +294,17 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { connectedAt: time.Now().Truncate(time.Millisecond), rttUpdatedAt: time.Now(), cachedDownTracks: make(map[livekit.TrackID]*downTrackState), - dataChannelStats: telemetry.NewBytesTrackStats( - telemetry.BytesTrackIDForParticipantID(telemetry.BytesTrackTypeData, params.SID), - params.SID, - params.Telemetry, - ), - connectionQuality: livekit.ConnectionQuality_EXCELLENT, - pubLogger: params.Logger.WithComponent(sutils.ComponentPub), - subLogger: params.Logger.WithComponent(sutils.ComponentSub), + connectionQuality: livekit.ConnectionQuality_EXCELLENT, + pubLogger: params.Logger.WithComponent(sutils.ComponentPub), + subLogger: params.Logger.WithComponent(sutils.ComponentSub), } + p.id.Store(params.SID) + p.dataChannelStats = telemetry.NewBytesTrackStats( + telemetry.BytesTrackIDForParticipantID(telemetry.BytesTrackTypeData, p.ID()), + p.ID(), + params.Telemetry, + ) + p.participantHelper.Store(params.ParticipantHelper) if !params.DisableSupervisor { p.supervisor = supervisor.NewParticipantSupervisor(supervisor.ParticipantSupervisorParams{Logger: params.Logger}) } @@ -349,6 +351,10 @@ func (p *ParticipantImpl) GetLogger() logger.Logger { return p.params.Logger } +func (p *ParticipantImpl) GetLoggerResolver() logger.DeferredFieldResolver { + return p.params.LoggerResolver +} + func (p *ParticipantImpl) GetAdaptiveStream() bool { return p.params.AdaptiveStream } @@ -362,7 +368,7 @@ func (p *ParticipantImpl) GetDisableSenderReportPassThrough() bool { } func (p *ParticipantImpl) ID() livekit.ParticipantID { - return p.params.SID + return p.id.Load().(livekit.ParticipantID) } func (p *ParticipantImpl) Identity() livekit.ParticipantIdentity { @@ -660,7 +666,7 @@ func (p *ParticipantImpl) ToProtoWithVersion() (*livekit.ParticipantInfo, utils. piv := p.timedVersion pi := &livekit.ParticipantInfo{ - Sid: string(p.params.SID), + Sid: string(p.ID()), Identity: string(p.params.Identity), Name: grants.Name, State: p.State(), @@ -832,6 +838,11 @@ func (p *ParticipantImpl) getOnMetrics() func(types.Participant, *livekit.DataPa } func (p *ParticipantImpl) OnClose(callback func(types.LocalParticipant)) { + if p.isClosed.Load() { + go callback(p) + return + } + p.lock.Lock() p.onClose = callback p.lock.Unlock() @@ -1437,10 +1448,8 @@ func (p *ParticipantImpl) VerifySubscribeParticipantInfo(pID livekit.Participant return } - if f := p.params.GetParticipantInfo; f != nil { - if info := f(pID); info != nil { - _ = p.SendParticipantUpdate([]*livekit.ParticipantInfo{info}) - } + if info := p.helper().GetParticipantInfo(pID); info != nil { + _ = p.SendParticipantUpdate([]*livekit.ParticipantInfo{info}) } } @@ -1616,8 +1625,6 @@ func (p *ParticipantImpl) setupTransportManager() error { } params := TransportManagerParams{ - Identity: p.params.Identity, - SID: p.params.SID, // primary connection does not change, canSubscribe can change if permission was updated // after the participant has joined SubscriberAsPrimary: subscriberAsPrimary, @@ -1697,9 +1704,11 @@ func (p *ParticipantImpl) setupUpTrackManager() { func (p *ParticipantImpl) setupSubscriptionManager() { p.SubscriptionManager = NewSubscriptionManager(SubscriptionManagerParams{ - Participant: p, - Logger: p.subLogger.WithoutSampler(), - TrackResolver: p.params.TrackResolver, + Participant: p, + Logger: p.subLogger.WithoutSampler(), + TrackResolver: func(lp types.LocalParticipant, ti livekit.TrackID) types.MediaResolverResult { + return p.helper().ResolveMediaTrack(lp, ti) + }, Telemetry: p.params.Telemetry, OnTrackSubscribed: p.onTrackSubscribed, OnTrackUnsubscribed: p.onTrackUnsubscribed, @@ -1962,7 +1971,7 @@ func (p *ParticipantImpl) onReceivedDataMessage(kind livekit.DataPacket_Kind, da u.ParticipantSid = "" u.ParticipantIdentity = "" } else { - u.ParticipantSid = string(p.params.SID) + u.ParticipantSid = string(p.ID()) u.ParticipantIdentity = string(p.params.Identity) } if len(dp.DestinationIdentities) != 0 { @@ -2670,7 +2679,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *liv mt := NewMediaTrack(MediaTrackParams{ SignalCid: signalCid, SdpCid: sdpCid, - ParticipantID: p.params.SID, + ParticipantID: p.ID, ParticipantIdentity: p.params.Identity, ParticipantVersion: p.version.Load(), BufferFactory: p.params.Config.BufferFactory, @@ -2685,7 +2694,9 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *liv OnRTCP: p.postRtcp, ForwardStats: p.params.ForwardStats, OnTrackEverSubscribed: p.sendTrackHasBeenSubscribed, - ShouldRegressCodec: p.params.ShouldRegressCodec, + ShouldRegressCodec: func() bool { + return p.helper().ShouldRegressCodec() + }, }, ti) mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) @@ -2891,7 +2902,7 @@ func (p *ParticipantImpl) getPublishedTrackBySdpCid(clientId string) types.Media func (p *ParticipantImpl) DebugInfo() map[string]interface{} { info := map[string]interface{}{ - "ID": p.params.SID, + "ID": p.ID(), "State": p.State().String(), } @@ -2954,16 +2965,14 @@ func (p *ParticipantImpl) setDownTracksConnected() { func (p *ParticipantImpl) cacheForwarderState() { // if migrating in, get forwarder state from migrating out node to facilitate resume - if f := p.params.GetSubscriberForwarderState; f != nil { - if fs, err := f(p); err == nil { - p.lock.Lock() - p.forwarderState = fs - p.lock.Unlock() + if fs, err := p.helper().GetSubscriberForwarderState(p); err == nil && fs != nil { + p.lock.Lock() + p.forwarderState = fs + p.lock.Unlock() - for _, t := range p.SubscriptionManager.GetSubscribedTracks() { - if dt := t.DownTrack(); dt != nil { - dt.SeedState(sfu.DownTrackState{ForwarderState: p.getAndDeleteForwarderState(t.ID())}) - } + for _, t := range p.SubscriptionManager.GetSubscribedTracks() { + if dt := t.DownTrack(); dt != nil { + dt.SeedState(sfu.DownTrackState{ForwarderState: p.getAndDeleteForwarderState(t.ID())}) } } } @@ -3281,3 +3290,43 @@ func (p *ParticipantImpl) HandleMetrics(senderParticipantID livekit.ParticipantI func (p *ParticipantImpl) SupportsCodecChange() bool { return p.params.ClientInfo.SupportsCodecChange() } + +func (p *ParticipantImpl) SupportsMoving() bool { + return p.ProtocolVersion().SupportsMoving() +} + +func (p *ParticipantImpl) MoveToRoom(params types.MoveToRoomParams) { + // fire onClose callback for original room + p.lock.Lock() + onClose := p.onClose + p.onClose = nil + p.lock.Unlock() + if onClose != nil { + onClose(p) + } + + for _, track := range p.GetPublishedTracks() { + trackInfo := track.ToProto() + p.params.Telemetry.TrackUnpublished( + context.Background(), + p.ID(), + p.Identity(), + trackInfo, + true, + ) + } + + p.params.Logger.Infow("move participant to new room", "newRoomName", params.RoomName, "newID", params.ParticipantID) + + p.participantHelper.Store(params.Helper) + p.SubscriptionManager.ClearAllSubscriptions() + p.id.Store(params.ParticipantID) + grants := p.grants.Load().Clone() + grants.Video.Room = string(params.RoomName) + p.grants.Store(grants) + p.params.LoggerResolver("pID", p.ID()) +} + +func (p *ParticipantImpl) helper() types.LocalParticipantHelper { + return p.participantHelper.Load().(types.LocalParticipantHelper) +} diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 1de11ee49..e71fcba84 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -705,6 +705,7 @@ func newParticipantForTestWithOpts(identity livekit.ParticipantIdentity, opts *p Logger: LoggerWithParticipant(logger.GetLogger(), identity, sid, false), Telemetry: &telemetryfakes.FakeTelemetryService{}, VersionGenerator: utils.NewDefaultTimedVersionGenerator(), + ParticipantHelper: &typesfakes.FakeLocalParticipantHelper{}, }) p.isPublisher.Store(opts.publisher) p.updateState(livekit.ParticipantInfo_ACTIVE) diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 81fda0a59..75733e633 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -111,7 +111,7 @@ func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit. isValid = false } } - if pi.Permission != nil && pi.Permission.Hidden && pi.Sid != string(p.params.SID) { + if pi.Permission != nil && pi.Permission.Hidden && pi.Sid != string(p.ID()) { p.params.Logger.Debugw("skipping hidden participant update", "otherParticipant", pi.Identity) isValid = false } @@ -213,6 +213,14 @@ func (p *ParticipantImpl) SendRequestResponse(requestResponse *livekit.RequestRe }) } +func (p *ParticipantImpl) SendRoomMovedResponse(roomMovedResponse *livekit.RoomMovedResponse) error { + return p.writeMessage(&livekit.SignalResponse{ + Message: &livekit.SignalResponse_RoomMoved{ + RoomMoved: roomMovedResponse, + }, + }) +} + func (p *ParticipantImpl) HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error { p.TransportManager.HandleClientReconnect(reconnectReason) @@ -377,9 +385,9 @@ func (p *ParticipantImpl) sendLeaveRequest( default: leave.Action = livekit.LeaveRequest_DISCONNECT } - if leave.Action != livekit.LeaveRequest_DISCONNECT && p.params.GetRegionSettings != nil { + if leave.Action != livekit.LeaveRequest_DISCONNECT { // sending region settings even for RESUME just in case client wants to a full reconnect despite server saying RESUME - leave.Regions = p.params.GetRegionSettings(p.params.ClientInfo.Address) + leave.Regions = p.helper().GetRegionSettings(p.params.ClientInfo.Address) } } else { if !sendOnlyIfSupportingLeaveRequestWithAction { diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 2d41f37c3..e680e3bfe 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -181,6 +181,23 @@ func (m *SubscriptionManager) UnsubscribeFromTrack(trackID livekit.TrackID) { m.queueReconcile(trackID) } +func (m *SubscriptionManager) ClearAllSubscriptions() { + m.params.Logger.Debugw("clearing all subscriptions") + + if m.params.UseOneShotSignallingMode { + for _, track := range m.GetSubscribedTracks() { + m.unsubscribeSynchronous(track.ID()) + } + } + + m.lock.RLock() + for _, sub := range m.subscriptions { + sub.setDesired(false) + } + m.lock.RUnlock() + m.ReconcileAll() +} + func (m *SubscriptionManager) GetSubscribedTracks() []types.SubscribedTrack { m.lock.RLock() defer m.lock.RUnlock() @@ -356,7 +373,7 @@ func (m *SubscriptionManager) reconcileSubscription(s *trackSubscription) { if numAttempts == 0 { m.params.Telemetry.TrackSubscribeRequested( context.Background(), - m.params.Participant.ID(), + s.subscriberID, &livekit.TrackInfo{ Sid: string(s.trackID), }, @@ -375,14 +392,14 @@ func (m *SubscriptionManager) reconcileSubscription(s *trackSubscription) { // - ErrSubscriptionLimitExceeded: the participant have reached the limit of subscriptions, wait for the other subscription to be unsubscribed // We'll still log an event to reflect this in telemetry since it's been too long if s.durationSinceStart() > subscriptionTimeout { - s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), err, true) + s.maybeRecordError(m.params.Telemetry, s.subscriberID, err, true) } case ErrTrackNotFound: // source track was never published or closed // if after timeout we'd unsubscribe from it. // this is the *only* case we'd change desired state if s.durationSinceStart() > notFoundTimeout { - s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), err, true) + s.maybeRecordError(m.params.Telemetry, s.subscriberID, err, true) s.logger.Infow("unsubscribing from track after notFoundTimeout", "error", err) s.setDesired(false) m.queueReconcile(s.trackID) @@ -394,7 +411,7 @@ func (m *SubscriptionManager) reconcileSubscription(s *trackSubscription) { s.logger.Warnw("failed to subscribe, triggering error handler", err, "attempt", numAttempts, ) - s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), err, false) + s.maybeRecordError(m.params.Telemetry, s.subscriberID, err, false) m.params.OnSubscriptionError(s.trackID, true, err) } else { s.logger.Debugw("failed to subscribe, retrying", @@ -430,7 +447,7 @@ func (m *SubscriptionManager) reconcileSubscription(s *trackSubscription) { // back to needsSubscribe state if s.durationSinceStart() > subscriptionTimeout { s.logger.Warnw("track not bound after timeout", nil) - s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), ErrTrackNotBound, false) + s.maybeRecordError(m.params.Telemetry, s.subscriberID, ErrTrackNotBound, false) m.params.OnSubscriptionError(s.trackID, true, ErrTrackNotBound) } } @@ -510,12 +527,12 @@ func (m *SubscriptionManager) subscribe(s *trackSubscription) error { // set callback only when we haven't done it before // we set the observer before checking for existence of track, so that we may get notified // when the track becomes available - res.TrackChangedNotifier.AddObserver(string(m.params.Participant.ID()), func() { + res.TrackChangedNotifier.AddObserver(string(s.subscriberID), func() { m.queueReconcile(trackID) }) } if res.TrackRemovedNotifier != nil && s.setRemovedNotifier(res.TrackRemovedNotifier) { - res.TrackRemovedNotifier.AddObserver(string(m.params.Participant.ID()), func() { + res.TrackRemovedNotifier.AddObserver(string(s.subscriberID), func() { // re-resolve the track in case the same track had been re-published res := m.params.TrackResolver(m.params.Participant, trackID) if res.Track != nil { @@ -569,13 +586,13 @@ func (m *SubscriptionManager) subscribe(s *trackSubscription) error { subTrack.AddOnBind(func(err error) { if err != nil { s.logger.Infow("failed to bind track", "err", err) - s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), err, true) + s.maybeRecordError(m.params.Telemetry, s.subscriberID, err, true) m.UnsubscribeFromTrack(trackID) m.params.OnSubscriptionError(trackID, false, err) return } s.setBound() - s.maybeRecordSuccess(m.params.Telemetry, m.params.Participant.ID()) + s.maybeRecordSuccess(m.params.Telemetry, s.subscriberID) }) s.setSubscribedTrack(subTrack) @@ -670,13 +687,13 @@ func (m *SubscriptionManager) subscribeSynchronous(trackID livekit.TrackID) erro subTrack.AddOnBind(func(err error) { if err != nil { sub.logger.Infow("failed to bind track", "err", err) - sub.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), err, true) + sub.maybeRecordError(m.params.Telemetry, sub.subscriberID, err, true) m.UnsubscribeFromTrack(trackID) m.params.OnSubscriptionError(trackID, false, err) return } sub.setBound() - sub.maybeRecordSuccess(m.params.Telemetry, m.params.Participant.ID()) + sub.maybeRecordSuccess(m.params.Telemetry, sub.subscriberID) }) sub.setSubscribedTrack(subTrack) @@ -709,7 +726,7 @@ func (m *SubscriptionManager) unsubscribe(s *trackSubscription) error { } track := subTrack.MediaTrack() - pID := m.params.Participant.ID() + pID := s.subscriberID m.pendingUnsubscribes.Inc() go func() { defer m.pendingUnsubscribes.Dec() @@ -738,7 +755,7 @@ func (m *SubscriptionManager) unsubscribeSynchronous(trackID livekit.TrackID) er } track := subTrack.MediaTrack() - track.RemoveSubscriber(m.params.Participant.ID(), false) + track.RemoveSubscriber(sub.subscriberID, false) return nil } @@ -806,7 +823,7 @@ func (m *SubscriptionManager) handleSubscribedTrackClose(s *trackSubscription, i if wasBound { m.params.Telemetry.TrackUnsubscribed( context.Background(), - m.params.Participant.ID(), + s.subscriberID, &livekit.TrackInfo{Sid: string(s.trackID), Type: subTrack.MediaTrack().Kind()}, !isExpectedToResume, ) @@ -817,7 +834,7 @@ func (m *SubscriptionManager) handleSubscribedTrackClose(s *trackSubscription, i if stats != nil { m.params.Telemetry.TrackSubscribeRTPStats( context.Background(), - m.params.Participant.ID(), + s.subscriberID, s.trackID, dt.Mime(), stats, diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index db877c5ac..4744a2dad 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -256,8 +256,6 @@ type PCTransport struct { type TransportParams struct { Handler transport.Handler - ParticipantID livekit.ParticipantID - ParticipantIdentity livekit.ParticipantIdentity ProtocolVersion types.ProtocolVersion Config *WebRTCConfig Twcc *lktwcc.Responder diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index ac0c1cc97..fb4ee17e9 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -35,10 +35,8 @@ import ( func TestMissingAnswerDuringICERestart(t *testing.T) { params := TransportParams{ - ParticipantID: "id", - ParticipantIdentity: "identity", - Config: &WebRTCConfig{}, - IsOfferer: true, + Config: &WebRTCConfig{}, + IsOfferer: true, } paramsA := params @@ -91,10 +89,8 @@ func TestMissingAnswerDuringICERestart(t *testing.T) { func TestNegotiationTiming(t *testing.T) { params := TransportParams{ - ParticipantID: "id", - ParticipantIdentity: "identity", - Config: &WebRTCConfig{}, - IsOfferer: true, + Config: &WebRTCConfig{}, + IsOfferer: true, } paramsA := params @@ -190,10 +186,8 @@ func TestNegotiationTiming(t *testing.T) { func TestFirstOfferMissedDuringICERestart(t *testing.T) { params := TransportParams{ - ParticipantID: "id", - ParticipantIdentity: "identity", - Config: &WebRTCConfig{}, - IsOfferer: true, + Config: &WebRTCConfig{}, + IsOfferer: true, } paramsA := params @@ -264,10 +258,8 @@ func TestFirstOfferMissedDuringICERestart(t *testing.T) { func TestFirstAnswerMissedDuringICERestart(t *testing.T) { params := TransportParams{ - ParticipantID: "id", - ParticipantIdentity: "identity", - Config: &WebRTCConfig{}, - IsOfferer: true, + Config: &WebRTCConfig{}, + IsOfferer: true, } paramsA := params @@ -343,10 +335,8 @@ func TestFirstAnswerMissedDuringICERestart(t *testing.T) { func TestNegotiationFailed(t *testing.T) { params := TransportParams{ - ParticipantID: "id", - ParticipantIdentity: "identity", - Config: &WebRTCConfig{}, - IsOfferer: true, + Config: &WebRTCConfig{}, + IsOfferer: true, } paramsA := params @@ -386,9 +376,7 @@ func TestNegotiationFailed(t *testing.T) { func TestFilteringCandidates(t *testing.T) { params := TransportParams{ - ParticipantID: "id", - ParticipantIdentity: "identity", - Config: &WebRTCConfig{}, + Config: &WebRTCConfig{}, EnabledCodecs: []*livekit.Codec{ {Mime: mime.MimeTypeOpus.String()}, {Mime: mime.MimeTypeVP8.String()}, diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 719dda382..0bb6960ae 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -83,8 +83,6 @@ func (h TransportManagerPublisherTransportHandler) OnAnswer(sd webrtc.SessionDes // ------------------------------- type TransportManagerParams struct { - Identity livekit.ParticipantIdentity - SID livekit.ParticipantID SubscriberAsPrimary bool Config *WebRTCConfig Twcc *twcc.Responder @@ -151,8 +149,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER) publisher, err := NewPCTransport(TransportParams{ - ParticipantID: params.SID, - ParticipantIdentity: params.Identity, ProtocolVersion: params.ProtocolVersion, Config: params.Config, Twcc: params.Twcc, @@ -176,8 +172,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro lgr = LoggerWithPCTarget(params.Logger, livekit.SignalTarget_SUBSCRIBER) subscriber, err := NewPCTransport(TransportParams{ - ParticipantID: params.SID, - ParticipantIdentity: params.Identity, ProtocolVersion: params.ProtocolVersion, Config: params.Config, DirectionConfig: params.Config.Subscriber, diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 4741bcf30..d876cba67 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -111,6 +111,7 @@ const ( ParticipantCloseReasonRoomClosed ParticipantCloseReasonUserUnavailable ParticipantCloseReasonUserRejected + ParticipantCloseReasonMoveFailed ) func (p ParticipantCloseReason) String() string { @@ -169,6 +170,8 @@ func (p ParticipantCloseReason) String() string { return "USER_UNAVAILABLE" case ParticipantCloseReasonUserRejected: return "USER_REJECTED" + case ParticipantCloseReasonMoveFailed: + return "MOVE_FAILED" default: return fmt.Sprintf("%d", int(p)) } @@ -195,7 +198,8 @@ func (p ParticipantCloseReason) ToDisconnectReason() livekit.DisconnectReason { return livekit.DisconnectReason_ROOM_DELETED case ParticipantCloseReasonSimulateNodeFailure, ParticipantCloseReasonSimulateServerLeave: return livekit.DisconnectReason_SERVER_SHUTDOWN - case ParticipantCloseReasonNegotiateFailed, ParticipantCloseReasonPublicationError, ParticipantCloseReasonSubscriptionError, ParticipantCloseReasonDataChannelError, ParticipantCloseReasonMigrateCodecMismatch: + case ParticipantCloseReasonNegotiateFailed, ParticipantCloseReasonPublicationError, ParticipantCloseReasonSubscriptionError, + ParticipantCloseReasonDataChannelError, ParticipantCloseReasonMigrateCodecMismatch, ParticipantCloseReasonMoveFailed: return livekit.DisconnectReason_STATE_MISMATCH case ParticipantCloseReasonSignalSourceClose: return livekit.DisconnectReason_SIGNAL_CLOSE @@ -313,6 +317,21 @@ type AddTrackParams struct { Red bool } +type MoveToRoomParams struct { + RoomName livekit.RoomName + ParticipantID livekit.ParticipantID + Helper LocalParticipantHelper +} + +//counterfeiter:generate . LocalParticipantHelper +type LocalParticipantHelper interface { + ResolveMediaTrack(LocalParticipant, livekit.TrackID) MediaResolverResult + GetParticipantInfo(pID livekit.ParticipantID) *livekit.ParticipantInfo + GetRegionSettings(ip string) *livekit.RegionSettings + GetSubscriberForwarderState(p LocalParticipant) (map[livekit.TrackID]*livekit.RTPForwarderState, error) + ShouldRegressCodec() bool +} + //counterfeiter:generate . LocalParticipant type LocalParticipant interface { Participant @@ -322,6 +341,7 @@ type LocalParticipant interface { // getters GetTrailer() []byte GetLogger() logger.Logger + GetLoggerResolver() logger.DeferredFieldResolver GetAdaptiveStream() bool ProtocolVersion() ProtocolVersion SupportsSyncStreamID() bool @@ -341,6 +361,7 @@ type LocalParticipant interface { HasConnected() bool GetEnabledPublishCodecs() []*livekit.Codec GetPublisherICESessionUfrag() (string, error) + SupportsMoving() bool SetResponseSink(sink routing.MessageSink) CloseSignalConnection(reason SignallingCloseReason) @@ -415,6 +436,7 @@ type LocalParticipant interface { SendRequestResponse(requestResponse *livekit.RequestResponse) error HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error IssueFullReconnect(reason ParticipantCloseReason) + SendRoomMovedResponse(moved *livekit.RoomMovedResponse) error // callbacks OnStateChange(func(p LocalParticipant)) @@ -447,6 +469,7 @@ type LocalParticipant interface { dataChannels []*livekit.DataChannelInfo, ) IsReconnect() bool + MoveToRoom(params MoveToRoomParams) UpdateMediaRTT(rtt uint32) UpdateSignalingRTT(rtt uint32) @@ -507,6 +530,7 @@ type MediaTrack interface { PublisherID() livekit.ParticipantID PublisherIdentity() livekit.ParticipantIdentity PublisherVersion() uint32 + Logger() logger.Logger IsMuted() bool SetMuted(muted bool) diff --git a/pkg/rtc/types/protocol_version.go b/pkg/rtc/types/protocol_version.go index 79b2d71e8..a6a7e2056 100644 --- a/pkg/rtc/types/protocol_version.go +++ b/pkg/rtc/types/protocol_version.go @@ -95,3 +95,7 @@ func (v ProtocolVersion) SupportsRegionsInLeaveRequest() bool { func (v ProtocolVersion) SupportsNonErrorSignalResponse() bool { return v > 14 } + +func (v ProtocolVersion) SupportsMoving() bool { + return v > 15 +} diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index 877267f6c..9445bc8fa 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -8,6 +8,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/mime" "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" ) type FakeLocalMediaTrack struct { @@ -200,6 +201,16 @@ type FakeLocalMediaTrack struct { kindReturnsOnCall map[int]struct { result1 livekit.TrackType } + LoggerStub func() logger.Logger + loggerMutex sync.RWMutex + loggerArgsForCall []struct { + } + loggerReturns struct { + result1 logger.Logger + } + loggerReturnsOnCall map[int]struct { + result1 logger.Logger + } NameStub func() string nameMutex sync.RWMutex nameArgsForCall []struct { @@ -1352,6 +1363,59 @@ func (fake *FakeLocalMediaTrack) KindReturnsOnCall(i int, result1 livekit.TrackT }{result1} } +func (fake *FakeLocalMediaTrack) Logger() logger.Logger { + fake.loggerMutex.Lock() + ret, specificReturn := fake.loggerReturnsOnCall[len(fake.loggerArgsForCall)] + fake.loggerArgsForCall = append(fake.loggerArgsForCall, struct { + }{}) + stub := fake.LoggerStub + fakeReturns := fake.loggerReturns + fake.recordInvocation("Logger", []interface{}{}) + fake.loggerMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalMediaTrack) LoggerCallCount() int { + fake.loggerMutex.RLock() + defer fake.loggerMutex.RUnlock() + return len(fake.loggerArgsForCall) +} + +func (fake *FakeLocalMediaTrack) LoggerCalls(stub func() logger.Logger) { + fake.loggerMutex.Lock() + defer fake.loggerMutex.Unlock() + fake.LoggerStub = stub +} + +func (fake *FakeLocalMediaTrack) LoggerReturns(result1 logger.Logger) { + fake.loggerMutex.Lock() + defer fake.loggerMutex.Unlock() + fake.LoggerStub = nil + fake.loggerReturns = struct { + result1 logger.Logger + }{result1} +} + +func (fake *FakeLocalMediaTrack) LoggerReturnsOnCall(i int, result1 logger.Logger) { + fake.loggerMutex.Lock() + defer fake.loggerMutex.Unlock() + fake.LoggerStub = nil + if fake.loggerReturnsOnCall == nil { + fake.loggerReturnsOnCall = make(map[int]struct { + result1 logger.Logger + }) + } + fake.loggerReturnsOnCall[i] = struct { + result1 logger.Logger + }{result1} +} + func (fake *FakeLocalMediaTrack) Name() string { fake.nameMutex.Lock() ret, specificReturn := fake.nameReturnsOnCall[len(fake.nameArgsForCall)] @@ -2248,6 +2312,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.isSubscriberMutex.RUnlock() fake.kindMutex.RLock() defer fake.kindMutex.RUnlock() + fake.loggerMutex.RLock() + defer fake.loggerMutex.RUnlock() fake.nameMutex.RLock() defer fake.nameMutex.RUnlock() fake.notifySubscriberNodeMaxQualityMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 49232f2e0..7cc24d5fa 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -338,6 +338,16 @@ type FakeLocalParticipant struct { getLoggerReturnsOnCall map[int]struct { result1 logger.Logger } + GetLoggerResolverStub func() logger.DeferredFieldResolver + getLoggerResolverMutex sync.RWMutex + getLoggerResolverArgsForCall []struct { + } + getLoggerResolverReturns struct { + result1 logger.DeferredFieldResolver + } + getLoggerResolverReturnsOnCall map[int]struct { + result1 logger.DeferredFieldResolver + } GetPacerStub func() pacer.Pacer getPacerMutex sync.RWMutex getPacerArgsForCall []struct { @@ -713,6 +723,11 @@ type FakeLocalParticipant struct { migrateStateReturnsOnCall map[int]struct { result1 types.MigrateState } + MoveToRoomStub func(types.MoveToRoomParams) + moveToRoomMutex sync.RWMutex + moveToRoomArgsForCall []struct { + arg1 types.MoveToRoomParams + } NegotiateStub func(bool) negotiateMutex sync.RWMutex negotiateArgsForCall []struct { @@ -900,6 +915,17 @@ type FakeLocalParticipant struct { sendRequestResponseReturnsOnCall map[int]struct { result1 error } + SendRoomMovedResponseStub func(*livekit.RoomMovedResponse) error + sendRoomMovedResponseMutex sync.RWMutex + sendRoomMovedResponseArgsForCall []struct { + arg1 *livekit.RoomMovedResponse + } + sendRoomMovedResponseReturns struct { + result1 error + } + sendRoomMovedResponseReturnsOnCall map[int]struct { + result1 error + } SendRoomUpdateStub func(*livekit.Room) error sendRoomUpdateMutex sync.RWMutex sendRoomUpdateArgsForCall []struct { @@ -1064,6 +1090,16 @@ type FakeLocalParticipant struct { supportsCodecChangeReturnsOnCall map[int]struct { result1 bool } + SupportsMovingStub func() bool + supportsMovingMutex sync.RWMutex + supportsMovingArgsForCall []struct { + } + supportsMovingReturns struct { + result1 bool + } + supportsMovingReturnsOnCall map[int]struct { + result1 bool + } SupportsSyncStreamIDStub func() bool supportsSyncStreamIDMutex sync.RWMutex supportsSyncStreamIDArgsForCall []struct { @@ -2886,6 +2922,59 @@ func (fake *FakeLocalParticipant) GetLoggerReturnsOnCall(i int, result1 logger.L }{result1} } +func (fake *FakeLocalParticipant) GetLoggerResolver() logger.DeferredFieldResolver { + fake.getLoggerResolverMutex.Lock() + ret, specificReturn := fake.getLoggerResolverReturnsOnCall[len(fake.getLoggerResolverArgsForCall)] + fake.getLoggerResolverArgsForCall = append(fake.getLoggerResolverArgsForCall, struct { + }{}) + stub := fake.GetLoggerResolverStub + fakeReturns := fake.getLoggerResolverReturns + fake.recordInvocation("GetLoggerResolver", []interface{}{}) + fake.getLoggerResolverMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetLoggerResolverCallCount() int { + fake.getLoggerResolverMutex.RLock() + defer fake.getLoggerResolverMutex.RUnlock() + return len(fake.getLoggerResolverArgsForCall) +} + +func (fake *FakeLocalParticipant) GetLoggerResolverCalls(stub func() logger.DeferredFieldResolver) { + fake.getLoggerResolverMutex.Lock() + defer fake.getLoggerResolverMutex.Unlock() + fake.GetLoggerResolverStub = stub +} + +func (fake *FakeLocalParticipant) GetLoggerResolverReturns(result1 logger.DeferredFieldResolver) { + fake.getLoggerResolverMutex.Lock() + defer fake.getLoggerResolverMutex.Unlock() + fake.GetLoggerResolverStub = nil + fake.getLoggerResolverReturns = struct { + result1 logger.DeferredFieldResolver + }{result1} +} + +func (fake *FakeLocalParticipant) GetLoggerResolverReturnsOnCall(i int, result1 logger.DeferredFieldResolver) { + fake.getLoggerResolverMutex.Lock() + defer fake.getLoggerResolverMutex.Unlock() + fake.GetLoggerResolverStub = nil + if fake.getLoggerResolverReturnsOnCall == nil { + fake.getLoggerResolverReturnsOnCall = make(map[int]struct { + result1 logger.DeferredFieldResolver + }) + } + fake.getLoggerResolverReturnsOnCall[i] = struct { + result1 logger.DeferredFieldResolver + }{result1} +} + func (fake *FakeLocalParticipant) GetPacer() pacer.Pacer { fake.getPacerMutex.Lock() ret, specificReturn := fake.getPacerReturnsOnCall[len(fake.getPacerArgsForCall)] @@ -4887,6 +4976,38 @@ func (fake *FakeLocalParticipant) MigrateStateReturnsOnCall(i int, result1 types }{result1} } +func (fake *FakeLocalParticipant) MoveToRoom(arg1 types.MoveToRoomParams) { + fake.moveToRoomMutex.Lock() + fake.moveToRoomArgsForCall = append(fake.moveToRoomArgsForCall, struct { + arg1 types.MoveToRoomParams + }{arg1}) + stub := fake.MoveToRoomStub + fake.recordInvocation("MoveToRoom", []interface{}{arg1}) + fake.moveToRoomMutex.Unlock() + if stub != nil { + fake.MoveToRoomStub(arg1) + } +} + +func (fake *FakeLocalParticipant) MoveToRoomCallCount() int { + fake.moveToRoomMutex.RLock() + defer fake.moveToRoomMutex.RUnlock() + return len(fake.moveToRoomArgsForCall) +} + +func (fake *FakeLocalParticipant) MoveToRoomCalls(stub func(types.MoveToRoomParams)) { + fake.moveToRoomMutex.Lock() + defer fake.moveToRoomMutex.Unlock() + fake.MoveToRoomStub = stub +} + +func (fake *FakeLocalParticipant) MoveToRoomArgsForCall(i int) types.MoveToRoomParams { + fake.moveToRoomMutex.RLock() + defer fake.moveToRoomMutex.RUnlock() + argsForCall := fake.moveToRoomArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) Negotiate(arg1 bool) { fake.negotiateMutex.Lock() fake.negotiateArgsForCall = append(fake.negotiateArgsForCall, struct { @@ -5984,6 +6105,67 @@ func (fake *FakeLocalParticipant) SendRequestResponseReturnsOnCall(i int, result }{result1} } +func (fake *FakeLocalParticipant) SendRoomMovedResponse(arg1 *livekit.RoomMovedResponse) error { + fake.sendRoomMovedResponseMutex.Lock() + ret, specificReturn := fake.sendRoomMovedResponseReturnsOnCall[len(fake.sendRoomMovedResponseArgsForCall)] + fake.sendRoomMovedResponseArgsForCall = append(fake.sendRoomMovedResponseArgsForCall, struct { + arg1 *livekit.RoomMovedResponse + }{arg1}) + stub := fake.SendRoomMovedResponseStub + fakeReturns := fake.sendRoomMovedResponseReturns + fake.recordInvocation("SendRoomMovedResponse", []interface{}{arg1}) + fake.sendRoomMovedResponseMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) SendRoomMovedResponseCallCount() int { + fake.sendRoomMovedResponseMutex.RLock() + defer fake.sendRoomMovedResponseMutex.RUnlock() + return len(fake.sendRoomMovedResponseArgsForCall) +} + +func (fake *FakeLocalParticipant) SendRoomMovedResponseCalls(stub func(*livekit.RoomMovedResponse) error) { + fake.sendRoomMovedResponseMutex.Lock() + defer fake.sendRoomMovedResponseMutex.Unlock() + fake.SendRoomMovedResponseStub = stub +} + +func (fake *FakeLocalParticipant) SendRoomMovedResponseArgsForCall(i int) *livekit.RoomMovedResponse { + fake.sendRoomMovedResponseMutex.RLock() + defer fake.sendRoomMovedResponseMutex.RUnlock() + argsForCall := fake.sendRoomMovedResponseArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) SendRoomMovedResponseReturns(result1 error) { + fake.sendRoomMovedResponseMutex.Lock() + defer fake.sendRoomMovedResponseMutex.Unlock() + fake.SendRoomMovedResponseStub = nil + fake.sendRoomMovedResponseReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeLocalParticipant) SendRoomMovedResponseReturnsOnCall(i int, result1 error) { + fake.sendRoomMovedResponseMutex.Lock() + defer fake.sendRoomMovedResponseMutex.Unlock() + fake.SendRoomMovedResponseStub = nil + if fake.sendRoomMovedResponseReturnsOnCall == nil { + fake.sendRoomMovedResponseReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.sendRoomMovedResponseReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeLocalParticipant) SendRoomUpdate(arg1 *livekit.Room) error { fake.sendRoomUpdateMutex.Lock() ret, specificReturn := fake.sendRoomUpdateReturnsOnCall[len(fake.sendRoomUpdateArgsForCall)] @@ -6903,6 +7085,59 @@ func (fake *FakeLocalParticipant) SupportsCodecChangeReturnsOnCall(i int, result }{result1} } +func (fake *FakeLocalParticipant) SupportsMoving() bool { + fake.supportsMovingMutex.Lock() + ret, specificReturn := fake.supportsMovingReturnsOnCall[len(fake.supportsMovingArgsForCall)] + fake.supportsMovingArgsForCall = append(fake.supportsMovingArgsForCall, struct { + }{}) + stub := fake.SupportsMovingStub + fakeReturns := fake.supportsMovingReturns + fake.recordInvocation("SupportsMoving", []interface{}{}) + fake.supportsMovingMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) SupportsMovingCallCount() int { + fake.supportsMovingMutex.RLock() + defer fake.supportsMovingMutex.RUnlock() + return len(fake.supportsMovingArgsForCall) +} + +func (fake *FakeLocalParticipant) SupportsMovingCalls(stub func() bool) { + fake.supportsMovingMutex.Lock() + defer fake.supportsMovingMutex.Unlock() + fake.SupportsMovingStub = stub +} + +func (fake *FakeLocalParticipant) SupportsMovingReturns(result1 bool) { + fake.supportsMovingMutex.Lock() + defer fake.supportsMovingMutex.Unlock() + fake.SupportsMovingStub = nil + fake.supportsMovingReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) SupportsMovingReturnsOnCall(i int, result1 bool) { + fake.supportsMovingMutex.Lock() + defer fake.supportsMovingMutex.Unlock() + fake.SupportsMovingStub = nil + if fake.supportsMovingReturnsOnCall == nil { + fake.supportsMovingReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.supportsMovingReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) SupportsSyncStreamID() bool { fake.supportsSyncStreamIDMutex.Lock() ret, specificReturn := fake.supportsSyncStreamIDReturnsOnCall[len(fake.supportsSyncStreamIDArgsForCall)] @@ -7950,6 +8185,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getICEConnectionInfoMutex.RUnlock() fake.getLoggerMutex.RLock() defer fake.getLoggerMutex.RUnlock() + fake.getLoggerResolverMutex.RLock() + defer fake.getLoggerResolverMutex.RUnlock() fake.getPacerMutex.RLock() defer fake.getPacerMutex.RUnlock() fake.getPendingTrackMutex.RLock() @@ -8026,6 +8263,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.maybeStartMigrationMutex.RUnlock() fake.migrateStateMutex.RLock() defer fake.migrateStateMutex.RUnlock() + fake.moveToRoomMutex.RLock() + defer fake.moveToRoomMutex.RUnlock() fake.negotiateMutex.RLock() defer fake.negotiateMutex.RUnlock() fake.notifyMigrationMutex.RLock() @@ -8078,6 +8317,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.sendRefreshTokenMutex.RUnlock() fake.sendRequestResponseMutex.RLock() defer fake.sendRequestResponseMutex.RUnlock() + fake.sendRoomMovedResponseMutex.RLock() + defer fake.sendRoomMovedResponseMutex.RUnlock() fake.sendRoomUpdateMutex.RLock() defer fake.sendRoomUpdateMutex.RUnlock() fake.sendSpeakerUpdateMutex.RLock() @@ -8120,6 +8361,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.subscriptionPermissionUpdateMutex.RUnlock() fake.supportsCodecChangeMutex.RLock() defer fake.supportsCodecChangeMutex.RUnlock() + fake.supportsMovingMutex.RLock() + defer fake.supportsMovingMutex.RUnlock() fake.supportsSyncStreamIDMutex.RLock() defer fake.supportsSyncStreamIDMutex.RUnlock() fake.supportsTransceiverReuseMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant_helper.go b/pkg/rtc/types/typesfakes/fake_local_participant_helper.go new file mode 100644 index 000000000..6e7d8ea77 --- /dev/null +++ b/pkg/rtc/types/typesfakes/fake_local_participant_helper.go @@ -0,0 +1,406 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package typesfakes + +import ( + "sync" + + "github.com/livekit/livekit-server/pkg/rtc/types" + "github.com/livekit/protocol/livekit" +) + +type FakeLocalParticipantHelper struct { + GetParticipantInfoStub func(livekit.ParticipantID) *livekit.ParticipantInfo + getParticipantInfoMutex sync.RWMutex + getParticipantInfoArgsForCall []struct { + arg1 livekit.ParticipantID + } + getParticipantInfoReturns struct { + result1 *livekit.ParticipantInfo + } + getParticipantInfoReturnsOnCall map[int]struct { + result1 *livekit.ParticipantInfo + } + GetRegionSettingsStub func(string) *livekit.RegionSettings + getRegionSettingsMutex sync.RWMutex + getRegionSettingsArgsForCall []struct { + arg1 string + } + getRegionSettingsReturns struct { + result1 *livekit.RegionSettings + } + getRegionSettingsReturnsOnCall map[int]struct { + result1 *livekit.RegionSettings + } + GetSubscriberForwarderStateStub func(types.LocalParticipant) (map[livekit.TrackID]*livekit.RTPForwarderState, error) + getSubscriberForwarderStateMutex sync.RWMutex + getSubscriberForwarderStateArgsForCall []struct { + arg1 types.LocalParticipant + } + getSubscriberForwarderStateReturns struct { + result1 map[livekit.TrackID]*livekit.RTPForwarderState + result2 error + } + getSubscriberForwarderStateReturnsOnCall map[int]struct { + result1 map[livekit.TrackID]*livekit.RTPForwarderState + result2 error + } + ResolveMediaTrackStub func(types.LocalParticipant, livekit.TrackID) types.MediaResolverResult + resolveMediaTrackMutex sync.RWMutex + resolveMediaTrackArgsForCall []struct { + arg1 types.LocalParticipant + arg2 livekit.TrackID + } + resolveMediaTrackReturns struct { + result1 types.MediaResolverResult + } + resolveMediaTrackReturnsOnCall map[int]struct { + result1 types.MediaResolverResult + } + ShouldRegressCodecStub func() bool + shouldRegressCodecMutex sync.RWMutex + shouldRegressCodecArgsForCall []struct { + } + shouldRegressCodecReturns struct { + result1 bool + } + shouldRegressCodecReturnsOnCall map[int]struct { + result1 bool + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeLocalParticipantHelper) GetParticipantInfo(arg1 livekit.ParticipantID) *livekit.ParticipantInfo { + fake.getParticipantInfoMutex.Lock() + ret, specificReturn := fake.getParticipantInfoReturnsOnCall[len(fake.getParticipantInfoArgsForCall)] + fake.getParticipantInfoArgsForCall = append(fake.getParticipantInfoArgsForCall, struct { + arg1 livekit.ParticipantID + }{arg1}) + stub := fake.GetParticipantInfoStub + fakeReturns := fake.getParticipantInfoReturns + fake.recordInvocation("GetParticipantInfo", []interface{}{arg1}) + fake.getParticipantInfoMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipantHelper) GetParticipantInfoCallCount() int { + fake.getParticipantInfoMutex.RLock() + defer fake.getParticipantInfoMutex.RUnlock() + return len(fake.getParticipantInfoArgsForCall) +} + +func (fake *FakeLocalParticipantHelper) GetParticipantInfoCalls(stub func(livekit.ParticipantID) *livekit.ParticipantInfo) { + fake.getParticipantInfoMutex.Lock() + defer fake.getParticipantInfoMutex.Unlock() + fake.GetParticipantInfoStub = stub +} + +func (fake *FakeLocalParticipantHelper) GetParticipantInfoArgsForCall(i int) livekit.ParticipantID { + fake.getParticipantInfoMutex.RLock() + defer fake.getParticipantInfoMutex.RUnlock() + argsForCall := fake.getParticipantInfoArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipantHelper) GetParticipantInfoReturns(result1 *livekit.ParticipantInfo) { + fake.getParticipantInfoMutex.Lock() + defer fake.getParticipantInfoMutex.Unlock() + fake.GetParticipantInfoStub = nil + fake.getParticipantInfoReturns = struct { + result1 *livekit.ParticipantInfo + }{result1} +} + +func (fake *FakeLocalParticipantHelper) GetParticipantInfoReturnsOnCall(i int, result1 *livekit.ParticipantInfo) { + fake.getParticipantInfoMutex.Lock() + defer fake.getParticipantInfoMutex.Unlock() + fake.GetParticipantInfoStub = nil + if fake.getParticipantInfoReturnsOnCall == nil { + fake.getParticipantInfoReturnsOnCall = make(map[int]struct { + result1 *livekit.ParticipantInfo + }) + } + fake.getParticipantInfoReturnsOnCall[i] = struct { + result1 *livekit.ParticipantInfo + }{result1} +} + +func (fake *FakeLocalParticipantHelper) GetRegionSettings(arg1 string) *livekit.RegionSettings { + fake.getRegionSettingsMutex.Lock() + ret, specificReturn := fake.getRegionSettingsReturnsOnCall[len(fake.getRegionSettingsArgsForCall)] + fake.getRegionSettingsArgsForCall = append(fake.getRegionSettingsArgsForCall, struct { + arg1 string + }{arg1}) + stub := fake.GetRegionSettingsStub + fakeReturns := fake.getRegionSettingsReturns + fake.recordInvocation("GetRegionSettings", []interface{}{arg1}) + fake.getRegionSettingsMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipantHelper) GetRegionSettingsCallCount() int { + fake.getRegionSettingsMutex.RLock() + defer fake.getRegionSettingsMutex.RUnlock() + return len(fake.getRegionSettingsArgsForCall) +} + +func (fake *FakeLocalParticipantHelper) GetRegionSettingsCalls(stub func(string) *livekit.RegionSettings) { + fake.getRegionSettingsMutex.Lock() + defer fake.getRegionSettingsMutex.Unlock() + fake.GetRegionSettingsStub = stub +} + +func (fake *FakeLocalParticipantHelper) GetRegionSettingsArgsForCall(i int) string { + fake.getRegionSettingsMutex.RLock() + defer fake.getRegionSettingsMutex.RUnlock() + argsForCall := fake.getRegionSettingsArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipantHelper) GetRegionSettingsReturns(result1 *livekit.RegionSettings) { + fake.getRegionSettingsMutex.Lock() + defer fake.getRegionSettingsMutex.Unlock() + fake.GetRegionSettingsStub = nil + fake.getRegionSettingsReturns = struct { + result1 *livekit.RegionSettings + }{result1} +} + +func (fake *FakeLocalParticipantHelper) GetRegionSettingsReturnsOnCall(i int, result1 *livekit.RegionSettings) { + fake.getRegionSettingsMutex.Lock() + defer fake.getRegionSettingsMutex.Unlock() + fake.GetRegionSettingsStub = nil + if fake.getRegionSettingsReturnsOnCall == nil { + fake.getRegionSettingsReturnsOnCall = make(map[int]struct { + result1 *livekit.RegionSettings + }) + } + fake.getRegionSettingsReturnsOnCall[i] = struct { + result1 *livekit.RegionSettings + }{result1} +} + +func (fake *FakeLocalParticipantHelper) GetSubscriberForwarderState(arg1 types.LocalParticipant) (map[livekit.TrackID]*livekit.RTPForwarderState, error) { + fake.getSubscriberForwarderStateMutex.Lock() + ret, specificReturn := fake.getSubscriberForwarderStateReturnsOnCall[len(fake.getSubscriberForwarderStateArgsForCall)] + fake.getSubscriberForwarderStateArgsForCall = append(fake.getSubscriberForwarderStateArgsForCall, struct { + arg1 types.LocalParticipant + }{arg1}) + stub := fake.GetSubscriberForwarderStateStub + fakeReturns := fake.getSubscriberForwarderStateReturns + fake.recordInvocation("GetSubscriberForwarderState", []interface{}{arg1}) + fake.getSubscriberForwarderStateMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeLocalParticipantHelper) GetSubscriberForwarderStateCallCount() int { + fake.getSubscriberForwarderStateMutex.RLock() + defer fake.getSubscriberForwarderStateMutex.RUnlock() + return len(fake.getSubscriberForwarderStateArgsForCall) +} + +func (fake *FakeLocalParticipantHelper) GetSubscriberForwarderStateCalls(stub func(types.LocalParticipant) (map[livekit.TrackID]*livekit.RTPForwarderState, error)) { + fake.getSubscriberForwarderStateMutex.Lock() + defer fake.getSubscriberForwarderStateMutex.Unlock() + fake.GetSubscriberForwarderStateStub = stub +} + +func (fake *FakeLocalParticipantHelper) GetSubscriberForwarderStateArgsForCall(i int) types.LocalParticipant { + fake.getSubscriberForwarderStateMutex.RLock() + defer fake.getSubscriberForwarderStateMutex.RUnlock() + argsForCall := fake.getSubscriberForwarderStateArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipantHelper) GetSubscriberForwarderStateReturns(result1 map[livekit.TrackID]*livekit.RTPForwarderState, result2 error) { + fake.getSubscriberForwarderStateMutex.Lock() + defer fake.getSubscriberForwarderStateMutex.Unlock() + fake.GetSubscriberForwarderStateStub = nil + fake.getSubscriberForwarderStateReturns = struct { + result1 map[livekit.TrackID]*livekit.RTPForwarderState + result2 error + }{result1, result2} +} + +func (fake *FakeLocalParticipantHelper) GetSubscriberForwarderStateReturnsOnCall(i int, result1 map[livekit.TrackID]*livekit.RTPForwarderState, result2 error) { + fake.getSubscriberForwarderStateMutex.Lock() + defer fake.getSubscriberForwarderStateMutex.Unlock() + fake.GetSubscriberForwarderStateStub = nil + if fake.getSubscriberForwarderStateReturnsOnCall == nil { + fake.getSubscriberForwarderStateReturnsOnCall = make(map[int]struct { + result1 map[livekit.TrackID]*livekit.RTPForwarderState + result2 error + }) + } + fake.getSubscriberForwarderStateReturnsOnCall[i] = struct { + result1 map[livekit.TrackID]*livekit.RTPForwarderState + result2 error + }{result1, result2} +} + +func (fake *FakeLocalParticipantHelper) ResolveMediaTrack(arg1 types.LocalParticipant, arg2 livekit.TrackID) types.MediaResolverResult { + fake.resolveMediaTrackMutex.Lock() + ret, specificReturn := fake.resolveMediaTrackReturnsOnCall[len(fake.resolveMediaTrackArgsForCall)] + fake.resolveMediaTrackArgsForCall = append(fake.resolveMediaTrackArgsForCall, struct { + arg1 types.LocalParticipant + arg2 livekit.TrackID + }{arg1, arg2}) + stub := fake.ResolveMediaTrackStub + fakeReturns := fake.resolveMediaTrackReturns + fake.recordInvocation("ResolveMediaTrack", []interface{}{arg1, arg2}) + fake.resolveMediaTrackMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipantHelper) ResolveMediaTrackCallCount() int { + fake.resolveMediaTrackMutex.RLock() + defer fake.resolveMediaTrackMutex.RUnlock() + return len(fake.resolveMediaTrackArgsForCall) +} + +func (fake *FakeLocalParticipantHelper) ResolveMediaTrackCalls(stub func(types.LocalParticipant, livekit.TrackID) types.MediaResolverResult) { + fake.resolveMediaTrackMutex.Lock() + defer fake.resolveMediaTrackMutex.Unlock() + fake.ResolveMediaTrackStub = stub +} + +func (fake *FakeLocalParticipantHelper) ResolveMediaTrackArgsForCall(i int) (types.LocalParticipant, livekit.TrackID) { + fake.resolveMediaTrackMutex.RLock() + defer fake.resolveMediaTrackMutex.RUnlock() + argsForCall := fake.resolveMediaTrackArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeLocalParticipantHelper) ResolveMediaTrackReturns(result1 types.MediaResolverResult) { + fake.resolveMediaTrackMutex.Lock() + defer fake.resolveMediaTrackMutex.Unlock() + fake.ResolveMediaTrackStub = nil + fake.resolveMediaTrackReturns = struct { + result1 types.MediaResolverResult + }{result1} +} + +func (fake *FakeLocalParticipantHelper) ResolveMediaTrackReturnsOnCall(i int, result1 types.MediaResolverResult) { + fake.resolveMediaTrackMutex.Lock() + defer fake.resolveMediaTrackMutex.Unlock() + fake.ResolveMediaTrackStub = nil + if fake.resolveMediaTrackReturnsOnCall == nil { + fake.resolveMediaTrackReturnsOnCall = make(map[int]struct { + result1 types.MediaResolverResult + }) + } + fake.resolveMediaTrackReturnsOnCall[i] = struct { + result1 types.MediaResolverResult + }{result1} +} + +func (fake *FakeLocalParticipantHelper) ShouldRegressCodec() bool { + fake.shouldRegressCodecMutex.Lock() + ret, specificReturn := fake.shouldRegressCodecReturnsOnCall[len(fake.shouldRegressCodecArgsForCall)] + fake.shouldRegressCodecArgsForCall = append(fake.shouldRegressCodecArgsForCall, struct { + }{}) + stub := fake.ShouldRegressCodecStub + fakeReturns := fake.shouldRegressCodecReturns + fake.recordInvocation("ShouldRegressCodec", []interface{}{}) + fake.shouldRegressCodecMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipantHelper) ShouldRegressCodecCallCount() int { + fake.shouldRegressCodecMutex.RLock() + defer fake.shouldRegressCodecMutex.RUnlock() + return len(fake.shouldRegressCodecArgsForCall) +} + +func (fake *FakeLocalParticipantHelper) ShouldRegressCodecCalls(stub func() bool) { + fake.shouldRegressCodecMutex.Lock() + defer fake.shouldRegressCodecMutex.Unlock() + fake.ShouldRegressCodecStub = stub +} + +func (fake *FakeLocalParticipantHelper) ShouldRegressCodecReturns(result1 bool) { + fake.shouldRegressCodecMutex.Lock() + defer fake.shouldRegressCodecMutex.Unlock() + fake.ShouldRegressCodecStub = nil + fake.shouldRegressCodecReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipantHelper) ShouldRegressCodecReturnsOnCall(i int, result1 bool) { + fake.shouldRegressCodecMutex.Lock() + defer fake.shouldRegressCodecMutex.Unlock() + fake.ShouldRegressCodecStub = nil + if fake.shouldRegressCodecReturnsOnCall == nil { + fake.shouldRegressCodecReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.shouldRegressCodecReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipantHelper) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + fake.getParticipantInfoMutex.RLock() + defer fake.getParticipantInfoMutex.RUnlock() + fake.getRegionSettingsMutex.RLock() + defer fake.getRegionSettingsMutex.RUnlock() + fake.getSubscriberForwarderStateMutex.RLock() + defer fake.getSubscriberForwarderStateMutex.RUnlock() + fake.resolveMediaTrackMutex.RLock() + defer fake.resolveMediaTrackMutex.RUnlock() + fake.shouldRegressCodecMutex.RLock() + defer fake.shouldRegressCodecMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeLocalParticipantHelper) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ types.LocalParticipantHelper = new(FakeLocalParticipantHelper) diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 67304714d..58415b3f2 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -8,6 +8,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/mime" "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" ) type FakeMediaTrack struct { @@ -167,6 +168,16 @@ type FakeMediaTrack struct { kindReturnsOnCall map[int]struct { result1 livekit.TrackType } + LoggerStub func() logger.Logger + loggerMutex sync.RWMutex + loggerArgsForCall []struct { + } + loggerReturns struct { + result1 logger.Logger + } + loggerReturnsOnCall map[int]struct { + result1 logger.Logger + } NameStub func() string nameMutex sync.RWMutex nameArgsForCall []struct { @@ -1118,6 +1129,59 @@ func (fake *FakeMediaTrack) KindReturnsOnCall(i int, result1 livekit.TrackType) }{result1} } +func (fake *FakeMediaTrack) Logger() logger.Logger { + fake.loggerMutex.Lock() + ret, specificReturn := fake.loggerReturnsOnCall[len(fake.loggerArgsForCall)] + fake.loggerArgsForCall = append(fake.loggerArgsForCall, struct { + }{}) + stub := fake.LoggerStub + fakeReturns := fake.loggerReturns + fake.recordInvocation("Logger", []interface{}{}) + fake.loggerMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) LoggerCallCount() int { + fake.loggerMutex.RLock() + defer fake.loggerMutex.RUnlock() + return len(fake.loggerArgsForCall) +} + +func (fake *FakeMediaTrack) LoggerCalls(stub func() logger.Logger) { + fake.loggerMutex.Lock() + defer fake.loggerMutex.Unlock() + fake.LoggerStub = stub +} + +func (fake *FakeMediaTrack) LoggerReturns(result1 logger.Logger) { + fake.loggerMutex.Lock() + defer fake.loggerMutex.Unlock() + fake.LoggerStub = nil + fake.loggerReturns = struct { + result1 logger.Logger + }{result1} +} + +func (fake *FakeMediaTrack) LoggerReturnsOnCall(i int, result1 logger.Logger) { + fake.loggerMutex.Lock() + defer fake.loggerMutex.Unlock() + fake.LoggerStub = nil + if fake.loggerReturnsOnCall == nil { + fake.loggerReturnsOnCall = make(map[int]struct { + result1 logger.Logger + }) + } + fake.loggerReturnsOnCall[i] = struct { + result1 logger.Logger + }{result1} +} + func (fake *FakeMediaTrack) Name() string { fake.nameMutex.Lock() ret, specificReturn := fake.nameReturnsOnCall[len(fake.nameArgsForCall)] @@ -1828,6 +1892,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.isSubscriberMutex.RUnlock() fake.kindMutex.RLock() defer fake.kindMutex.RUnlock() + fake.loggerMutex.RLock() + defer fake.loggerMutex.RUnlock() fake.nameMutex.RLock() defer fake.nameMutex.RUnlock() fake.onTrackSubscribedMutex.RLock() diff --git a/pkg/service/auth.go b/pkg/service/auth.go index b259aa698..9c71ee9bd 100644 --- a/pkg/service/auth.go +++ b/pkg/service/auth.go @@ -219,7 +219,7 @@ func EnsureSIPCallPermission(ctx context.Context) error { return nil } -func EnsureForwardPermission(ctx context.Context, source livekit.RoomName, destination livekit.RoomName) error { +func EnsureDestRoomPermission(ctx context.Context, source livekit.RoomName, destination livekit.RoomName) error { claims := GetGrants(ctx) if claims == nil || claims.Video == nil { return ErrPermissionDenied diff --git a/pkg/service/errors.go b/pkg/service/errors.go index 5c7579abb..3d93eb7ab 100644 --- a/pkg/service/errors.go +++ b/pkg/service/errors.go @@ -30,7 +30,7 @@ var ( ErrAttributeExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "attribute size exceeds limits") ErrRoomNameExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "room name length exceeds limits") ErrParticipantIdentityExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "participant identity length exceeds limits") - ErrForwardToSameRoom = psrpc.NewErrorf(psrpc.InvalidArgument, "cannot forward to the same room") + ErrDestinationSameAsSourceRoom = psrpc.NewErrorf(psrpc.InvalidArgument, "destination room cannot be the same as source room") ErrOperationFailed = psrpc.NewErrorf(psrpc.Internal, "operation cannot be completed") ErrParticipantNotFound = psrpc.NewErrorf(psrpc.NotFound, "participant does not exist") ErrRoomNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested room does not exist") diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 382545ca5..9fc441e08 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -444,17 +444,14 @@ func (r *RoomManager) StartSession( AdaptiveStream: pi.AdaptiveStream, AllowTCPFallback: allowFallback, TURNSEnabled: r.config.IsTURNSEnabled(), - GetParticipantInfo: func(pID livekit.ParticipantID) *livekit.ParticipantInfo { - if p := room.GetParticipantByID(pID); p != nil { - return p.ToProto() - } - return nil + ParticipantHelper: &roomManagerParticipantHelper{ + room: room, + codecRegressionThreshold: r.config.Video.CodecRegressionThreshold, }, ReconnectOnPublicationError: reconnectOnPublicationError, ReconnectOnSubscriptionError: reconnectOnSubscriptionError, ReconnectOnDataChannelError: reconnectOnDataChannelError, VersionGenerator: r.versionGenerator, - TrackResolver: room.ResolveMediaTrackForSubscriber, SubscriberAllowPause: subscriberAllowPause, SubscriptionLimitAudio: r.config.Limit.SubscriptionLimitAudio, SubscriptionLimitVideo: r.config.Limit.SubscriptionLimitVideo, @@ -466,9 +463,6 @@ func (r *RoomManager) StartSession( DataChannelMaxBufferedAmount: r.config.RTC.DataChannelMaxBufferedAmount, DatachannelSlowThreshold: r.config.RTC.DatachannelSlowThreshold, FireOnTrackBySdp: true, - ShouldRegressCodec: func() bool { - return r.config.Video.CodecRegressionThreshold == 0 || room.GetParticipantCount() < r.config.Video.CodecRegressionThreshold - }, }) if err != nil { return err @@ -1019,3 +1013,31 @@ func iceServerForStunServers(servers []string) *livekit.ICEServer { } return iceServer } + +type roomManagerParticipantHelper struct { + room *rtc.Room + codecRegressionThreshold int +} + +func (h *roomManagerParticipantHelper) GetParticipantInfo(pID livekit.ParticipantID) *livekit.ParticipantInfo { + if p := h.room.GetParticipantByID(pID); p != nil { + return p.ToProto() + } + return nil +} + +func (h *roomManagerParticipantHelper) GetRegionSettings(ip string) *livekit.RegionSettings { + return nil +} + +func (h *roomManagerParticipantHelper) GetSubscriberForwarderState(lp types.LocalParticipant) (map[livekit.TrackID]*livekit.RTPForwarderState, error) { + return nil, nil +} + +func (h *roomManagerParticipantHelper) ResolveMediaTrack(lp types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult { + return h.room.ResolveMediaTrackForSubscriber(lp, trackID) +} + +func (h *roomManagerParticipantHelper) ShouldRegressCodec() bool { + return h.codecRegressionThreshold == 0 || h.room.GetParticipantCount() < h.codecRegressionThreshold +} diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index e19e3d035..f43f0e2bd 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -16,7 +16,6 @@ package service import ( "context" - "errors" "fmt" "strconv" @@ -322,12 +321,12 @@ func (s *RoomService) ForwardParticipant(ctx context.Context, req *livekit.Forwa roomName := livekit.RoomName(req.Room) AppendLogFields(ctx, "room", roomName, "participant", req.Identity) - if err := EnsureForwardPermission(ctx, roomName, livekit.RoomName(req.DestinationRoom)); err != nil { + if err := EnsureDestRoomPermission(ctx, roomName, livekit.RoomName(req.DestinationRoom)); err != nil { return nil, twirpAuthError(err) } if req.Room == req.DestinationRoom { - return nil, twirp.InvalidArgumentError(ErrForwardToSameRoom.Error(), "") + return nil, twirp.InvalidArgumentError(ErrDestinationSameAsSourceRoom.Error(), "") } res, err := s.participantClient.ForwardParticipant(ctx, s.topicFormatter.ParticipantTopic(ctx, livekit.RoomName(req.Room), livekit.ParticipantIdentity(req.Identity)), req) @@ -336,7 +335,21 @@ func (s *RoomService) ForwardParticipant(ctx context.Context, req *livekit.Forwa } func (s *RoomService) MoveParticipant(ctx context.Context, req *livekit.MoveParticipantRequest) (*livekit.MoveParticipantResponse, error) { - return nil, errors.New("unimplemented") + RecordRequest(ctx, req) + + roomName := livekit.RoomName(req.Room) + AppendLogFields(ctx, "room", roomName, "participant", req.Identity) + if err := EnsureDestRoomPermission(ctx, roomName, livekit.RoomName(req.DestinationRoom)); err != nil { + return nil, twirpAuthError(err) + } + + if req.Room == req.DestinationRoom { + return nil, twirp.InvalidArgumentError(ErrDestinationSameAsSourceRoom.Error(), "") + } + + res, err := s.participantClient.MoveParticipant(ctx, s.topicFormatter.ParticipantTopic(ctx, livekit.RoomName(req.Room), livekit.ParticipantIdentity(req.Identity)), req) + RecordResponse(ctx, res) + return res, err } func redactCreateRoomRequest(req *livekit.CreateRoomRequest) *livekit.CreateRoomRequest { diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index b8c15dedf..b27d5d203 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -227,13 +227,13 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { loggerFields := []any{ "participant", pi.Identity, - "room", roomName, "remote", false, } if pi.ID != "" { loggerFields = append(loggerFields, "pID", pi.ID) } - pLogger := utils.GetLogger(r.Context()).WithValues(loggerFields...) + pLogger, loggerResolver := utils.GetLogger(r.Context()).WithValues(loggerFields...).WithDeferredValues() + loggerResolver("room", roomName) // give it a few attempts to start session var cr connectionResult @@ -263,7 +263,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { pLogger = pLogger.WithValues("connID", cr.ConnectionID) if !pi.Reconnect && initialResponse.GetJoin() != nil { pi.ID = livekit.ParticipantID(initialResponse.GetJoin().GetParticipant().GetSid()) - pLogger = pLogger.WithValues("pID", pi.ID) + loggerResolver("pID", pi.ID) } signalStats := telemetry.NewBytesSignalStats(r.Context(), s.telemetry) @@ -372,6 +372,9 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { signalStats.ResolveRoom(m.RoomUpdate.GetRoom()) case *livekit.SignalResponse_Update: pLogger.Debugw("sending participant update", "participantUpdate", m) + case *livekit.SignalResponse_RoomMoved: + loggerResolver("room", m.RoomMoved.GetRoom(), "pID", m.RoomMoved.GetParticipant().GetSid()) + pLogger.Debugw("sending room moved", "roomMoved", m) } if count, err := sigConn.WriteResponse(res); err != nil {