From 3b236f0d72000c7862f9a4ac282de497e27aa0a0 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 19 Apr 2021 16:51:20 -0700 Subject: [PATCH] handle client reconnect with ICE restarts --- pkg/rtc/participant.go | 7 +++ pkg/rtc/transport.go | 21 ++++--- pkg/rtc/types/interfaces.go | 1 + pkg/rtc/types/typesfakes/fake_participant.go | 65 ++++++++++++++++++++ pkg/service/roommanager.go | 6 ++ 5 files changed, 92 insertions(+), 8 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 0a2bab78f..03137b564 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -360,6 +360,13 @@ func (p *ParticipantImpl) Negotiate() { p.subscriber.Negotiate() } +// ICERestart restarts subscriber ICE connections +func (p *ParticipantImpl) ICERestart() error { + return p.subscriber.CreateAndSendOffer(&webrtc.OfferOptions{ + ICERestart: true, + }) +} + // AddSubscriber subscribes op to all publishedTracks func (p *ParticipantImpl) AddSubscriber(op types.Participant) (int, error) { p.lock.RLock() diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 9f125b23a..530084bc2 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -6,9 +6,9 @@ import ( "time" "github.com/bep/debounce" + "github.com/livekit/livekit-server/pkg/logger" "github.com/pion/webrtc/v3" - "github.com/livekit/livekit-server/pkg/logger" livekit "github.com/livekit/livekit-server/proto" ) @@ -123,12 +123,16 @@ func (t *PCTransport) OnOffer(f func(sd webrtc.SessionDescription)) { } func (t *PCTransport) Negotiate() { - t.debouncedNegotiate(t.handleNegotiate) + t.debouncedNegotiate(func() { + if err := t.CreateAndSendOffer(nil); err != nil { + logger.Errorw("could not negotiate", "error", err) + } + }) } -func (t *PCTransport) handleNegotiate() { +func (t *PCTransport) CreateAndSendOffer(options *webrtc.OfferOptions) error { if t.onOffer == nil { - return + return nil } state := t.negotiationState.Load().(int) @@ -136,23 +140,24 @@ func (t *PCTransport) handleNegotiate() { if state == negotiationStateClient { logger.Debugw("skipping negotiation, trying again later") t.negotiationState.Store(negotiationRetry) - return + return nil } - offer, err := t.pc.CreateOffer(nil) + offer, err := t.pc.CreateOffer(options) if err != nil { logger.Errorw("could not create offer", "err", err) - return + return err } err = t.pc.SetLocalDescription(offer) if err != nil { logger.Errorw("could not set local description", "err", err) - return + return err } // indicate waiting for client t.negotiationState.Store(negotiationStateClient) t.onOffer(offer) + return nil } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 32ec31763..15feb6d74 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -36,6 +36,7 @@ type Participant interface { SetResponseSink(sink routing.MessageSink) SubscriberMediaEngine() *webrtc.MediaEngine Negotiate() + ICERestart() error AddTrack(clientId, name string, trackType livekit.TrackType) GetPublishedTracks() []PublishedTrack diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 80b245571..cb6f69ec6 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -157,6 +157,16 @@ type FakeParticipant struct { result1 webrtc.SessionDescription result2 error } + ICERestartStub func() error + iCERestartMutex sync.RWMutex + iCERestartArgsForCall []struct { + } + iCERestartReturns struct { + result1 error + } + iCERestartReturnsOnCall map[int]struct { + result1 error + } IDStub func() string iDMutex sync.RWMutex iDArgsForCall []struct { @@ -1096,6 +1106,59 @@ func (fake *FakeParticipant) HandleOfferReturnsOnCall(i int, result1 webrtc.Sess }{result1, result2} } +func (fake *FakeParticipant) ICERestart() error { + fake.iCERestartMutex.Lock() + ret, specificReturn := fake.iCERestartReturnsOnCall[len(fake.iCERestartArgsForCall)] + fake.iCERestartArgsForCall = append(fake.iCERestartArgsForCall, struct { + }{}) + stub := fake.ICERestartStub + fakeReturns := fake.iCERestartReturns + fake.recordInvocation("ICERestart", []interface{}{}) + fake.iCERestartMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) ICERestartCallCount() int { + fake.iCERestartMutex.RLock() + defer fake.iCERestartMutex.RUnlock() + return len(fake.iCERestartArgsForCall) +} + +func (fake *FakeParticipant) ICERestartCalls(stub func() error) { + fake.iCERestartMutex.Lock() + defer fake.iCERestartMutex.Unlock() + fake.ICERestartStub = stub +} + +func (fake *FakeParticipant) ICERestartReturns(result1 error) { + fake.iCERestartMutex.Lock() + defer fake.iCERestartMutex.Unlock() + fake.ICERestartStub = nil + fake.iCERestartReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeParticipant) ICERestartReturnsOnCall(i int, result1 error) { + fake.iCERestartMutex.Lock() + defer fake.iCERestartMutex.Unlock() + fake.ICERestartStub = nil + if fake.iCERestartReturnsOnCall == nil { + fake.iCERestartReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.iCERestartReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeParticipant) ID() string { fake.iDMutex.Lock() ret, specificReturn := fake.iDReturnsOnCall[len(fake.iDArgsForCall)] @@ -2211,6 +2274,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.handleAnswerMutex.RUnlock() fake.handleOfferMutex.RLock() defer fake.handleOfferMutex.RUnlock() + fake.iCERestartMutex.RLock() + defer fake.iCERestartMutex.RUnlock() fake.iDMutex.RLock() defer fake.iDMutex.RUnlock() fake.identityMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index d2f18a3bc..bf205ef35 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -196,6 +196,12 @@ func (r *RoomManager) StartSession(roomName string, pi routing.ParticipantInit, prevSink.Close() } participant.SetResponseSink(responseSink) + if err := participant.ICERestart(); err != nil { + logger.Warnw("could not restart ICE", + "participant", pi.Identity, + "error", err, + ) + } return } else { // we need to clean up the existing participant, so a new one can join