diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 165b515cf..a55c716f4 100644 --- a/pkg/rtc/room.go +++ b/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 { diff --git a/pkg/sfu/buffer/factory.go b/pkg/sfu/buffer/factory.go index 1f1c5a009..03486f5d6 100644 --- a/pkg/sfu/buffer/factory.go +++ b/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()