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

@@ -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

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
 		}
 

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
 	}

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) {

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)

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()