diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index a0fb8c91e..0534b0ae5 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -136,7 +136,7 @@ func (r *Room) GetActiveSpeakers() []*livekit.SpeakerInfo { return speakers } -func (r *Room) GetBufferFactor() *buffer.Factory { +func (r *Room) GetBufferFactory() *buffer.Factory { return r.bufferFactory } diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 9d1719c38..2f37549e4 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -30,6 +30,8 @@ type UptrackManager struct { rtcpCh chan []rtcp.Packet pliThrottle *pliThrottle + closed bool + // hold reference for MediaTrack twcc *twcc.Responder @@ -70,13 +72,16 @@ func (u *UptrackManager) Close() { u.lock.Lock() defer u.lock.Unlock() + u.closed = true + // remove all subscribers for _, t := range u.publishedTracks { - // skip updates t.RemoveAllSubscribers() } - close(u.rtcpCh) + if len(u.publishedTracks) == 0 { + close(u.rtcpCh) + } } func (u *UptrackManager) ToProto() []*livekit.TrackInfo { @@ -432,6 +437,11 @@ func (u *UptrackManager) handleTrackPublished(track types.PublishedTrack) { delete(u.publishedTracks, trackSid) delete(u.pendingSubscriptions, trackSid) // not modifying subscription permissions, will get reset on next update from participant + + // as rtcpCh handles RTCP for all published tracks, close only after all published tracks are closed + if u.closed && len(u.publishedTracks) == 0 { + close(u.rtcpCh) + } u.lock.Unlock() // only send this when client is in a ready state if u.onTrackUpdated != nil { diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index baf8bab31..bca81e9d1 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -225,7 +225,7 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName string, pi rout pv := types.ProtocolVersion(pi.Client.Protocol) rtcConf := *r.rtcConfig - rtcConf.SetBufferFactory(room.GetBufferFactor()) + rtcConf.SetBufferFactory(room.GetBufferFactory()) sid := utils.NewGuid(utils.ParticipantPrefix) pLogger := rtc.LoggerWithParticipant(room.Logger, pi.Identity, sid) participant, err = rtc.NewParticipant(rtc.ParticipantParams{