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

fallback to turn over tls when tcp short connection happen (#950)

Browse Source 
* fallback to tls when tcp failed

* go mod

* magefile
This commit is contained in:
cnderrauber
2022-08-24 20:42:56 +08:00
committed by GitHub
parent 00b131da50
commit 1350400c3a
8 changed files with 68 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -16,7 +16,7 @@ require (
github.com/gorilla/websocket v1.5.0
github.com/hashicorp/go-version v1.6.0
github.com/hashicorp/golang-lru v0.5.4
github.com/livekit/protocol v1.0.2-0.20220819200418-cd39f4ca4d10
github.com/livekit/protocol v1.0.2-0.20220824112019-a1c2809ddc67
github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995
github.com/mackerelio/go-osstat v0.2.2
github.com/magefile/mage v1.13.0
go.sum
+2 -2
View File
@@ -240,8 +240,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/lithammer/shortuuid/v3 v3.0.7 h1:trX0KTHy4Pbwo/6ia8fscyHoGA+mf1jWbPJVuvyJQQ8=
github.com/lithammer/shortuuid/v3 v3.0.7/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
github.com/livekit/protocol v1.0.2-0.20220819200418-cd39f4ca4d10 h1:6W0YbYSIa8xpcmYaNlYj86tnwCZTMrl8cs1Pd2y/uc4=
github.com/livekit/protocol v1.0.2-0.20220819200418-cd39f4ca4d10/go.mod h1:M+JZ29cEbUozppVg3LlDC9rZPE5d218Dft3h1qBP1qM=
github.com/livekit/protocol v1.0.2-0.20220824112019-a1c2809ddc67 h1:mh2hHoW2nMaqrEXPCy0G8utMTv8AbqFvzpetJ4YzYzw=
github.com/livekit/protocol v1.0.2-0.20220824112019-a1c2809ddc67/go.mod h1:M+JZ29cEbUozppVg3LlDC9rZPE5d218Dft3h1qBP1qM=
github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 h1:vOaY2qvfLihDyeZtnGGN1Law9wRrw8BMGCr1TygTvMw=
github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995/go.mod h1:116ych8UaEs9vfIE8n6iZCZ30iagUFTls0vRmC+Ix5U=
github.com/mackerelio/go-osstat v0.2.2 h1:7jVyXGXTkQL3+6lDVUDBY+Fpo8VQPfyOkZeXxxsXX4c=
pkg/rtc/participant.go
+10 -2
View File
@@ -233,6 +233,8 @@ func (p *ParticipantImpl) ConnectedAt() time.Time {
}
func (p *ParticipantImpl) GetClientConfiguration() *livekit.ClientConfiguration {
p.lock.RLock()
defer p.lock.RUnlock()
return p.params.ClientConf
}
@@ -1035,9 +1037,15 @@ func (p *ParticipantImpl) setupTransportManager() error {
}
tm.OnICEConfigChanged(func(iceConfig types.IceConfig) {
p.lock.RLock()
p.lock.Lock()
onICEConfigChanged := p.onICEConfigChanged
p.lock.RUnlock()
if iceConfig.PreferSub == types.PreferTls {
if p.params.ClientConf == nil {
p.params.ClientConf = &livekit.ClientConfiguration{}
}
p.params.ClientConf.ForceRelay = livekit.ClientConfigSetting_ENABLED
}
p.lock.Unlock()
if onICEConfigChanged != nil {
onICEConfigChanged(p, iceConfig)
pkg/rtc/room.go
+2 -2
View File
@@ -637,8 +637,8 @@ func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScen
case *livekit.SimulateScenario_SwitchCandidateProtocol:
r.Logger.Infow("simulating switch candidate protocol", "participant", participant.Identity())
participant.ICERestart(&types.IceConfig{
PreferSubTcp: scenario.SwitchCandidateProtocol == livekit.CandidateProtocol_TCP,
PreferPubTcp: scenario.SwitchCandidateProtocol == livekit.CandidateProtocol_TCP,
PreferSub: types.PreferCandidateType(scenario.SwitchCandidateProtocol),
PreferPub: types.PreferCandidateType(scenario.SwitchCandidateProtocol),
})
}
return nil
pkg/rtc/transport.go
-2
View File
@@ -473,8 +473,6 @@ func (t *PCTransport) onICEConnectionStateChange(state webrtc.ICEConnectionState
} else {
t.params.Logger.Infow("selected ICE candidate pair", "pair", pair)
}
case webrtc.ICEConnectionStateFailed:
t.handleConnectionFailed()
}
}
pkg/rtc/transportmanager.go
+31 -15
View File
@@ -46,6 +46,7 @@ type TransportManager struct {
pendingOfferPublisher *webrtc.SessionDescription
pendingDataChannelsPublisher []*livekit.DataChannelInfo
lastPublisherAnswer atomic.Value
iceConfig types.IceConfig
onPublisherInitialConnected func()
onSubscriberInitialConnected func()
@@ -101,7 +102,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
}
})
t.publisher.OnFailed(func(isShortLived bool) {
t.handleConnectionFailed(isShortLived)
if t.onAnyTransportFailed != nil {
t.onAnyTransportFailed()
}
@@ -422,17 +422,14 @@ func (t *TransportManager) OnICEConfigChanged(f func(iceConfig types.IceConfig))
func (t *TransportManager) SetICEConfig(iceConfig types.IceConfig) {
t.params.Logger.Infow("setting ICE config", "iceConfig", iceConfig)
if iceConfig.PreferPubTcp {
t.publisher.SetPreferTCP(true)
}
if iceConfig.PreferSubTcp {
t.subscriber.SetPreferTCP(true)
}
t.publisher.SetPreferTCP(iceConfig.PreferPub == types.PreferTcp)
t.subscriber.SetPreferTCP(iceConfig.PreferSub == types.PreferTcp)
t.lock.RLock()
t.lock.Lock()
onICEConfigChanged := t.onICEConfigChanged
t.lock.RUnlock()
t.iceConfig = iceConfig
t.lock.Unlock()
if onICEConfigChanged != nil {
onICEConfigChanged(iceConfig)
@@ -456,14 +453,33 @@ func (t *TransportManager) handleConnectionFailed(isShortLived bool) {
if !t.params.AllowTCPFallback || !isShortLived {
return
}
t.lock.RLock()
iceConfig := t.iceConfig
t.lock.RUnlock()
// irrespective of which one fails, force TCP on both as the other one might
var nextConfig types.IceConfig
// irrespective of which one fails, force prefer candidate on both as the other one might
// fail at a different time and cause another disruption
t.params.Logger.Infow("restricting transport to TCP on both peer connections")
t.SetICEConfig(types.IceConfig{
PreferPubTcp: true,
PreferSubTcp: true,
})
switch iceConfig.PreferSub {
case types.PreferNone:
t.params.Logger.Infow("restricting transport to TCP on both peer connections")
nextConfig = types.IceConfig{
PreferPub: types.PreferTcp,
PreferSub: types.PreferTcp,
}
case types.PreferTcp:
t.params.Logger.Infow("prefer transport to TLS on both peer connections")
nextConfig = types.IceConfig{
PreferPub: types.PreferTls,
PreferSub: types.PreferTls,
}
default:
return
}
t.SetICEConfig(nextConfig)
}
func (t *TransportManager) SetMigrateInfo(previousAnswer *webrtc.SessionDescription, dataChannels []*livekit.DataChannelInfo) {
pkg/rtc/types/interfaces.go
+10 -2
View File
@@ -195,9 +195,17 @@ type Participant interface {
DebugInfo() map[string]interface{}
}
type PreferCandidateType int
const (
PreferNone PreferCandidateType = iota
PreferTcp
PreferTls
)
type IceConfig struct {
PreferSubTcp bool
PreferPubTcp bool
PreferSub PreferCandidateType
PreferPub PreferCandidateType
}
//counterfeiter:generate . LocalParticipant
pkg/service/roommanager.go
+12 -6
View File
@@ -292,13 +292,13 @@ func (r *RoomManager) StartSession(
if err != nil {
return err
}
r.setIceConfig(participant)
iceConfig := r.setIceConfig(participant)
// join room
opts := rtc.ParticipantOptions{
AutoSubscribe: pi.AutoSubscribe,
}
if err = room.Join(participant, &opts, r.iceServersForRoom(protoRoom)); err != nil {
if err = room.Join(participant, &opts, r.iceServersForRoom(protoRoom, iceConfig.PreferSub == types.PreferTls)); err != nil {
pLogger.Errorw("could not join room", err)
_ = participant.Close(true, types.ParticipantCloseReasonJoinFailed)
return err
@@ -577,10 +577,15 @@ func (r *RoomManager) handleRTCMessage(ctx context.Context, roomName livekit.Roo
}
}
func (r *RoomManager) iceServersForRoom(ri *livekit.Room) []*livekit.ICEServer {
func (r *RoomManager) iceServersForRoom(ri *livekit.Room, tlsOnly bool) []*livekit.ICEServer {
var iceServers []*livekit.ICEServer
rtcConf := r.config.RTC
if tlsOnly && r.config.TURN.TLSPort == 0 {
logger.Warnw("tls only enabled but no turn tls config", nil)
tlsOnly = false
}
hasSTUN := false
if r.config.TURN.Enabled {
var urls []string
@@ -590,7 +595,7 @@ func (r *RoomManager) iceServersForRoom(ri *livekit.Room) []*livekit.ICEServer {
urls = append(urls, fmt.Sprintf("turn:%s:%d?transport=udp", r.config.RTC.NodeIP, r.config.TURN.UDPPort))
}
if r.config.TURN.TLSPort > 0 {
urls = append(urls, fmt.Sprintf("turns:%s:443?transport=tcp", r.config.TURN.Domain))
urls = append(urls, fmt.Sprintf("turns:%s:%d?transport=tcp", r.config.TURN.Domain, r.config.TURN.TLSPort))
}
if len(urls) > 0 {
iceServers = append(iceServers, &livekit.ICEServer{
@@ -654,17 +659,18 @@ func (r *RoomManager) refreshToken(participant types.LocalParticipant) error {
return nil
}
func (r *RoomManager) setIceConfig(participant types.LocalParticipant) {
func (r *RoomManager) setIceConfig(participant types.LocalParticipant) types.IceConfig {
r.lock.Lock()
iceConfigCacheEntry, ok := r.iceConfigCache[participant.Identity()]
if !ok || time.Since(iceConfigCacheEntry.modifiedAt) > iceConfigTTL {
delete(r.iceConfigCache, participant.Identity())
r.lock.Unlock()
return
return types.IceConfig{}
}
r.lock.Unlock()
participant.SetICEConfig(iceConfigCacheEntry.iceConfig)
return iceConfigCacheEntry.iceConfig
}
// ------------------------------------