From 16896e4d665c63a2d58d5b740d7f5b7b1745ff39 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 28 Aug 2024 04:00:28 +0000 Subject: [PATCH] Add FastPublish in JoinResponse (#2964) Set FastPublish in JoinResponse for participant can publish and no ice fallback setting. --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/participant.go | 4 + pkg/rtc/room.go | 3 + pkg/rtc/transportmanager.go | 9 ++ pkg/rtc/types/interfaces.go | 2 + .../typesfakes/fake_local_participant.go | 130 ++++++++++++++++++ 7 files changed, 151 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index ebc760c13..57632acb2 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598 - github.com/livekit/protocol v1.20.1-0.20240823101247-81856d28076a + github.com/livekit/protocol v1.20.1-0.20240828034551-2ecf29c2305c github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a github.com/mackerelio/go-osstat v0.2.5 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index ad6349d28..341647c1e 100644 --- a/go.sum +++ b/go.sum @@ -169,8 +169,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598 h1:yLlkHk2feSLHstD9n4VKg7YEBR4rLODTI4WE8gNBEnQ= github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598/go.mod h1:jwKUCmObuiEDH0iiuJHaGMXwRs3RjrB4G6qqgkr/5oE= -github.com/livekit/protocol v1.20.1-0.20240823101247-81856d28076a h1:S3nK/EXOfKdKDZXSwLhzyH9w6gZj5CGeegC4UpuSz78= -github.com/livekit/protocol v1.20.1-0.20240823101247-81856d28076a/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI= +github.com/livekit/protocol v1.20.1-0.20240828034551-2ecf29c2305c h1:PSqOiH80cUVKUs4Hu/eEKwc97TKKXLq4+8iKusfgCDQ= +github.com/livekit/protocol v1.20.1-0.20240828034551-2ecf29c2305c/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI= github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a h1:EQAHmcYEGlc6V517cQ3Iy0+jHgP6+tM/B4l2vGuLpQo= github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a/go.mod h1:CQUBSPfYYAaevg1TNCc6/aYsa8DJH4jSRFdCeSZk5u0= github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 5179660d2..37f90b127 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1179,6 +1179,10 @@ func (p *ParticipantImpl) IsPublisher() bool { return p.isPublisher.Load() } +func (p *ParticipantImpl) CanPublish() bool { + return p.grants.Load().Video.GetCanPublish() +} + func (p *ParticipantImpl) CanPublishSource(source livekit.TrackSource) bool { return p.grants.Load().Video.GetCanPublishSource(source) } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 329b6f1e8..acb15b81a 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1139,6 +1139,8 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS } } + iceConfig := participant.GetICEConfig() + hasICEFallback := iceConfig.GetPreferencePublisher() != livekit.ICECandidateType_ICT_NONE || iceConfig.GetPreferenceSubscriber() != livekit.ICECandidateType_ICT_NONE return &livekit.JoinResponse{ Room: r.ToProto(), Participant: participant.ToProto(), @@ -1155,6 +1157,7 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS ServerRegion: r.serverInfo.Region, SifTrailer: r.trailer, EnabledPublishCodecs: participant.GetEnabledPublishCodecs(), + FastPublish: participant.CanPublish() && !hasICEFallback, } } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 64569f196..cc06c8b55 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -433,6 +433,15 @@ func (t *TransportManager) SetICEConfig(iceConfig *livekit.ICEConfig) { } } +func (t *TransportManager) GetICEConfig() *livekit.ICEConfig { + t.lock.RLock() + defer t.lock.RUnlock() + if t.iceConfig == nil { + return nil + } + return proto.Clone(t.iceConfig).(*livekit.ICEConfig) +} + func (t *TransportManager) resetTransportConfigureLocked(reconfigured bool) { t.failureCount = 0 t.isTransportReconfigured = reconfigured diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 7077420d9..ad5294a80 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -342,6 +342,7 @@ type LocalParticipant interface { // permissions ClaimGrants() *auth.ClaimGrants SetPermission(permission *livekit.ParticipantPermission) bool + CanPublish() bool CanPublishSource(source livekit.TrackSource) bool CanSubscribe() bool CanPublishData() bool @@ -428,6 +429,7 @@ type LocalParticipant interface { GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.RTPTransceiver, sfu.DownTrackState) SetICEConfig(iceConfig *livekit.ICEConfig) + GetICEConfig() *livekit.ICEConfig OnICEConfigChanged(callback func(participant LocalParticipant, iceConfig *livekit.ICEConfig)) UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQualities []SubscribedCodecQuality) error diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index e4bf3fa47..3b067681f 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -69,6 +69,16 @@ type FakeLocalParticipant struct { arg2 *webrtc.RTPTransceiver arg3 sfu.DownTrackState } + CanPublishStub func() bool + canPublishMutex sync.RWMutex + canPublishArgsForCall []struct { + } + canPublishReturns struct { + result1 bool + } + canPublishReturnsOnCall map[int]struct { + result1 bool + } CanPublishDataStub func() bool canPublishDataMutex sync.RWMutex canPublishDataArgsForCall []struct { @@ -286,6 +296,16 @@ type FakeLocalParticipant struct { getEnabledPublishCodecsReturnsOnCall map[int]struct { result1 []*livekit.Codec } + GetICEConfigStub func() *livekit.ICEConfig + getICEConfigMutex sync.RWMutex + getICEConfigArgsForCall []struct { + } + getICEConfigReturns struct { + result1 *livekit.ICEConfig + } + getICEConfigReturnsOnCall map[int]struct { + result1 *livekit.ICEConfig + } GetICEConnectionDetailsStub func() []*types.ICEConnectionDetails getICEConnectionDetailsMutex sync.RWMutex getICEConnectionDetailsArgsForCall []struct { @@ -1308,6 +1328,59 @@ func (fake *FakeLocalParticipant) CacheDownTrackArgsForCall(i int) (livekit.Trac return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } +func (fake *FakeLocalParticipant) CanPublish() bool { + fake.canPublishMutex.Lock() + ret, specificReturn := fake.canPublishReturnsOnCall[len(fake.canPublishArgsForCall)] + fake.canPublishArgsForCall = append(fake.canPublishArgsForCall, struct { + }{}) + stub := fake.CanPublishStub + fakeReturns := fake.canPublishReturns + fake.recordInvocation("CanPublish", []interface{}{}) + fake.canPublishMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) CanPublishCallCount() int { + fake.canPublishMutex.RLock() + defer fake.canPublishMutex.RUnlock() + return len(fake.canPublishArgsForCall) +} + +func (fake *FakeLocalParticipant) CanPublishCalls(stub func() bool) { + fake.canPublishMutex.Lock() + defer fake.canPublishMutex.Unlock() + fake.CanPublishStub = stub +} + +func (fake *FakeLocalParticipant) CanPublishReturns(result1 bool) { + fake.canPublishMutex.Lock() + defer fake.canPublishMutex.Unlock() + fake.CanPublishStub = nil + fake.canPublishReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) CanPublishReturnsOnCall(i int, result1 bool) { + fake.canPublishMutex.Lock() + defer fake.canPublishMutex.Unlock() + fake.CanPublishStub = nil + if fake.canPublishReturnsOnCall == nil { + fake.canPublishReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.canPublishReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) CanPublishData() bool { fake.canPublishDataMutex.Lock() ret, specificReturn := fake.canPublishDataReturnsOnCall[len(fake.canPublishDataArgsForCall)] @@ -2442,6 +2515,59 @@ func (fake *FakeLocalParticipant) GetEnabledPublishCodecsReturnsOnCall(i int, re }{result1} } +func (fake *FakeLocalParticipant) GetICEConfig() *livekit.ICEConfig { + fake.getICEConfigMutex.Lock() + ret, specificReturn := fake.getICEConfigReturnsOnCall[len(fake.getICEConfigArgsForCall)] + fake.getICEConfigArgsForCall = append(fake.getICEConfigArgsForCall, struct { + }{}) + stub := fake.GetICEConfigStub + fakeReturns := fake.getICEConfigReturns + fake.recordInvocation("GetICEConfig", []interface{}{}) + fake.getICEConfigMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetICEConfigCallCount() int { + fake.getICEConfigMutex.RLock() + defer fake.getICEConfigMutex.RUnlock() + return len(fake.getICEConfigArgsForCall) +} + +func (fake *FakeLocalParticipant) GetICEConfigCalls(stub func() *livekit.ICEConfig) { + fake.getICEConfigMutex.Lock() + defer fake.getICEConfigMutex.Unlock() + fake.GetICEConfigStub = stub +} + +func (fake *FakeLocalParticipant) GetICEConfigReturns(result1 *livekit.ICEConfig) { + fake.getICEConfigMutex.Lock() + defer fake.getICEConfigMutex.Unlock() + fake.GetICEConfigStub = nil + fake.getICEConfigReturns = struct { + result1 *livekit.ICEConfig + }{result1} +} + +func (fake *FakeLocalParticipant) GetICEConfigReturnsOnCall(i int, result1 *livekit.ICEConfig) { + fake.getICEConfigMutex.Lock() + defer fake.getICEConfigMutex.Unlock() + fake.GetICEConfigStub = nil + if fake.getICEConfigReturnsOnCall == nil { + fake.getICEConfigReturnsOnCall = make(map[int]struct { + result1 *livekit.ICEConfig + }) + } + fake.getICEConfigReturnsOnCall[i] = struct { + result1 *livekit.ICEConfig + }{result1} +} + func (fake *FakeLocalParticipant) GetICEConnectionDetails() []*types.ICEConnectionDetails { fake.getICEConnectionDetailsMutex.Lock() ret, specificReturn := fake.getICEConnectionDetailsReturnsOnCall[len(fake.getICEConnectionDetailsArgsForCall)] @@ -6797,6 +6923,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.addTransceiverFromTrackToSubscriberMutex.RUnlock() fake.cacheDownTrackMutex.RLock() defer fake.cacheDownTrackMutex.RUnlock() + fake.canPublishMutex.RLock() + defer fake.canPublishMutex.RUnlock() fake.canPublishDataMutex.RLock() defer fake.canPublishDataMutex.RUnlock() fake.canPublishSourceMutex.RLock() @@ -6839,6 +6967,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getDisableSenderReportPassThroughMutex.RUnlock() fake.getEnabledPublishCodecsMutex.RLock() defer fake.getEnabledPublishCodecsMutex.RUnlock() + fake.getICEConfigMutex.RLock() + defer fake.getICEConfigMutex.RUnlock() fake.getICEConnectionDetailsMutex.RLock() defer fake.getICEConnectionDetailsMutex.RUnlock() fake.getLoggerMutex.RLock()