Fix potential ssrc collision between participants (#1173)

pkg/rtc/room.go

@@ -51,7 +51,7 @@ type Room struct {
 	// map of identity -> Participant
 	participants    map[livekit.ParticipantIdentity]types.LocalParticipant
 	participantOpts map[livekit.ParticipantIdentity]*ParticipantOptions
-	bufferFactory   *buffer.Factory
+	bufferFactory   *buffer.FactoryOfBufferFactory
 
 	// batch update participant info for non-publishers
 	batchedUpdates   map[livekit.ParticipantIdentity]*livekit.ParticipantInfo
@@ -93,7 +93,7 @@ func NewRoom(
 		serverInfo:      serverInfo,
 		participants:    make(map[livekit.ParticipantIdentity]types.LocalParticipant),
 		participantOpts: make(map[livekit.ParticipantIdentity]*ParticipantOptions),
-		bufferFactory:   buffer.NewBufferFactory(config.Receiver.PacketBufferSize),
+		bufferFactory:   buffer.NewFactoryOfBufferFactory(config.Receiver.PacketBufferSize),
 		batchedUpdates:  make(map[livekit.ParticipantIdentity]*livekit.ParticipantInfo),
 		closed:          make(chan struct{}),
 	}
@@ -183,7 +183,7 @@ func (r *Room) GetActiveSpeakers() []*livekit.SpeakerInfo {
 }
 
 func (r *Room) GetBufferFactory() *buffer.Factory {
-	return r.bufferFactory
+	return r.bufferFactory.CreateBufferFactory()
 }
 
 func (r *Room) FirstJoinedAt() int64 {

pkg/sfu/buffer/factory.go

@@ -9,16 +9,13 @@ import (
 	"github.com/livekit/mediatransportutil/pkg/bucket"
 )
 
-type Factory struct {
-	sync.RWMutex
-	videoPool   *sync.Pool
-	audioPool   *sync.Pool
-	rtpBuffers  map[uint32]*Buffer
-	rtcpReaders map[uint32]*RTCPReader
+type FactoryOfBufferFactory struct {
+	videoPool *sync.Pool
+	audioPool *sync.Pool
 }
 
-func NewBufferFactory(trackingPackets int) *Factory {
-	return &Factory{
+func NewFactoryOfBufferFactory(trackingPackets int) *FactoryOfBufferFactory {
+	return &FactoryOfBufferFactory{
 		videoPool: &sync.Pool{
 			New: func() interface{} {
 				b := make([]byte, trackingPackets*bucket.MaxPktSize)
@@ -31,11 +28,26 @@ func NewBufferFactory(trackingPackets int) *Factory {
 				return &b
 			},
 		},
+	}
+}
+
+func (f *FactoryOfBufferFactory) CreateBufferFactory() *Factory {
+	return &Factory{
+		videoPool:   f.videoPool,
+		audioPool:   f.audioPool,
 		rtpBuffers:  make(map[uint32]*Buffer),
 		rtcpReaders: make(map[uint32]*RTCPReader),
 	}
 }
 
+type Factory struct {
+	sync.RWMutex
+	videoPool   *sync.Pool
+	audioPool   *sync.Pool
+	rtpBuffers  map[uint32]*Buffer
+	rtcpReaders map[uint32]*RTCPReader
+}
+
 func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser {
 	f.Lock()
 	defer f.Unlock()