From 6dcb29c5591d5eab49194ee3c36429a48b32b211 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sat, 20 Nov 2021 22:28:36 -0800 Subject: [PATCH] Use publisher PC as default with publish-only connections (#198) --- pkg/rtc/participant.go | 18 ++++-- pkg/rtc/participant_internal_test.go | 32 +++++++++- pkg/rtc/room.go | 2 +- pkg/rtc/types/interfaces.go | 1 + pkg/rtc/types/typesfakes/fake_participant.go | 65 ++++++++++++++++++++ 5 files changed, 111 insertions(+), 7 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 2ee957812..7931b8e24 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -155,7 +155,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { primaryPC := p.publisher.pc - if p.ProtocolVersion().SubscriberAsPrimary() { + if p.SubscriberAsPrimary() { primaryPC = p.subscriber.pc ordered := true // also create data channels for subs @@ -505,7 +505,11 @@ func (p *ParticipantImpl) RemoveSubscriber(participantId string) { // signal connection methods -func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticipants []*livekit.ParticipantInfo, iceServers []*livekit.ICEServer) error { +func (p *ParticipantImpl) SendJoinResponse( + roomInfo *livekit.Room, + otherParticipants []*livekit.ParticipantInfo, + iceServers []*livekit.ICEServer, +) error { // send Join response return p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Join{ @@ -516,7 +520,7 @@ func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticip ServerVersion: version.Version, IceServers: iceServers, // indicates both server and client support subscriber as primary - SubscriberPrimary: p.ProtocolVersion().SubscriberAsPrimary(), + SubscriberPrimary: p.SubscriberAsPrimary(), }, }, }) @@ -574,13 +578,13 @@ func (p *ParticipantImpl) SendDataPacket(dp *livekit.DataPacket) error { var dc *webrtc.DataChannel if dp.Kind == livekit.DataPacket_RELIABLE { - if p.ProtocolVersion().SubscriberAsPrimary() { + if p.SubscriberAsPrimary() { dc = p.reliableDCSub } else { dc = p.reliableDC } } else { - if p.ProtocolVersion().SubscriberAsPrimary() { + if p.SubscriberAsPrimary() { dc = p.lossyDCSub } else { dc = p.lossyDC @@ -753,6 +757,10 @@ func (p *ParticipantImpl) Hidden() bool { return p.params.Hidden } +func (p *ParticipantImpl) SubscriberAsPrimary() bool { + return p.ProtocolVersion().SubscriberAsPrimary() && p.CanSubscribe() +} + func (p *ParticipantImpl) SubscriberPC() *webrtc.PeerConnection { return p.subscriber.pc } diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 765bc394c..e64c10bd7 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -272,6 +272,36 @@ func TestConnectionQuality(t *testing.T) { }) } +func TestSubscriberAsPrimary(t *testing.T) { + t.Run("protocol 4 uses subs as primary", func(t *testing.T) { + p := newParticipantForTest("test") + p.SetPermission(&livekit.ParticipantPermission{ + CanSubscribe: true, + CanPublish: true, + }) + require.True(t, p.SubscriberAsPrimary()) + }) + + t.Run("protocol 2 uses pub as primary", func(t *testing.T) { + p := newParticipantForTest("test") + p.params.ProtocolVersion = 2 + p.SetPermission(&livekit.ParticipantPermission{ + CanSubscribe: true, + CanPublish: true, + }) + require.False(t, p.SubscriberAsPrimary()) + }) + + t.Run("publisher only uses pub as primary", func(t *testing.T) { + p := newParticipantForTest("test") + p.SetPermission(&livekit.ParticipantPermission{ + CanSubscribe: false, + CanPublish: true, + }) + require.False(t, p.SubscriberAsPrimary()) + }) +} + func newParticipantForTest(identity string) *ParticipantImpl { conf, _ := config.NewConfig("", nil) // disable mux, it doesn't play too well with unit test @@ -285,7 +315,7 @@ func newParticipantForTest(identity string) *ParticipantImpl { Identity: identity, Config: rtcConf, Sink: &routingfakes.FakeMessageSink{}, - ProtocolVersion: 0, + ProtocolVersion: 4, ThrottleConfig: conf.RTC.PLIThrottle, }) return p diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index fddf804a8..6aa5b9443 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -222,7 +222,7 @@ func (r *Room) Join(participant types.Participant, opts *ParticipantOptions, ice return err } - if participant.ProtocolVersion().SubscriberAsPrimary() { + if participant.SubscriberAsPrimary() { // initiates sub connection as primary participant.Negotiate() } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index e04fb3499..53c847696 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -67,6 +67,7 @@ type Participant interface { CanSubscribe() bool CanPublishData() bool Hidden() bool + SubscriberAsPrimary() bool Start() Close() error diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 6a3803a55..43c4f0b4a 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -446,6 +446,16 @@ type FakeParticipant struct { stateReturnsOnCall map[int]struct { result1 livekit.ParticipantInfo_State } + SubscriberAsPrimaryStub func() bool + subscriberAsPrimaryMutex sync.RWMutex + subscriberAsPrimaryArgsForCall []struct { + } + subscriberAsPrimaryReturns struct { + result1 bool + } + subscriberAsPrimaryReturnsOnCall map[int]struct { + result1 bool + } SubscriberMediaEngineStub func() *webrtc.MediaEngine subscriberMediaEngineMutex sync.RWMutex subscriberMediaEngineArgsForCall []struct { @@ -2864,6 +2874,59 @@ func (fake *FakeParticipant) StateReturnsOnCall(i int, result1 livekit.Participa }{result1} } +func (fake *FakeParticipant) SubscriberAsPrimary() bool { + fake.subscriberAsPrimaryMutex.Lock() + ret, specificReturn := fake.subscriberAsPrimaryReturnsOnCall[len(fake.subscriberAsPrimaryArgsForCall)] + fake.subscriberAsPrimaryArgsForCall = append(fake.subscriberAsPrimaryArgsForCall, struct { + }{}) + stub := fake.SubscriberAsPrimaryStub + fakeReturns := fake.subscriberAsPrimaryReturns + fake.recordInvocation("SubscriberAsPrimary", []interface{}{}) + fake.subscriberAsPrimaryMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) SubscriberAsPrimaryCallCount() int { + fake.subscriberAsPrimaryMutex.RLock() + defer fake.subscriberAsPrimaryMutex.RUnlock() + return len(fake.subscriberAsPrimaryArgsForCall) +} + +func (fake *FakeParticipant) SubscriberAsPrimaryCalls(stub func() bool) { + fake.subscriberAsPrimaryMutex.Lock() + defer fake.subscriberAsPrimaryMutex.Unlock() + fake.SubscriberAsPrimaryStub = stub +} + +func (fake *FakeParticipant) SubscriberAsPrimaryReturns(result1 bool) { + fake.subscriberAsPrimaryMutex.Lock() + defer fake.subscriberAsPrimaryMutex.Unlock() + fake.SubscriberAsPrimaryStub = nil + fake.subscriberAsPrimaryReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeParticipant) SubscriberAsPrimaryReturnsOnCall(i int, result1 bool) { + fake.subscriberAsPrimaryMutex.Lock() + defer fake.subscriberAsPrimaryMutex.Unlock() + fake.SubscriberAsPrimaryStub = nil + if fake.subscriberAsPrimaryReturnsOnCall == nil { + fake.subscriberAsPrimaryReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.subscriberAsPrimaryReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeParticipant) SubscriberMediaEngine() *webrtc.MediaEngine { fake.subscriberMediaEngineMutex.Lock() ret, specificReturn := fake.subscriberMediaEngineReturnsOnCall[len(fake.subscriberMediaEngineArgsForCall)] @@ -3124,6 +3187,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.startMutex.RUnlock() fake.stateMutex.RLock() defer fake.stateMutex.RUnlock() + fake.subscriberAsPrimaryMutex.RLock() + defer fake.subscriberAsPrimaryMutex.RUnlock() fake.subscriberMediaEngineMutex.RLock() defer fake.subscriberMediaEngineMutex.RUnlock() fake.subscriberPCMutex.RLock()