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

Cache ICE config in room manager. (#872)

Browse Source 
* Cache ICE config in room manager.

* mage generate

* Read ICE config within lock
This commit is contained in:
Raja Subramanian
2022-08-05 12:49:19 +05:30
committed by GitHub
parent 339181a534
commit 3f16018b62
4 changed files with 128 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+21
View File
@@ -152,6 +152,7 @@ type ParticipantImpl struct {
pendingDataChannels []*livekit.DataChannelInfo
onClose func(types.LocalParticipant, map[livekit.TrackID]livekit.ParticipantID)
onClaimsChanged func(participant types.LocalParticipant)
onICEConfigChanged func(participant types.LocalParticipant, iceConfig types.IceConfig)
activeCounter atomic.Int32
firstConnected atomic.Bool
@@ -538,7 +539,9 @@ func (p *ParticipantImpl) OnClose(callback func(types.LocalParticipant, map[live
}
func (p *ParticipantImpl) OnClaimsChanged(callback func(types.LocalParticipant)) {
p.lock.Lock()
p.onClaimsChanged = callback
p.lock.Unlock()
}
// HandleOffer an offer from remote participant, used when clients make the initial connection
@@ -856,6 +859,18 @@ func (p *ParticipantImpl) ICERestart(iceConfig *types.IceConfig) error {
})
}
func (p *ParticipantImpl) OnICEConfigChanged(f func(participant types.LocalParticipant, iceConfig types.IceConfig)) {
p.lock.Lock()
p.onICEConfigChanged = f
p.lock.Unlock()
}
func (p *ParticipantImpl) SetICEConfig(iceConfig types.IceConfig) {
p.lock.Lock()
p.iceConfig = iceConfig
p.lock.Unlock()
}
//
// signal connection methods
//
@@ -1328,7 +1343,13 @@ func (p *ParticipantImpl) handleConnectionFailed(isPrimary bool) {
p.lock.Lock()
p.iceConfig.PreferSubTcp = true
p.iceConfig.PreferPubTcp = true
onICEConfigChanged := p.onICEConfigChanged
iceConfig := p.iceConfig
p.lock.Unlock()
if onICEConfigChanged != nil {
onICEConfigChanged(p, iceConfig)
}
}
}
pkg/rtc/types/interfaces.go
+5 -2
View File
@@ -274,8 +274,8 @@ type LocalParticipant interface {
OnParticipantUpdate(callback func(LocalParticipant))
OnDataPacket(callback func(LocalParticipant, *livekit.DataPacket))
OnSubscribedTo(callback func(LocalParticipant, livekit.ParticipantID))
OnClose(_callback func(LocalParticipant, map[livekit.TrackID]livekit.ParticipantID))
OnClaimsChanged(_callback func(LocalParticipant))
OnClose(callback func(LocalParticipant, map[livekit.TrackID]livekit.ParticipantID))
OnClaimsChanged(callback func(LocalParticipant))
// session migration
SetMigrateState(s MigrateState)
@@ -292,6 +292,9 @@ type LocalParticipant interface {
EnqueueUnsubscribeTrack(trackID livekit.TrackID, willBeResumed bool, f func(subscriberID livekit.ParticipantID, willBeResumed bool) error)
ProcessSubscriptionRequestsQueue(trackID livekit.TrackID)
ClearInProgressAndProcessSubscriptionRequestsQueue(trackID livekit.TrackID)
SetICEConfig(iceConfig IceConfig)
OnICEConfigChanged(callback func(participant LocalParticipant, iceConfig IceConfig))
}
// Room is a container of participants, and can provide room-level actions
pkg/rtc/types/typesfakes/fake_local_participant.go
+78
View File
@@ -406,6 +406,11 @@ type FakeLocalParticipant struct {
onDataPacketArgsForCall []struct {
arg1 func(types.LocalParticipant, *livekit.DataPacket)
}
OnICEConfigChangedStub func(func(participant types.LocalParticipant, iceConfig types.IceConfig))
onICEConfigChangedMutex sync.RWMutex
onICEConfigChangedArgsForCall []struct {
arg1 func(participant types.LocalParticipant, iceConfig types.IceConfig)
}
OnParticipantUpdateStub func(func(types.LocalParticipant))
onParticipantUpdateMutex sync.RWMutex
onParticipantUpdateArgsForCall []struct {
@@ -538,6 +543,11 @@ type FakeLocalParticipant struct {
sendSpeakerUpdateReturnsOnCall map[int]struct {
result1 error
}
SetICEConfigStub func(types.IceConfig)
setICEConfigMutex sync.RWMutex
setICEConfigArgsForCall []struct {
arg1 types.IceConfig
}
SetMetadataStub func(string)
setMetadataMutex sync.RWMutex
setMetadataArgsForCall []struct {
@@ -2823,6 +2833,38 @@ func (fake *FakeLocalParticipant) OnDataPacketArgsForCall(i int) func(types.Loca
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) OnICEConfigChanged(arg1 func(participant types.LocalParticipant, iceConfig types.IceConfig)) {
fake.onICEConfigChangedMutex.Lock()
fake.onICEConfigChangedArgsForCall = append(fake.onICEConfigChangedArgsForCall, struct {
arg1 func(participant types.LocalParticipant, iceConfig types.IceConfig)
}{arg1})
stub := fake.OnICEConfigChangedStub
fake.recordInvocation("OnICEConfigChanged", []interface{}{arg1})
fake.onICEConfigChangedMutex.Unlock()
if stub != nil {
fake.OnICEConfigChangedStub(arg1)
}
}
func (fake *FakeLocalParticipant) OnICEConfigChangedCallCount() int {
fake.onICEConfigChangedMutex.RLock()
defer fake.onICEConfigChangedMutex.RUnlock()
return len(fake.onICEConfigChangedArgsForCall)
}
func (fake *FakeLocalParticipant) OnICEConfigChangedCalls(stub func(func(participant types.LocalParticipant, iceConfig types.IceConfig))) {
fake.onICEConfigChangedMutex.Lock()
defer fake.onICEConfigChangedMutex.Unlock()
fake.OnICEConfigChangedStub = stub
}
func (fake *FakeLocalParticipant) OnICEConfigChangedArgsForCall(i int) func(participant types.LocalParticipant, iceConfig types.IceConfig) {
fake.onICEConfigChangedMutex.RLock()
defer fake.onICEConfigChangedMutex.RUnlock()
argsForCall := fake.onICEConfigChangedArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) OnParticipantUpdate(arg1 func(types.LocalParticipant)) {
fake.onParticipantUpdateMutex.Lock()
fake.onParticipantUpdateArgsForCall = append(fake.onParticipantUpdateArgsForCall, struct {
@@ -3584,6 +3626,38 @@ func (fake *FakeLocalParticipant) SendSpeakerUpdateReturnsOnCall(i int, result1
}{result1}
}
func (fake *FakeLocalParticipant) SetICEConfig(arg1 types.IceConfig) {
fake.setICEConfigMutex.Lock()
fake.setICEConfigArgsForCall = append(fake.setICEConfigArgsForCall, struct {
arg1 types.IceConfig
}{arg1})
stub := fake.SetICEConfigStub
fake.recordInvocation("SetICEConfig", []interface{}{arg1})
fake.setICEConfigMutex.Unlock()
if stub != nil {
fake.SetICEConfigStub(arg1)
}
}
func (fake *FakeLocalParticipant) SetICEConfigCallCount() int {
fake.setICEConfigMutex.RLock()
defer fake.setICEConfigMutex.RUnlock()
return len(fake.setICEConfigArgsForCall)
}
func (fake *FakeLocalParticipant) SetICEConfigCalls(stub func(types.IceConfig)) {
fake.setICEConfigMutex.Lock()
defer fake.setICEConfigMutex.Unlock()
fake.SetICEConfigStub = stub
}
func (fake *FakeLocalParticipant) SetICEConfigArgsForCall(i int) types.IceConfig {
fake.setICEConfigMutex.RLock()
defer fake.setICEConfigMutex.RUnlock()
argsForCall := fake.setICEConfigArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) SetMetadata(arg1 string) {
fake.setMetadataMutex.Lock()
fake.setMetadataArgsForCall = append(fake.setMetadataArgsForCall, struct {
@@ -4667,6 +4741,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.onCloseMutex.RUnlock()
fake.onDataPacketMutex.RLock()
defer fake.onDataPacketMutex.RUnlock()
fake.onICEConfigChangedMutex.RLock()
defer fake.onICEConfigChangedMutex.RUnlock()
fake.onParticipantUpdateMutex.RLock()
defer fake.onParticipantUpdateMutex.RUnlock()
fake.onStateChangeMutex.RLock()
@@ -4699,6 +4775,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.sendRoomUpdateMutex.RUnlock()
fake.sendSpeakerUpdateMutex.RLock()
defer fake.sendSpeakerUpdateMutex.RUnlock()
fake.setICEConfigMutex.RLock()
defer fake.setICEConfigMutex.RUnlock()
fake.setMetadataMutex.RLock()
defer fake.setMetadataMutex.RUnlock()
fake.setMigrateInfoMutex.RLock()
pkg/service/roommanager.go
+24
View File
@@ -41,6 +41,8 @@ type RoomManager struct {
clientConfManager clientconfiguration.ClientConfigurationManager
rooms map[livekit.RoomName]*rtc.Room
iceConfigCache map[livekit.ParticipantIdentity]types.IceConfig
}
func NewLocalRoomManager(
@@ -67,6 +69,8 @@ func NewLocalRoomManager(
clientConfManager: clientConfManager,
rooms: make(map[livekit.RoomName]*rtc.Room),
iceConfigCache: make(map[livekit.ParticipantIdentity]types.IceConfig),
}
// hook up to router
@@ -264,6 +268,7 @@ func (r *RoomManager) StartSession(
if err != nil {
return err
}
r.setIceConfig(participant)
// join room
opts := rtc.ParticipantOptions{
@@ -310,6 +315,11 @@ func (r *RoomManager) StartSession(
logger.Errorw("could not refresh token", err)
}
})
participant.OnICEConfigChanged(func(participant types.LocalParticipant, iceConfig types.IceConfig) {
r.lock.Lock()
r.iceConfigCache[participant.Identity()] = iceConfig
r.lock.Unlock()
})
go r.rtcSessionWorker(room, participant, requestSource)
return nil
@@ -618,6 +628,20 @@ func (r *RoomManager) refreshToken(participant types.LocalParticipant) error {
return nil
}
func (r *RoomManager) setIceConfig(participant types.LocalParticipant) {
r.lock.RLock()
iceConfig, ok := r.iceConfigCache[participant.Identity()]
if !ok {
r.lock.RUnlock()
return
}
r.lock.RUnlock()
participant.SetICEConfig(iceConfig)
}
// ------------------------------------
func iceServerForStunServers(servers []string) *livekit.ICEServer {
iceServer := &livekit.ICEServer{}
for _, stunServer := range servers {