From 0cd9c87dc9baa66702276380bc2ee6b8406518b5 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 19 Aug 2022 11:49:12 +0530 Subject: [PATCH] Misc clean up (#931) * Start RTCP workers after peer connection connects * Move more things into transport module * Start RTCP workers only on connected * Test needs PeerConnection() method * adjust comment --- pkg/rtc/mediatracksubscriptions.go | 74 +---- pkg/rtc/participant.go | 21 +- pkg/rtc/transport.go | 47 ++- pkg/rtc/transportmanager.go | 28 +- pkg/rtc/types/interfaces.go | 4 +- .../typesfakes/fake_local_participant.go | 307 ++++++++++++++---- 6 files changed, 328 insertions(+), 153 deletions(-) diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 1e18b7fe4..2680d5fd9 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -4,9 +4,7 @@ import ( "context" "errors" "sync" - "time" - "github.com/pion/rtcp" "github.com/pion/webrtc/v3" "github.com/pion/webrtc/v3/pkg/rtcerr" "go.uber.org/atomic" @@ -22,8 +20,6 @@ import ( var ( errAlreadySubscribed = errors.New("already subscribed") - errNoTransceiver = errors.New("cannot subscribe without a transceiver in place") - errNoSender = errors.New("cannot subscribe without a sender in place") errNotFound = errors.New("not found") ) @@ -154,9 +150,6 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * go subTrack.Bound() - // when down track is bound, start loop to send reports - go t.sendDownTrackBindingReports(sub) - subTrack.SetPublisherMuted(t.params.MediaTrack.IsMuted()) }) @@ -215,35 +208,17 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * // if the attributes match. This prevents SDP from bloating // because of dormant transceivers building up. // - sender, err = sub.SubscriberPC().AddTrack(downTrack) + sender, transceiver, err = sub.AddTrackToSubscriber(downTrack) if err != nil { return 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 - } - } } else { - transceiver, err = sub.SubscriberPC().AddTransceiverFromTrack(downTrack) + sender, transceiver, err = sub.AddTransceiverFromTrackToSubscriber(downTrack) if err != nil { return err } - - sender = transceiver.Sender() } } - if transceiver == nil { - // cannot add, no transceiver - return errNoTransceiver - } - if sender == nil { - // cannot add, no sender - return errNoSender - } // wthether re-using or stopping remove transceiver from cache // NOTE: safety net, if somehow a cached transceiver is re-used by a different track @@ -385,44 +360,6 @@ func (t *MediaTrackSubscriptions) getAllSubscribedTracksLocked() []types.Subscri return subTracks } -// TODO: send for all down tracks from the source participant -// https://tools.ietf.org/html/rfc7941 -func (t *MediaTrackSubscriptions) sendDownTrackBindingReports(sub types.LocalParticipant) { - 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 { - sub.GetLogger().Errorw("could not write RTCP", err) - return - } - if i > 5 { - return - } - i++ - time.Sleep(20 * time.Millisecond) - } - }() -} - func (t *MediaTrackSubscriptions) DebugInfo() []map[string]interface{} { subscribedTrackInfo := make([]map[string]interface{}, 0) for _, val := range t.getAllSubscribedTracks() { @@ -457,11 +394,6 @@ func (t *MediaTrackSubscriptions) downTrackClosed( if !willBeResumed { 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 the subscribed tracks // however, if the dest sub has disconnected, then we can skip if sender == nil { @@ -472,7 +404,7 @@ func (t *MediaTrackSubscriptions) downTrackClosed( "publisherID", subTrack.PublisherID(), "kind", t.params.MediaTrack.Kind(), ) - if err := sub.SubscriberPC().RemoveTrack(sender); err != nil { + if err := sub.RemoveTrackFromSubscriber(sender); err != nil { if err == webrtc.ErrConnectionClosed { // sub closing, can skip removing subscribedtracks return diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index fc1c53d81..46f06b263 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -570,8 +570,6 @@ func (p *ParticipantImpl) SetMigrateInfo(previousAnswer *webrtc.SessionDescripti func (p *ParticipantImpl) Start() { p.once.Do(func() { p.UpTrackManager.Start() - go p.rtcpSendWorker() - go p.downTracksRTCPWorker() }) } @@ -1002,6 +1000,7 @@ func (p *ParticipantImpl) setupTransportManager() error { tm.OnPublisherGetDTX(p.onPublisherGetDTX) tm.OnPublisherAnswer(p.onPublisherAnswer) tm.OnPublisherTrack(p.onMediaTrack) + tm.OnPublisherInitialConnected(p.onPublisherInitialConnected) tm.OnSubscriberOffer(p.onSubscriberOffer) tm.OnSubscriberICECandidate(func(c *webrtc.ICECandidate) { @@ -1159,7 +1158,13 @@ func (p *ParticipantImpl) onICECandidate(c *webrtc.ICECandidate, target livekit. p.sendICECandidate(c, target) } +func (p *ParticipantImpl) onPublisherInitialConnected() { + go p.publisherRTCPWorker() +} + func (p *ParticipantImpl) onSubscriberInitialConnected() { + go p.subscriberRTCPWorker() + p.setDowntracksConnected() } @@ -1198,13 +1203,11 @@ func (p *ParticipantImpl) onAnyTransportFailed() { p.lock.Unlock() } -// downTracksRTCPWorker sends SenderReports periodically when the participant is subscribed to +// subscriberRTCPWorker sends SenderReports periodically when the participant is subscribed to // other publishedTracks in the room. -func (p *ParticipantImpl) downTracksRTCPWorker() { +func (p *ParticipantImpl) subscriberRTCPWorker() { defer Recover() for { - time.Sleep(5 * time.Second) - if p.State() == livekit.ParticipantInfo_DISCONNECTED { return } @@ -1257,6 +1260,8 @@ func (p *ParticipantImpl) downTracksRTCPWorker() { pkts = pkts[:0] batchSize = 0 } + + time.Sleep(5 * time.Second) } } @@ -1752,13 +1757,13 @@ func (p *ParticipantImpl) getPublishedTrackBySdpCid(clientId string) types.Media return nil } -func (p *ParticipantImpl) rtcpSendWorker() { +func (p *ParticipantImpl) publisherRTCPWorker() { defer Recover() // read from rtcpChan for pkts := range p.rtcpCh { if pkts == nil { - p.params.Logger.Infow("exiting RTCP send worker") + p.params.Logger.Infow("exiting publisher RTCP worker") return } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 8e9fbf28e..12d6e53d6 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -48,6 +48,8 @@ const ( var ( ErrIceRestartWithoutLocalSDP = errors.New("ICE restart without local SDP settled") + ErrNoTransceiver = errors.New("no transceiver") + ErrNoSender = errors.New("no sender") ) const ( @@ -550,6 +552,47 @@ func (t *PCTransport) PeerConnection() *webrtc.PeerConnection { return t.pc } +func (t *PCTransport) AddTrack(trackLocal webrtc.TrackLocal) (sender *webrtc.RTPSender, transceiver *webrtc.RTPTransceiver, err error) { + sender, err = t.pc.AddTrack(trackLocal) + if err != nil { + return + } + + // as there is no way to get transceiver from sender, search + for _, tr := range t.pc.GetTransceivers() { + if tr.Sender() == sender { + transceiver = tr + break + } + } + + if transceiver == nil { + err = ErrNoTransceiver + return + } + + return +} + +func (t *PCTransport) AddTransceiverFromTrack(trackLocal webrtc.TrackLocal) (sender *webrtc.RTPSender, transceiver *webrtc.RTPTransceiver, err error) { + transceiver, err = t.pc.AddTransceiverFromTrack(trackLocal) + if err != nil { + return + } + + sender = transceiver.Sender() + if sender == nil { + err = ErrNoSender + return + } + + return +} + +func (t *PCTransport) RemoveTrack(sender *webrtc.RTPSender) error { + return t.pc.RemoveTrack(sender) +} + func (t *PCTransport) GetMid(rtpReceiver *webrtc.RTPReceiver) string { for _, tr := range t.pc.GetTransceivers() { if tr.Receiver() == rtpReceiver { @@ -1232,7 +1275,7 @@ func (t *PCTransport) OnStreamStateChange(f func(update *sfu.StreamStateUpdate) t.streamAllocator.OnStreamStateChange(f) } -func (t *PCTransport) AddTrack(subTrack types.SubscribedTrack) { +func (t *PCTransport) AddTrackToStreamAllocator(subTrack types.SubscribedTrack) { if t.streamAllocator == nil { return } @@ -1244,7 +1287,7 @@ func (t *PCTransport) AddTrack(subTrack types.SubscribedTrack) { }) } -func (t *PCTransport) RemoveTrack(subTrack types.SubscribedTrack) { +func (t *PCTransport) RemoveTrackFromStreamAllocator(subTrack types.SubscribedTrack) { if t.streamAllocator == nil { return } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 734cdab8a..b733f9926 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -49,6 +49,7 @@ type TransportManager struct { onPublisherGetDTX func() bool + onPublisherInitialConnected func() onSubscriberInitialConnected func() onPrimaryTransportInitialConnected func() onAnyTransportFailed func() @@ -94,6 +95,9 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro t.publisher = publisher t.publisher.OnRemoteDescriptionSettled(t.createPublisherAnswerAndSend) t.publisher.OnInitialConnected(func() { + if t.onPublisherInitialConnected != nil { + t.onPublisherInitialConnected() + } if !t.params.SubscriberAsPrimary && t.onPrimaryTransportInitialConnected != nil { t.onPrimaryTransportInitialConnected() } @@ -163,6 +167,10 @@ func (t *TransportManager) OnPublisherAnswer(f func(answer webrtc.SessionDescrip }) } +func (t *TransportManager) OnPublisherInitialConnected(f func()) { + t.onPublisherInitialConnected = f +} + func (t *TransportManager) OnPublisherTrack(f func(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver)) { t.publisher.OnTrack(f) } @@ -215,6 +223,18 @@ func (t *TransportManager) HasSubscriberEverConnected() bool { return t.subscriber.HasEverConnected() } +func (t *TransportManager) AddTrackToSubscriber(trackLocal webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) { + return t.subscriber.AddTrack(trackLocal) +} + +func (t *TransportManager) AddTransceiverFromTrackToSubscriber(trackLocal webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) { + return t.subscriber.AddTransceiverFromTrack(trackLocal) +} + +func (t *TransportManager) RemoveTrackFromSubscriber(sender *webrtc.RTPSender) error { + return t.subscriber.RemoveTrack(sender) +} + func (t *TransportManager) WriteSubscriberRTCP(pkts []rtcp.Packet) error { return t.subscriber.WriteRTCP(pkts) } @@ -232,11 +252,11 @@ func (t *TransportManager) OnAnyTransportFailed(f func()) { } func (t *TransportManager) AddSubscribedTrack(subTrack types.SubscribedTrack) { - t.subscriber.AddTrack(subTrack) + t.subscriber.AddTrackToStreamAllocator(subTrack) } func (t *TransportManager) RemoveSubscribedTrack(subTrack types.SubscribedTrack) { - t.subscriber.RemoveTrack(subTrack) + t.subscriber.RemoveTrackFromStreamAllocator(subTrack) } func (t *TransportManager) OnDataMessage(f func(kind livekit.DataPacket_Kind, data []byte)) { @@ -458,10 +478,6 @@ func (t *TransportManager) SubscriberAsPrimary() bool { return t.params.SubscriberAsPrimary } -func (t *TransportManager) SubscriberPC() *webrtc.PeerConnection { - return t.subscriber.PeerConnection() -} - func (t *TransportManager) getTransport(isPrimary bool) *PCTransport { pcTransport := t.publisher if (isPrimary && t.params.SubscriberAsPrimary) || (!isPrimary && !t.params.SubscriberAsPrimary) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 28fb88c68..96083ba0e 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -230,12 +230,14 @@ type LocalParticipant interface { AddTrack(req *livekit.AddTrackRequest) SetTrackMuted(trackID livekit.TrackID, muted bool, fromAdmin bool) - SubscriberPC() *webrtc.PeerConnection HandleAnswer(sdp webrtc.SessionDescription) error Negotiate(force bool) AddNegotiationPending(publisherID livekit.ParticipantID) IsNegotiationPending(publisherID livekit.ParticipantID) bool ICERestart(iceConfig *IceConfig) error + AddTrackToSubscriber(trackLocal webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) + AddTransceiverFromTrackToSubscriber(trackLocal webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) + RemoveTrackFromSubscriber(sender *webrtc.RTPSender) error // subscriptions AddSubscribedTrack(st SubscribedTrack) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 7c17e2251..83ede51d7 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -56,6 +56,36 @@ type FakeLocalParticipant struct { addTrackArgsForCall []struct { arg1 *livekit.AddTrackRequest } + AddTrackToSubscriberStub func(webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) + addTrackToSubscriberMutex sync.RWMutex + addTrackToSubscriberArgsForCall []struct { + arg1 webrtc.TrackLocal + } + addTrackToSubscriberReturns struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + } + addTrackToSubscriberReturnsOnCall map[int]struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + } + AddTransceiverFromTrackToSubscriberStub func(webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) + addTransceiverFromTrackToSubscriberMutex sync.RWMutex + addTransceiverFromTrackToSubscriberArgsForCall []struct { + arg1 webrtc.TrackLocal + } + addTransceiverFromTrackToSubscriberReturns struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + } + addTransceiverFromTrackToSubscriberReturnsOnCall map[int]struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + } CacheDownTrackStub func(livekit.TrackID, *webrtc.RTPTransceiver, sfu.ForwarderState) cacheDownTrackMutex sync.RWMutex cacheDownTrackArgsForCall []struct { @@ -471,6 +501,17 @@ type FakeLocalParticipant struct { arg2 livekit.TrackID arg3 bool } + RemoveTrackFromSubscriberStub func(*webrtc.RTPSender) error + removeTrackFromSubscriberMutex sync.RWMutex + removeTrackFromSubscriberArgsForCall []struct { + arg1 *webrtc.RTPSender + } + removeTrackFromSubscriberReturns struct { + result1 error + } + removeTrackFromSubscriberReturnsOnCall map[int]struct { + result1 error + } SendConnectionQualityUpdateStub func(*livekit.ConnectionQualityUpdate) error sendConnectionQualityUpdateMutex sync.RWMutex sendConnectionQualityUpdateArgsForCall []struct { @@ -617,16 +658,6 @@ type FakeLocalParticipant struct { subscriberAsPrimaryReturnsOnCall map[int]struct { result1 bool } - SubscriberPCStub func() *webrtc.PeerConnection - subscriberPCMutex sync.RWMutex - subscriberPCArgsForCall []struct { - } - subscriberPCReturns struct { - result1 *webrtc.PeerConnection - } - subscriberPCReturnsOnCall map[int]struct { - result1 *webrtc.PeerConnection - } SubscriptionPermissionStub func() (*livekit.SubscriptionPermission, *livekit.TimedVersion) subscriptionPermissionMutex sync.RWMutex subscriptionPermissionArgsForCall []struct { @@ -956,6 +987,140 @@ func (fake *FakeLocalParticipant) AddTrackArgsForCall(i int) *livekit.AddTrackRe return argsForCall.arg1 } +func (fake *FakeLocalParticipant) AddTrackToSubscriber(arg1 webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) { + fake.addTrackToSubscriberMutex.Lock() + ret, specificReturn := fake.addTrackToSubscriberReturnsOnCall[len(fake.addTrackToSubscriberArgsForCall)] + fake.addTrackToSubscriberArgsForCall = append(fake.addTrackToSubscriberArgsForCall, struct { + arg1 webrtc.TrackLocal + }{arg1}) + stub := fake.AddTrackToSubscriberStub + fakeReturns := fake.addTrackToSubscriberReturns + fake.recordInvocation("AddTrackToSubscriber", []interface{}{arg1}) + fake.addTrackToSubscriberMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1, ret.result2, ret.result3 + } + return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3 +} + +func (fake *FakeLocalParticipant) AddTrackToSubscriberCallCount() int { + fake.addTrackToSubscriberMutex.RLock() + defer fake.addTrackToSubscriberMutex.RUnlock() + return len(fake.addTrackToSubscriberArgsForCall) +} + +func (fake *FakeLocalParticipant) AddTrackToSubscriberCalls(stub func(webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error)) { + fake.addTrackToSubscriberMutex.Lock() + defer fake.addTrackToSubscriberMutex.Unlock() + fake.AddTrackToSubscriberStub = stub +} + +func (fake *FakeLocalParticipant) AddTrackToSubscriberArgsForCall(i int) webrtc.TrackLocal { + fake.addTrackToSubscriberMutex.RLock() + defer fake.addTrackToSubscriberMutex.RUnlock() + argsForCall := fake.addTrackToSubscriberArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) AddTrackToSubscriberReturns(result1 *webrtc.RTPSender, result2 *webrtc.RTPTransceiver, result3 error) { + fake.addTrackToSubscriberMutex.Lock() + defer fake.addTrackToSubscriberMutex.Unlock() + fake.AddTrackToSubscriberStub = nil + fake.addTrackToSubscriberReturns = struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + }{result1, result2, result3} +} + +func (fake *FakeLocalParticipant) AddTrackToSubscriberReturnsOnCall(i int, result1 *webrtc.RTPSender, result2 *webrtc.RTPTransceiver, result3 error) { + fake.addTrackToSubscriberMutex.Lock() + defer fake.addTrackToSubscriberMutex.Unlock() + fake.AddTrackToSubscriberStub = nil + if fake.addTrackToSubscriberReturnsOnCall == nil { + fake.addTrackToSubscriberReturnsOnCall = make(map[int]struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + }) + } + fake.addTrackToSubscriberReturnsOnCall[i] = struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + }{result1, result2, result3} +} + +func (fake *FakeLocalParticipant) AddTransceiverFromTrackToSubscriber(arg1 webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) { + fake.addTransceiverFromTrackToSubscriberMutex.Lock() + ret, specificReturn := fake.addTransceiverFromTrackToSubscriberReturnsOnCall[len(fake.addTransceiverFromTrackToSubscriberArgsForCall)] + fake.addTransceiverFromTrackToSubscriberArgsForCall = append(fake.addTransceiverFromTrackToSubscriberArgsForCall, struct { + arg1 webrtc.TrackLocal + }{arg1}) + stub := fake.AddTransceiverFromTrackToSubscriberStub + fakeReturns := fake.addTransceiverFromTrackToSubscriberReturns + fake.recordInvocation("AddTransceiverFromTrackToSubscriber", []interface{}{arg1}) + fake.addTransceiverFromTrackToSubscriberMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1, ret.result2, ret.result3 + } + return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3 +} + +func (fake *FakeLocalParticipant) AddTransceiverFromTrackToSubscriberCallCount() int { + fake.addTransceiverFromTrackToSubscriberMutex.RLock() + defer fake.addTransceiverFromTrackToSubscriberMutex.RUnlock() + return len(fake.addTransceiverFromTrackToSubscriberArgsForCall) +} + +func (fake *FakeLocalParticipant) AddTransceiverFromTrackToSubscriberCalls(stub func(webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error)) { + fake.addTransceiverFromTrackToSubscriberMutex.Lock() + defer fake.addTransceiverFromTrackToSubscriberMutex.Unlock() + fake.AddTransceiverFromTrackToSubscriberStub = stub +} + +func (fake *FakeLocalParticipant) AddTransceiverFromTrackToSubscriberArgsForCall(i int) webrtc.TrackLocal { + fake.addTransceiverFromTrackToSubscriberMutex.RLock() + defer fake.addTransceiverFromTrackToSubscriberMutex.RUnlock() + argsForCall := fake.addTransceiverFromTrackToSubscriberArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) AddTransceiverFromTrackToSubscriberReturns(result1 *webrtc.RTPSender, result2 *webrtc.RTPTransceiver, result3 error) { + fake.addTransceiverFromTrackToSubscriberMutex.Lock() + defer fake.addTransceiverFromTrackToSubscriberMutex.Unlock() + fake.AddTransceiverFromTrackToSubscriberStub = nil + fake.addTransceiverFromTrackToSubscriberReturns = struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + }{result1, result2, result3} +} + +func (fake *FakeLocalParticipant) AddTransceiverFromTrackToSubscriberReturnsOnCall(i int, result1 *webrtc.RTPSender, result2 *webrtc.RTPTransceiver, result3 error) { + fake.addTransceiverFromTrackToSubscriberMutex.Lock() + defer fake.addTransceiverFromTrackToSubscriberMutex.Unlock() + fake.AddTransceiverFromTrackToSubscriberStub = nil + if fake.addTransceiverFromTrackToSubscriberReturnsOnCall == nil { + fake.addTransceiverFromTrackToSubscriberReturnsOnCall = make(map[int]struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + }) + } + fake.addTransceiverFromTrackToSubscriberReturnsOnCall[i] = struct { + result1 *webrtc.RTPSender + result2 *webrtc.RTPTransceiver + result3 error + }{result1, result2, result3} +} + func (fake *FakeLocalParticipant) CacheDownTrack(arg1 livekit.TrackID, arg2 *webrtc.RTPTransceiver, arg3 sfu.ForwarderState) { fake.cacheDownTrackMutex.Lock() fake.cacheDownTrackArgsForCall = append(fake.cacheDownTrackArgsForCall, struct { @@ -3221,6 +3386,67 @@ func (fake *FakeLocalParticipant) RemoveSubscriberArgsForCall(i int) (types.Loca return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } +func (fake *FakeLocalParticipant) RemoveTrackFromSubscriber(arg1 *webrtc.RTPSender) error { + fake.removeTrackFromSubscriberMutex.Lock() + ret, specificReturn := fake.removeTrackFromSubscriberReturnsOnCall[len(fake.removeTrackFromSubscriberArgsForCall)] + fake.removeTrackFromSubscriberArgsForCall = append(fake.removeTrackFromSubscriberArgsForCall, struct { + arg1 *webrtc.RTPSender + }{arg1}) + stub := fake.RemoveTrackFromSubscriberStub + fakeReturns := fake.removeTrackFromSubscriberReturns + fake.recordInvocation("RemoveTrackFromSubscriber", []interface{}{arg1}) + fake.removeTrackFromSubscriberMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) RemoveTrackFromSubscriberCallCount() int { + fake.removeTrackFromSubscriberMutex.RLock() + defer fake.removeTrackFromSubscriberMutex.RUnlock() + return len(fake.removeTrackFromSubscriberArgsForCall) +} + +func (fake *FakeLocalParticipant) RemoveTrackFromSubscriberCalls(stub func(*webrtc.RTPSender) error) { + fake.removeTrackFromSubscriberMutex.Lock() + defer fake.removeTrackFromSubscriberMutex.Unlock() + fake.RemoveTrackFromSubscriberStub = stub +} + +func (fake *FakeLocalParticipant) RemoveTrackFromSubscriberArgsForCall(i int) *webrtc.RTPSender { + fake.removeTrackFromSubscriberMutex.RLock() + defer fake.removeTrackFromSubscriberMutex.RUnlock() + argsForCall := fake.removeTrackFromSubscriberArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) RemoveTrackFromSubscriberReturns(result1 error) { + fake.removeTrackFromSubscriberMutex.Lock() + defer fake.removeTrackFromSubscriberMutex.Unlock() + fake.RemoveTrackFromSubscriberStub = nil + fake.removeTrackFromSubscriberReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeLocalParticipant) RemoveTrackFromSubscriberReturnsOnCall(i int, result1 error) { + fake.removeTrackFromSubscriberMutex.Lock() + defer fake.removeTrackFromSubscriberMutex.Unlock() + fake.RemoveTrackFromSubscriberStub = nil + if fake.removeTrackFromSubscriberReturnsOnCall == nil { + fake.removeTrackFromSubscriberReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.removeTrackFromSubscriberReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeLocalParticipant) SendConnectionQualityUpdate(arg1 *livekit.ConnectionQualityUpdate) error { fake.sendConnectionQualityUpdateMutex.Lock() ret, specificReturn := fake.sendConnectionQualityUpdateReturnsOnCall[len(fake.sendConnectionQualityUpdateArgsForCall)] @@ -4055,59 +4281,6 @@ func (fake *FakeLocalParticipant) SubscriberAsPrimaryReturnsOnCall(i int, result }{result1} } -func (fake *FakeLocalParticipant) SubscriberPC() *webrtc.PeerConnection { - fake.subscriberPCMutex.Lock() - ret, specificReturn := fake.subscriberPCReturnsOnCall[len(fake.subscriberPCArgsForCall)] - fake.subscriberPCArgsForCall = append(fake.subscriberPCArgsForCall, struct { - }{}) - stub := fake.SubscriberPCStub - fakeReturns := fake.subscriberPCReturns - fake.recordInvocation("SubscriberPC", []interface{}{}) - fake.subscriberPCMutex.Unlock() - if stub != nil { - return stub() - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeLocalParticipant) SubscriberPCCallCount() int { - fake.subscriberPCMutex.RLock() - defer fake.subscriberPCMutex.RUnlock() - return len(fake.subscriberPCArgsForCall) -} - -func (fake *FakeLocalParticipant) SubscriberPCCalls(stub func() *webrtc.PeerConnection) { - fake.subscriberPCMutex.Lock() - defer fake.subscriberPCMutex.Unlock() - fake.SubscriberPCStub = stub -} - -func (fake *FakeLocalParticipant) SubscriberPCReturns(result1 *webrtc.PeerConnection) { - fake.subscriberPCMutex.Lock() - defer fake.subscriberPCMutex.Unlock() - fake.SubscriberPCStub = nil - fake.subscriberPCReturns = struct { - result1 *webrtc.PeerConnection - }{result1} -} - -func (fake *FakeLocalParticipant) SubscriberPCReturnsOnCall(i int, result1 *webrtc.PeerConnection) { - fake.subscriberPCMutex.Lock() - defer fake.subscriberPCMutex.Unlock() - fake.SubscriberPCStub = nil - if fake.subscriberPCReturnsOnCall == nil { - fake.subscriberPCReturnsOnCall = make(map[int]struct { - result1 *webrtc.PeerConnection - }) - } - fake.subscriberPCReturnsOnCall[i] = struct { - result1 *webrtc.PeerConnection - }{result1} -} - func (fake *FakeLocalParticipant) SubscriptionPermission() (*livekit.SubscriptionPermission, *livekit.TimedVersion) { fake.subscriptionPermissionMutex.Lock() ret, specificReturn := fake.subscriptionPermissionReturnsOnCall[len(fake.subscriptionPermissionArgsForCall)] @@ -4646,6 +4819,10 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.addSubscriberMutex.RUnlock() fake.addTrackMutex.RLock() defer fake.addTrackMutex.RUnlock() + fake.addTrackToSubscriberMutex.RLock() + defer fake.addTrackToSubscriberMutex.RUnlock() + fake.addTransceiverFromTrackToSubscriberMutex.RLock() + defer fake.addTransceiverFromTrackToSubscriberMutex.RUnlock() fake.cacheDownTrackMutex.RLock() defer fake.cacheDownTrackMutex.RUnlock() fake.canPublishMutex.RLock() @@ -4742,6 +4919,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.removeSubscribedTrackMutex.RUnlock() fake.removeSubscriberMutex.RLock() defer fake.removeSubscriberMutex.RUnlock() + fake.removeTrackFromSubscriberMutex.RLock() + defer fake.removeTrackFromSubscriberMutex.RUnlock() fake.sendConnectionQualityUpdateMutex.RLock() defer fake.sendConnectionQualityUpdateMutex.RUnlock() fake.sendDataPacketMutex.RLock() @@ -4776,8 +4955,6 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.stateMutex.RUnlock() fake.subscriberAsPrimaryMutex.RLock() defer fake.subscriberAsPrimaryMutex.RUnlock() - fake.subscriberPCMutex.RLock() - defer fake.subscriberPCMutex.RUnlock() fake.subscriptionPermissionMutex.RLock() defer fake.subscriptionPermissionMutex.RUnlock() fake.subscriptionPermissionUpdateMutex.RLock()