diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index b9611349b..c4ec51c31 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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) + } } } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 43035ce50..95c09956f 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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 diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 4b1fe9182..e79f086ae 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 9ec95bd2e..d171adfd7 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -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 {