From 48cf30ba23103d4434e1f6b74deecdd2250503ad Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Tue, 7 Mar 2023 09:13:15 +0800 Subject: [PATCH] Send disconnected participant update for reconnecting user (#1495) * Send disconnected participant update for reconnecting user * clean code --- pkg/rtc/participant.go | 12 ++++++-- pkg/rtc/participant_signal.go | 46 ++++++++++++++++++++++++++++--- pkg/rtc/transportmanager.go | 6 ++++ pkg/rtc/types/protocol_version.go | 6 +++- 4 files changed, 62 insertions(+), 8 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 182649571..101eeecac 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -50,6 +50,12 @@ type downTrackState struct { downTrack sfu.DownTrackState } +type participantUpdateInfo struct { + version uint32 + state livekit.ParticipantInfo_State + updatedAt time.Time +} + type ParticipantParams struct { Identity livekit.ParticipantIdentity Name livekit.ParticipantName @@ -125,7 +131,7 @@ type ParticipantImpl struct { queuedUpdates []*livekit.ParticipantInfo // cache of recently sent updates, to ensuring ordering by version // guarded by updateLock - updateCache *lru.Cache[livekit.ParticipantID, uint32] + updateCache *lru.Cache[livekit.ParticipantID, participantUpdateInfo] updateLock utils.Mutex dataChannelStats *telemetry.BytesTrackStats @@ -192,7 +198,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { var err error // keep last participants and when updates were sent - if p.updateCache, err = lru.New[livekit.ParticipantID, uint32](128); err != nil { + if p.updateCache, err = lru.New[livekit.ParticipantID, participantUpdateInfo](128); err != nil { return nil, err } @@ -905,7 +911,7 @@ func (p *ParticipantImpl) VerifySubscribeParticipantInfo(pID livekit.Participant // we have not sent a JoinResponse yet. metadata would be covered in JoinResponse return } - if v, ok := p.updateCache.Get(pID); ok && v >= version { + if info, ok := p.updateCache.Get(pID); ok && info.version >= version { return } diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 9e8a1997e..60c5e4e79 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -2,6 +2,7 @@ package rtc import ( "fmt" + "time" "github.com/pion/webrtc/v3" @@ -33,7 +34,7 @@ func (p *ParticipantImpl) SendJoinResponse(joinResponse *livekit.JoinResponse) e // keep track of participant updates and versions p.updateLock.Lock() for _, op := range joinResponse.OtherParticipants { - p.updateCache.Add(livekit.ParticipantID(op.Sid), op.Version) + p.updateCache.Add(livekit.ParticipantID(op.Sid), participantUpdateInfo{version: op.Version, state: op.State, updatedAt: time.Now()}) } p.updateLock.Unlock() @@ -84,13 +85,13 @@ func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit. if lastVersion, ok := p.updateCache.Get(pID); ok { // this is a message delivered out of order, a more recent version of the message had already been // sent. - if pi.Version < lastVersion { + if pi.Version < lastVersion.version { p.params.Logger.Debugw("skipping outdated participant update", "version", pi.Version, "lastVersion", lastVersion) isValid = false } } if isValid { - p.updateCache.Add(pID, pi.Version) + p.updateCache.Add(pID, participantUpdateInfo{version: pi.Version, state: pi.State, updatedAt: time.Now()}) validUpdates = append(validUpdates, pi) } } @@ -182,10 +183,47 @@ func (p *ParticipantImpl) SendReconnectResponse(reconnectResponse *livekit.Recon if !p.params.ClientInfo.CanHandleReconnectResponse() { return nil } - return p.writeMessage(&livekit.SignalResponse{ + if err := p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Reconnect{ Reconnect: reconnectResponse, }, + }); err != nil { + return err + } + + if p.params.ProtocolVersion.SupportHandlesDisconnectedUpdate() { + return p.sendDisconnectUpdatesForReconnect() + } + + return nil +} + +func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error { + lastSignalAt := p.TransportManager.LastSeenSignalAt() + var disconnectedParticipants []*livekit.ParticipantInfo + p.updateLock.Lock() + keys := p.updateCache.Keys() + for i := len(keys) - 1; i >= 0; i-- { + if info, ok := p.updateCache.Get(keys[i]); ok { + if info.updatedAt.Before(lastSignalAt) { + break + } else if info.state == livekit.ParticipantInfo_DISCONNECTED { + disconnectedParticipants = append(disconnectedParticipants, &livekit.ParticipantInfo{ + Sid: string(keys[i]), + Version: info.version, + State: livekit.ParticipantInfo_DISCONNECTED, + }) + } + } + } + p.updateLock.Unlock() + + return p.writeMessage(&livekit.SignalResponse{ + Message: &livekit.SignalResponse_Update{ + Update: &livekit.ParticipantUpdate{ + Participants: disconnectedParticipants, + }, + }, }) } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 1b5787c65..569bd6b8c 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -733,6 +733,12 @@ func (t *TransportManager) SinceLastSignal() time.Duration { return time.Since(t.lastSignalAt) } +func (t *TransportManager) LastSeenSignalAt() time.Time { + t.lock.RLock() + defer t.lock.RUnlock() + return t.lastSignalAt +} + func (t *TransportManager) canUseICETCP() bool { return t.params.TCPFallbackRTTThreshold == 0 || int(t.signalingRTT) < t.params.TCPFallbackRTTThreshold } diff --git a/pkg/rtc/types/protocol_version.go b/pkg/rtc/types/protocol_version.go index d1260cd90..4c0257734 100644 --- a/pkg/rtc/types/protocol_version.go +++ b/pkg/rtc/types/protocol_version.go @@ -2,7 +2,7 @@ package types type ProtocolVersion int -const CurrentProtocol = 8 +const CurrentProtocol = 9 func (v ProtocolVersion) SupportsPackedStreamId() bool { return v > 0 @@ -53,3 +53,7 @@ func (v ProtocolVersion) SupportsUnpublish() bool { func (v ProtocolVersion) SupportFastStart() bool { return v > 7 } + +func (v ProtocolVersion) SupportHandlesDisconnectedUpdate() bool { + return v > 8 +}