diff --git a/go.mod b/go.mod index c8281d350..44fd1a0f2 100644 --- a/go.mod +++ b/go.mod @@ -13,10 +13,10 @@ require ( github.com/google/wire v0.5.0 github.com/gorilla/websocket v1.4.2 github.com/hashicorp/golang-lru v0.5.4 - github.com/livekit/protocol v0.13.3-0.20220530102138-51a8116f88b2 + github.com/livekit/protocol v0.13.3-0.20220607094330-e36e6426a094 github.com/livekit/rtcscore-go v0.0.0-20220524203225-dfd1ba40744a github.com/mackerelio/go-osstat v0.2.1 - github.com/magefile/mage v1.11.0 + github.com/magefile/mage v1.13.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 github.com/olekukonko/tablewriter v0.0.5 diff --git a/go.sum b/go.sum index 66fd7a88d..3fc277e16 100644 --- a/go.sum +++ b/go.sum @@ -135,12 +135,16 @@ github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= github.com/livekit/protocol v0.13.3-0.20220530102138-51a8116f88b2 h1:ioAZAmtoQnHELXBKOQr041ro8GwKcZ7lgNuDNbLG1vo= github.com/livekit/protocol v0.13.3-0.20220530102138-51a8116f88b2/go.mod h1:BLtSeVmn2rLP37xjzw7gHgaAmkWl3L/L9bPvgSbaOfo= +github.com/livekit/protocol v0.13.3-0.20220607094330-e36e6426a094 h1:DwZfNfyg5LaX3N1Eh19rcycTcyzpgqH0DpEkhB1CnI8= +github.com/livekit/protocol v0.13.3-0.20220607094330-e36e6426a094/go.mod h1:BLtSeVmn2rLP37xjzw7gHgaAmkWl3L/L9bPvgSbaOfo= github.com/livekit/rtcscore-go v0.0.0-20220524203225-dfd1ba40744a h1:cENjhGfslLSDV07gt8ASy47Wd12Q0kBS7hsdunyQ62I= github.com/livekit/rtcscore-go v0.0.0-20220524203225-dfd1ba40744a/go.mod h1:116ych8UaEs9vfIE8n6iZCZ30iagUFTls0vRmC+Ix5U= github.com/mackerelio/go-osstat v0.2.1 h1:5AeAcBEutEErAOlDz6WCkEvm6AKYgHTUQrfwm5RbeQc= github.com/mackerelio/go-osstat v0.2.1/go.mod h1:UzRL8dMCCTqG5WdRtsxbuljMpZt9PCAGXqxPst5QtaY= github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls= github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= +github.com/magefile/mage v1.13.0 h1:XtLJl8bcCM7EFoO8FyH8XK3t7G5hQAeK+i4tq+veT9M= +github.com/magefile/mage v1.13.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0= github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI= github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 1bf80ecb5..fb296ed0d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -136,6 +136,7 @@ type ParticipantImpl struct { activeCounter atomic.Int32 firstConnected atomic.Bool + iceConfig types.IceConfig } func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { @@ -597,6 +598,22 @@ func (p *ParticipantImpl) HandleAnswer(sdp webrtc.SessionDescription) error { // AddICECandidate adds candidates for remote peer func (p *ParticipantImpl) AddICECandidate(candidate webrtc.ICECandidateInit, target livekit.SignalTarget) error { + var filterOut bool + p.lock.RLock() + if target == livekit.SignalTarget_SUBSCRIBER { + if p.iceConfig.PreferSubTcp && !strings.Contains(candidate.Candidate, "tcp") { + filterOut = true + } + } else if target == livekit.SignalTarget_PUBLISHER { + if p.iceConfig.PreferPubTcp && !strings.Contains(candidate.Candidate, "tcp") { + filterOut = true + } + } + p.lock.RUnlock() + if filterOut { + return nil + } + var err error if target == livekit.SignalTarget_PUBLISHER { err = p.publisher.AddICECandidate(candidate) @@ -712,7 +729,13 @@ func (p *ParticipantImpl) MigrateState() types.MigrateState { } // ICERestart restarts subscriber ICE connections -func (p *ParticipantImpl) ICERestart() error { +func (p *ParticipantImpl) ICERestart(iceConfig *types.IceConfig) error { + if iceConfig != nil { + p.lock.Lock() + p.iceConfig = *iceConfig + p.lock.Unlock() + } + if p.subscriber.pc.RemoteDescription() == nil { // not connected, skip return nil diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 9a2d7242d..d4277534d 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -179,6 +179,22 @@ func (p *ParticipantImpl) SendRefreshToken(token string) error { } func (p *ParticipantImpl) sendIceCandidate(c *webrtc.ICECandidate, target livekit.SignalTarget) { + var filterOut bool + p.lock.RLock() + if target == livekit.SignalTarget_SUBSCRIBER { + if p.iceConfig.PreferSubTcp && c.Protocol != webrtc.ICEProtocolTCP { + filterOut = true + } + } else if target == livekit.SignalTarget_PUBLISHER { + if p.iceConfig.PreferPubTcp && c.Protocol != webrtc.ICEProtocolTCP { + filterOut = true + } + } + p.lock.RUnlock() + if filterOut { + return + } + ci := c.ToJSON() // write candidate diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 9e52d7949..3fdb619a7 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -334,7 +334,7 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing. return err } - if err := p.ICERestart(); err != nil { + if err := p.ICERestart(nil); err != nil { return err } return nil @@ -620,6 +620,15 @@ func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScen if err := participant.Close(true); err != nil { return err } + + case *livekit.SimulateScenario_SwitchCandidateProtocol: + r.Logger.Infow("simulating switch candidate protocol", "participant", participant.Identity()) + if err := participant.ICERestart(&types.IceConfig{ + PreferSubTcp: scenario.SwitchCandidateProtocol == livekit.CandidateProtocol_TCP, + PreferPubTcp: scenario.SwitchCandidateProtocol == livekit.CandidateProtocol_TCP, + }); err != nil { + return err + } } return nil } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index de4a9ac66..b8228af53 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -91,6 +91,11 @@ type Participant interface { DebugInfo() map[string]interface{} } +type IceConfig struct { + PreferSubTcp bool + PreferPubTcp bool +} + //counterfeiter:generate . LocalParticipant type LocalParticipant interface { Participant @@ -127,7 +132,7 @@ type LocalParticipant interface { SubscriberPC() *webrtc.PeerConnection HandleAnswer(sdp webrtc.SessionDescription) error Negotiate(force bool) - ICERestart() error + ICERestart(iceConfig *IceConfig) error AddSubscribedTrack(st SubscribedTrack) RemoveSubscribedTrack(st SubscribedTrack) UpdateSubscribedTrackSettings(trackID livekit.TrackID, settings *livekit.UpdateTrackSettings) error diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index edabb8df7..e5eaf564e 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -248,9 +248,10 @@ type FakeLocalParticipant struct { hiddenReturnsOnCall map[int]struct { result1 bool } - ICERestartStub func() error + ICERestartStub func(*types.IceConfig) error iCERestartMutex sync.RWMutex iCERestartArgsForCall []struct { + arg1 *types.IceConfig } iCERestartReturns struct { result1 error @@ -1873,17 +1874,18 @@ func (fake *FakeLocalParticipant) HiddenReturnsOnCall(i int, result1 bool) { }{result1} } -func (fake *FakeLocalParticipant) ICERestart() error { +func (fake *FakeLocalParticipant) ICERestart(arg1 *types.IceConfig) error { fake.iCERestartMutex.Lock() ret, specificReturn := fake.iCERestartReturnsOnCall[len(fake.iCERestartArgsForCall)] fake.iCERestartArgsForCall = append(fake.iCERestartArgsForCall, struct { - }{}) + arg1 *types.IceConfig + }{arg1}) stub := fake.ICERestartStub fakeReturns := fake.iCERestartReturns - fake.recordInvocation("ICERestart", []interface{}{}) + fake.recordInvocation("ICERestart", []interface{}{arg1}) fake.iCERestartMutex.Unlock() if stub != nil { - return stub() + return stub(arg1) } if specificReturn { return ret.result1 @@ -1897,12 +1899,19 @@ func (fake *FakeLocalParticipant) ICERestartCallCount() int { return len(fake.iCERestartArgsForCall) } -func (fake *FakeLocalParticipant) ICERestartCalls(stub func() error) { +func (fake *FakeLocalParticipant) ICERestartCalls(stub func(*types.IceConfig) error) { fake.iCERestartMutex.Lock() defer fake.iCERestartMutex.Unlock() fake.ICERestartStub = stub } +func (fake *FakeLocalParticipant) ICERestartArgsForCall(i int) *types.IceConfig { + fake.iCERestartMutex.RLock() + defer fake.iCERestartMutex.RUnlock() + argsForCall := fake.iCERestartArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) ICERestartReturns(result1 error) { fake.iCERestartMutex.Lock() defer fake.iCERestartMutex.Unlock()