Use publisher PC as default with publish-only connections (#198)

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
 }

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

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()
 	}

pkg/rtc/types/interfaces.go

@@ -67,6 +67,7 @@ type Participant interface {
 	CanSubscribe() bool
 	CanPublishData() bool
 	Hidden() bool
+	SubscriberAsPrimary() bool
 
 	Start()
 	Close() error

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()