Fix rtcp lost for downtrack used incorrect buffer factory (#1195)

* Fix rtcp lost for downtrack used incorrect buffer factory

In buffer factory change(#1173), every pariticipant has its own
buffer factory, can't use publisher's bufferfactory to create
DownTrack

* clean code

pkg/rtc/mediatrack.go

@@ -68,7 +68,6 @@ func NewMediaTrack(params MediaTrackParams) *MediaTrack {
 		ParticipantID:       params.ParticipantID,
 		ParticipantIdentity: params.ParticipantIdentity,
 		ParticipantVersion:  params.ParticipantVersion,
-		BufferFactory:       params.BufferFactory,
 		ReceiverConfig:      params.ReceiverConfig,
 		SubscriberConfig:    params.SubscriberConfig,
 		Telemetry:           params.Telemetry,

pkg/rtc/mediatrackreceiver.go

@@ -69,7 +69,6 @@ type MediaTrackReceiverParams struct {
 	ParticipantID       livekit.ParticipantID
 	ParticipantIdentity livekit.ParticipantIdentity
 	ParticipantVersion  uint32
-	BufferFactory       *buffer.Factory
 	ReceiverConfig      ReceiverConfig
 	SubscriberConfig    DirectionConfig
 	Telemetry           telemetry.TelemetryService
@@ -110,7 +109,6 @@ func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver
 	t.MediaTrackSubscriptions = NewMediaTrackSubscriptions(MediaTrackSubscriptionsParams{
 		MediaTrack:       params.MediaTrack,
 		IsRelayed:        params.IsRelayed,
-		BufferFactory:    params.BufferFactory,
 		ReceiverConfig:   params.ReceiverConfig,
 		SubscriberConfig: params.SubscriberConfig,
 		Telemetry:        params.Telemetry,

pkg/rtc/mediatracksubscriptions.go

@@ -15,7 +15,6 @@ import (
 
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu"
-	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 )
 
@@ -40,7 +39,6 @@ type MediaTrackSubscriptionsParams struct {
 	MediaTrack types.MediaTrack
 	IsRelayed  bool
 
-	BufferFactory    *buffer.Factory
 	ReceiverConfig   ReceiverConfig
 	SubscriberConfig DirectionConfig
 
@@ -110,7 +108,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *
 	downTrack, err := sfu.NewDownTrack(
 		codecs,
 		wr,
-		t.params.BufferFactory,
+		sub.GetBufferFactory(),
 		subscriberID,
 		t.params.ReceiverConfig.PacketBufferSize,
 		LoggerWithTrack(sub.GetLogger(), trackID, t.params.IsRelayed),

pkg/rtc/participant.go

@@ -21,6 +21,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/rtc/supervisor"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu"
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/mediatransportutil/pkg/twcc"
@@ -250,6 +251,10 @@ func (p *ParticipantImpl) GetICEConnectionType() types.ICEConnectionType {
 	return p.TransportManager.GetICEConnectionType()
 }
 
+func (p *ParticipantImpl) GetBufferFactory() *buffer.Factory {
+	return p.params.Config.BufferFactory
+}
+
 // SetMetadata attaches metadata to the participant
 func (p *ParticipantImpl) SetMetadata(metadata string) {
 	p.lock.Lock()

pkg/rtc/types/interfaces.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/sfu"
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 )
 
 //go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
@@ -244,6 +245,7 @@ type LocalParticipant interface {
 	SubscriberAsPrimary() bool
 	GetClientConfiguration() *livekit.ClientConfiguration
 	GetICEConnectionType() ICEConnectionType
+	GetBufferFactory() *buffer.Factory
 
 	SetResponseSink(sink routing.MessageSink)
 	CloseSignalConnection()

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -8,6 +8,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu"
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/protocol/auth"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
@@ -215,6 +216,16 @@ type FakeLocalParticipant struct {
 		result1 float64
 		result2 bool
 	}
+	GetBufferFactoryStub        func() *buffer.Factory
+	getBufferFactoryMutex       sync.RWMutex
+	getBufferFactoryArgsForCall []struct {
+	}
+	getBufferFactoryReturns struct {
+		result1 *buffer.Factory
+	}
+	getBufferFactoryReturnsOnCall map[int]struct {
+		result1 *buffer.Factory
+	}
 	GetCachedDownTrackStub        func(livekit.TrackID) (*webrtc.RTPTransceiver, sfu.DownTrackState)
 	getCachedDownTrackMutex       sync.RWMutex
 	getCachedDownTrackArgsForCall []struct {
@@ -1781,6 +1792,59 @@ func (fake *FakeLocalParticipant) GetAudioLevelReturnsOnCall(i int, result1 floa
 	}{result1, result2}
 }
 
+func (fake *FakeLocalParticipant) GetBufferFactory() *buffer.Factory {
+	fake.getBufferFactoryMutex.Lock()
+	ret, specificReturn := fake.getBufferFactoryReturnsOnCall[len(fake.getBufferFactoryArgsForCall)]
+	fake.getBufferFactoryArgsForCall = append(fake.getBufferFactoryArgsForCall, struct {
+	}{})
+	stub := fake.GetBufferFactoryStub
+	fakeReturns := fake.getBufferFactoryReturns
+	fake.recordInvocation("GetBufferFactory", []interface{}{})
+	fake.getBufferFactoryMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeLocalParticipant) GetBufferFactoryCallCount() int {
+	fake.getBufferFactoryMutex.RLock()
+	defer fake.getBufferFactoryMutex.RUnlock()
+	return len(fake.getBufferFactoryArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) GetBufferFactoryCalls(stub func() *buffer.Factory) {
+	fake.getBufferFactoryMutex.Lock()
+	defer fake.getBufferFactoryMutex.Unlock()
+	fake.GetBufferFactoryStub = stub
+}
+
+func (fake *FakeLocalParticipant) GetBufferFactoryReturns(result1 *buffer.Factory) {
+	fake.getBufferFactoryMutex.Lock()
+	defer fake.getBufferFactoryMutex.Unlock()
+	fake.GetBufferFactoryStub = nil
+	fake.getBufferFactoryReturns = struct {
+		result1 *buffer.Factory
+	}{result1}
+}
+
+func (fake *FakeLocalParticipant) GetBufferFactoryReturnsOnCall(i int, result1 *buffer.Factory) {
+	fake.getBufferFactoryMutex.Lock()
+	defer fake.getBufferFactoryMutex.Unlock()
+	fake.GetBufferFactoryStub = nil
+	if fake.getBufferFactoryReturnsOnCall == nil {
+		fake.getBufferFactoryReturnsOnCall = make(map[int]struct {
+			result1 *buffer.Factory
+		})
+	}
+	fake.getBufferFactoryReturnsOnCall[i] = struct {
+		result1 *buffer.Factory
+	}{result1}
+}
+
 func (fake *FakeLocalParticipant) GetCachedDownTrack(arg1 livekit.TrackID) (*webrtc.RTPTransceiver, sfu.DownTrackState) {
 	fake.getCachedDownTrackMutex.Lock()
 	ret, specificReturn := fake.getCachedDownTrackReturnsOnCall[len(fake.getCachedDownTrackArgsForCall)]
@@ -4902,6 +4966,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.getAdaptiveStreamMutex.RUnlock()
 	fake.getAudioLevelMutex.RLock()
 	defer fake.getAudioLevelMutex.RUnlock()
+	fake.getBufferFactoryMutex.RLock()
+	defer fake.getBufferFactoryMutex.RUnlock()
 	fake.getCachedDownTrackMutex.RLock()
 	defer fake.getCachedDownTrackMutex.RUnlock()
 	fake.getClientConfigurationMutex.RLock()