dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-29 05:29:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

queue restart ice offer while publisher pc gathering ice (#895)

Browse Source 
* queue restart ice offer while publisher pc gathering ice

* fix test case fail
This commit is contained in:
cnderrauber
2022-08-08 15:40:45 +08:00
committed by GitHub
parent 939757d2c8
commit 1d2bca373b
6 changed files with 172 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+18 -13
View File
@@ -255,6 +255,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
}
p.sendIceCandidate(c, livekit.SignalTarget_PUBLISHER)
})
p.publisher.OnRemoteDescripitonSettled(p.createPublsiherAnswerAndSend)
p.subscriber.pc.OnICECandidate(func(c *webrtc.ICECandidate) {
if c == nil || p.State() == livekit.ParticipantInfo_DISCONNECTED || p.MigrateState() == types.MigrateStateInit {
return
@@ -545,40 +546,44 @@ func (p *ParticipantImpl) OnClaimsChanged(callback func(types.LocalParticipant))
}
// HandleOffer an offer from remote participant, used when clients make the initial connection
func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer webrtc.SessionDescription, err error) {
func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) error {
p.lock.Lock()
if p.MigrateState() == types.MigrateStateInit {
p.pendingOffer = &sdp
p.lock.Unlock()
return
return nil
}
onParticipantUpdate := p.onParticipantUpdate
p.lock.Unlock()
p.params.Logger.Debugw("answering pub offer",
"state", p.State().String(),
// "sdp", sdp.SDP,
)
if err = p.publisher.SetRemoteDescription(sdp); err != nil {
if err := p.publisher.SetRemoteDescription(sdp); err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("answer", "error", "remote_description").Add(1)
return
return err
}
return nil
}
func (p *ParticipantImpl) createPublsiherAnswerAndSend() error {
p.lock.RLock()
onParticipantUpdate := p.onParticipantUpdate
p.lock.RUnlock()
p.configureReceiverDTX()
answer, err = p.publisher.pc.CreateAnswer(nil)
answer, err := p.publisher.pc.CreateAnswer(nil)
if err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("answer", "error", "create").Add(1)
err = errors.Wrap(err, "could not create answer")
return
return errors.Wrap(err, "could not create answer")
}
answer = p.publisher.FilterCandidates(answer)
if err = p.publisher.pc.SetLocalDescription(answer); err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("answer", "error", "local_description").Add(1)
err = errors.Wrap(err, "could not set local description")
return
return errors.Wrap(err, "could not set local description")
}
p.params.Logger.Debugw("sending answer to client")
@@ -590,7 +595,7 @@ func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer web
})
if err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("answer", "error", "write_message").Add(1)
return
return err
}
if p.isPublisher.Load() != p.CanPublish() {
@@ -607,7 +612,7 @@ func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer web
go p.handleMigrateMutedTrack()
}
return
return nil
}
func (p *ParticipantImpl) handleMigrateMutedTrack() {
@@ -830,7 +835,7 @@ func (p *ParticipantImpl) SetMigrateState(s types.MigrateState) {
}
if pendingOffer != nil {
_, err := p.HandleOffer(*pendingOffer)
err := p.HandleOffer(*pendingOffer)
if err != nil {
p.GetLogger().Errorw("could not handle offer", err)
}
pkg/rtc/participant_internal_test.go
+14 -2
View File
@@ -7,6 +7,7 @@ import (
"github.com/pion/webrtc/v3"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/proto"
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/livekit"
@@ -674,9 +675,20 @@ func TestDisableCodecs(t *testing.T) {
require.True(t, found264)
// negotiated codec should not contain h264
anwser, err := participant.HandleOffer(sdp)
sink := &routingfakes.FakeMessageSink{}
participant.SetResponseSink(sink)
var answer webrtc.SessionDescription
sink.WriteMessageStub = func(msg proto.Message) error {
if res, ok := msg.(*livekit.SignalResponse); ok {
if res.GetAnswer() != nil {
answer = FromProtoSessionDescription(res.GetAnswer())
}
}
return nil
}
err = participant.HandleOffer(sdp)
require.NoError(t, err)
require.NoError(t, pc.SetRemoteDescription(anwser), anwser.SDP, sdp.SDP)
require.NoError(t, pc.SetRemoteDescription(answer), answer.SDP, sdp.SDP)
codecs = transceiver.Receiver().GetParameters().Codecs
found264 = false
for _, c := range codecs {
pkg/rtc/signalhandler.go
+1 -1
View File
@@ -10,7 +10,7 @@ import (
func HandleParticipantSignal(room types.Room, participant types.LocalParticipant, req *livekit.SignalRequest, pLogger logger.Logger) error {
switch msg := req.Message.(type) {
case *livekit.SignalRequest_Offer:
_, err := participant.HandleOffer(FromProtoSessionDescription(msg.Offer))
err := participant.HandleOffer(FromProtoSessionDescription(msg.Offer))
if err != nil {
pLogger.Errorw("could not handle offer", err)
return err
pkg/rtc/transport.go
+124 -14
View File
@@ -2,6 +2,7 @@ package rtc
import (
"errors"
"fmt"
"strings"
"sync"
"time"
@@ -62,18 +63,19 @@ type PCTransport struct {
pc *webrtc.PeerConnection
me *webrtc.MediaEngine
lock sync.RWMutex
iceConnectedAt time.Time
pendingCandidates []webrtc.ICECandidateInit
debouncedNegotiate func(func())
negotiationPending map[livekit.ParticipantID]bool
onOffer func(offer webrtc.SessionDescription)
restartAfterGathering bool
restartAtNextOffer bool
negotiationState int
negotiateCounter atomic.Int32
signalStateCheckTimer *time.Timer
onNegotiationFailed func()
lock sync.RWMutex
iceConnectedAt time.Time
pendingCandidates []webrtc.ICECandidateInit
debouncedNegotiate func(func())
negotiationPending map[livekit.ParticipantID]bool
onOffer func(offer webrtc.SessionDescription)
onRemoteDescripitonSettled func() error
restartAfterGathering bool
restartAtNextOffer bool
negotiationState int
negotiateCounter atomic.Int32
signalStateCheckTimer *time.Timer
onNegotiationFailed func()
// stream allocator for subscriber PC
streamAllocator *sfu.StreamAllocator
@@ -81,6 +83,9 @@ type PCTransport struct {
previousAnswer *webrtc.SessionDescription
preferTCP bool
currentOfferIceCredential string // ice user:pwd, for publish side ice restart checking
pendingRestartIceOffer *webrtc.SessionDescription
}
type TransportParams struct {
@@ -278,12 +283,22 @@ func (t *PCTransport) createPeerConnection() error {
if state == webrtc.ICEGathererStateComplete {
go func() {
t.lock.Lock()
defer t.lock.Unlock()
if t.restartAfterGathering {
t.params.Logger.Debugw("restarting ICE after ICE gathering")
if err := t.createAndSendOffer(&webrtc.OfferOptions{ICERestart: true}); err != nil {
t.params.Logger.Warnw("could not restart ICE", err)
}
t.lock.Unlock()
} else if t.pendingRestartIceOffer != nil {
t.params.Logger.Debugw("accept remote restart ice offer after ICE gathering")
offer := t.pendingRestartIceOffer
t.pendingRestartIceOffer = nil
t.lock.Unlock()
if err := t.SetRemoteDescription(*offer); err != nil {
t.params.Logger.Warnw("could not accept remote restart ice offer", err)
}
} else {
t.lock.Unlock()
}
}()
}
@@ -337,12 +352,37 @@ func (t *PCTransport) Close() {
func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error {
t.lock.Lock()
defer t.lock.Unlock()
var (
iceCredential string
offerRestartICE bool
)
if sd.Type == webrtc.SDPTypeOffer {
var err error
iceCredential, offerRestartICE, err = t.isRemoteOfferRestartICE(sd)
if err != nil {
t.Logger().Errorw("check remote offer restart ice failed", err)
t.lock.Unlock()
return err
}
}
if offerRestartICE && t.pc.ICEGatheringState() == webrtc.ICEGatheringStateGathering {
t.Logger().Debugw("remote offer restart ice while ice gathering")
t.pendingRestartIceOffer = &sd
t.lock.Unlock()
return nil
}
if err := t.pc.SetRemoteDescription(sd); err != nil {
t.lock.Unlock()
return err
}
if t.currentOfferIceCredential == "" || offerRestartICE {
t.currentOfferIceCredential = iceCredential
}
// negotiated, reset flag
lastState := t.negotiationState
t.negotiationState = negotiationStateNone
@@ -354,6 +394,7 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error {
for _, c := range t.pendingCandidates {
if err := t.pc.AddICECandidate(c); err != nil {
t.lock.Unlock()
return err
}
}
@@ -366,14 +407,42 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error {
t.params.Logger.Errorw("could not negotiate", err)
}
}
onRemoteDescripitonSettled := t.onRemoteDescripitonSettled
t.lock.Unlock()
if onRemoteDescripitonSettled != nil {
return onRemoteDescripitonSettled()
}
return nil
}
func (t *PCTransport) isRemoteOfferRestartICE(sd webrtc.SessionDescription) (string, bool, error) {
parsed, err := sd.Unmarshal()
if err != nil {
return "", false, err
}
user, pwd, err := extractICECredential(parsed)
if err != nil {
return "", false, err
}
credential := fmt.Sprintf("%s:%s", user, pwd)
// ice credential changed, remote offer restart ice
restartICE := t.currentOfferIceCredential != "" && t.currentOfferIceCredential != credential
return credential, restartICE, nil
}
// OnOffer is called when the PeerConnection starts negotiation and prepares an offer
func (t *PCTransport) OnOffer(f func(sd webrtc.SessionDescription)) {
t.onOffer = f
}
func (t *PCTransport) OnRemoteDescripitonSettled(f func() error) {
t.lock.Lock()
t.onRemoteDescripitonSettled = f
t.lock.Unlock()
}
func (t *PCTransport) OnNegotiationFailed(f func()) {
t.onNegotiationFailed = f
}
@@ -799,3 +868,44 @@ func extractDTLSRole(desc *sdp.SessionDescription) webrtc.DTLSRole {
//
return webrtc.DTLSRoleClient
}
func extractICECredential(desc *sdp.SessionDescription) (string, string, error) {
remotePwds := []string{}
remoteUfrags := []string{}
if ufrag, haveUfrag := desc.Attribute("ice-ufrag"); haveUfrag {
remoteUfrags = append(remoteUfrags, ufrag)
}
if pwd, havePwd := desc.Attribute("ice-pwd"); havePwd {
remotePwds = append(remotePwds, pwd)
}
for _, m := range desc.MediaDescriptions {
if ufrag, haveUfrag := m.Attribute("ice-ufrag"); haveUfrag {
remoteUfrags = append(remoteUfrags, ufrag)
}
if pwd, havePwd := m.Attribute("ice-pwd"); havePwd {
remotePwds = append(remotePwds, pwd)
}
}
if len(remoteUfrags) == 0 {
return "", "", webrtc.ErrSessionDescriptionMissingIceUfrag
} else if len(remotePwds) == 0 {
return "", "", webrtc.ErrSessionDescriptionMissingIcePwd
}
for _, m := range remoteUfrags {
if m != remoteUfrags[0] {
return "", "", webrtc.ErrSessionDescriptionConflictingIceUfrag
}
}
for _, m := range remotePwds {
if m != remotePwds[0] {
return "", "", webrtc.ErrSessionDescriptionConflictingIcePwd
}
}
return remoteUfrags[0], remotePwds[0], nil
}
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -229,7 +229,7 @@ type LocalParticipant interface {
AddICECandidate(candidate webrtc.ICECandidateInit, target livekit.SignalTarget) error
HandleOffer(sdp webrtc.SessionDescription) (answer webrtc.SessionDescription, err error)
HandleOffer(sdp webrtc.SessionDescription) error
AddTrack(req *livekit.AddTrackRequest)
SetTrackMuted(trackID livekit.TrackID, muted bool, fromAdmin bool)
pkg/rtc/types/typesfakes/fake_local_participant.go
+14 -19
View File
@@ -270,18 +270,16 @@ type FakeLocalParticipant struct {
handleAnswerReturnsOnCall map[int]struct {
result1 error
}
HandleOfferStub func(webrtc.SessionDescription) (webrtc.SessionDescription, error)
HandleOfferStub func(webrtc.SessionDescription) error
handleOfferMutex sync.RWMutex
handleOfferArgsForCall []struct {
arg1 webrtc.SessionDescription
}
handleOfferReturns struct {
result1 webrtc.SessionDescription
result2 error
result1 error
}
handleOfferReturnsOnCall map[int]struct {
result1 webrtc.SessionDescription
result2 error
result1 error
}
HiddenStub func() bool
hiddenMutex sync.RWMutex
@@ -2087,7 +2085,7 @@ func (fake *FakeLocalParticipant) HandleAnswerReturnsOnCall(i int, result1 error
}{result1}
}
func (fake *FakeLocalParticipant) HandleOffer(arg1 webrtc.SessionDescription) (webrtc.SessionDescription, error) {
func (fake *FakeLocalParticipant) HandleOffer(arg1 webrtc.SessionDescription) error {
fake.handleOfferMutex.Lock()
ret, specificReturn := fake.handleOfferReturnsOnCall[len(fake.handleOfferArgsForCall)]
fake.handleOfferArgsForCall = append(fake.handleOfferArgsForCall, struct {
@@ -2101,9 +2099,9 @@ func (fake *FakeLocalParticipant) HandleOffer(arg1 webrtc.SessionDescription) (w
return stub(arg1)
}
if specificReturn {
return ret.result1, ret.result2
return ret.result1
}
return fakeReturns.result1, fakeReturns.result2
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) HandleOfferCallCount() int {
@@ -2112,7 +2110,7 @@ func (fake *FakeLocalParticipant) HandleOfferCallCount() int {
return len(fake.handleOfferArgsForCall)
}
func (fake *FakeLocalParticipant) HandleOfferCalls(stub func(webrtc.SessionDescription) (webrtc.SessionDescription, error)) {
func (fake *FakeLocalParticipant) HandleOfferCalls(stub func(webrtc.SessionDescription) error) {
fake.handleOfferMutex.Lock()
defer fake.handleOfferMutex.Unlock()
fake.HandleOfferStub = stub
@@ -2125,30 +2123,27 @@ func (fake *FakeLocalParticipant) HandleOfferArgsForCall(i int) webrtc.SessionDe
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) HandleOfferReturns(result1 webrtc.SessionDescription, result2 error) {
func (fake *FakeLocalParticipant) HandleOfferReturns(result1 error) {
fake.handleOfferMutex.Lock()
defer fake.handleOfferMutex.Unlock()
fake.HandleOfferStub = nil
fake.handleOfferReturns = struct {
result1 webrtc.SessionDescription
result2 error
}{result1, result2}
result1 error
}{result1}
}
func (fake *FakeLocalParticipant) HandleOfferReturnsOnCall(i int, result1 webrtc.SessionDescription, result2 error) {
func (fake *FakeLocalParticipant) HandleOfferReturnsOnCall(i int, result1 error) {
fake.handleOfferMutex.Lock()
defer fake.handleOfferMutex.Unlock()
fake.HandleOfferStub = nil
if fake.handleOfferReturnsOnCall == nil {
fake.handleOfferReturnsOnCall = make(map[int]struct {
result1 webrtc.SessionDescription
result2 error
result1 error
})
}
fake.handleOfferReturnsOnCall[i] = struct {
result1 webrtc.SessionDescription
result2 error
}{result1, result2}
result1 error
}{result1}
}
func (fake *FakeLocalParticipant) Hidden() bool {