From 17fb420c2faf78b1252c306a7fbd30d088bb29e0 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 19 Jun 2022 10:19:43 +0530 Subject: [PATCH] Send connection quality updates only if there is a change (#767) --- pkg/rtc/room.go | 50 +++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 42 insertions(+), 8 deletions(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 3fdb619a7..2d933f721 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -932,19 +932,51 @@ func (r *Room) connectionQualityWorker() { ticker := time.NewTicker(connectionquality.UpdateInterval) defer ticker.Stop() + prevConnectionInfos := make(map[livekit.ParticipantID]*livekit.ConnectionQualityInfo) // send updates to only users that are subscribed to each other for !r.IsClosed() { <-ticker.C participants := r.GetParticipants() - connectionInfos := make(map[livekit.ParticipantID]*livekit.ConnectionQualityInfo, len(participants)) + nowConnectionInfos := make(map[livekit.ParticipantID]*livekit.ConnectionQualityInfo, len(participants)) for _, p := range participants { if p.State() != livekit.ParticipantInfo_ACTIVE { continue } - connectionInfos[p.ID()] = p.GetConnectionQuality() + nowConnectionInfos[p.ID()] = p.GetConnectionQuality() + } + + // send an update if there is a change + // - new participant + // - quality change + // NOTE: participant leaving is explicitly omitted as `leave` signal notifies that a participant is not in the room anymore + sendUpdate := false + for _, p := range participants { + pID := p.ID() + prevInfo, prevOk := prevConnectionInfos[pID] + nowInfo, nowOk := nowConnectionInfos[pID] + if !nowOk { + // participant is not ACTIVE any more + continue + } + if !prevOk || nowInfo.Quality != prevInfo.Quality { + // new entrant OR change in quality + sendUpdate = true + break + } + } + + if !sendUpdate { + prevConnectionInfos = nowConnectionInfos + continue + } + + maybeAddToUpdate := func(pID livekit.ParticipantID, update *livekit.ConnectionQualityUpdate) { + if nowInfo, nowOk := nowConnectionInfos[pID]; nowOk { + update.Updates = append(update.Updates, nowInfo) + } } for _, op := range participants { @@ -954,21 +986,23 @@ func (r *Room) connectionQualityWorker() { update := &livekit.ConnectionQualityUpdate{} // send to user itself - if info, ok := connectionInfos[op.ID()]; ok { - update.Updates = append(update.Updates, info) - } + maybeAddToUpdate(op.ID(), update) // add connection quality of other participants its subscribed to for _, sid := range op.GetSubscribedParticipants() { - if info, ok := connectionInfos[sid]; ok { - update.Updates = append(update.Updates, info) - } + maybeAddToUpdate(sid, update) + } + if len(update.Updates) == 0 { + // no change + continue } if err := op.SendConnectionQualityUpdate(update); err != nil { r.Logger.Warnw("could not send connection quality update", err, "participant", op.Identity()) } } + + prevConnectionInfos = nowConnectionInfos } }