Grouping all signal messages into participant_signal. (#3801)

Currently, it is a bit of a mish-mash
- some compose the message fully and just call send()
- some give parameters and the message is composed in
  participant_signal.go

Was thinking about making an interface for signalling and have v1/v2
impls, but did not want to repeat composing messages if there are common
messages. And some of those function reach into `ParicipantImpl` object
and use information (simple example of p.IsReady()) which would become
more elaborate if the signaller is split out into its own struct.

Maybe, just need to make an interface for the sink and send to the
correct sink based on v1 /v2 signal transport.

But, for now, just grouping all signal messaages in one file
so that it is easier to manage later.

pkg/rtc/participant.go

@@ -1086,11 +1086,7 @@ func (p *ParticipantImpl) onPublisherAnswer(answer webrtc.SessionDescription, an
 		"answer", answer,
 		"answerId", answerId,
 	)
-	return p.writeMessage(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_Answer{
-			Answer: ToProtoSessionDescription(answer, answerId),
-		},
-	})
+	return p.sendSdpAnswer(answer, answerId)
 }
 
 func (p *ParticipantImpl) GetAnswer() (webrtc.SessionDescription, error) {
@@ -1638,22 +1634,6 @@ func (p *ParticipantImpl) onTrackUnsubscribed(subTrack types.SubscribedTrack) {
 	p.TransportManager.RemoveSubscribedTrack(subTrack)
 }
 
-func (p *ParticipantImpl) SubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool) {
-	p.subLogger.Debugw("sending subscription permission update", "publisherID", publisherID, "trackID", trackID, "allowed", allowed)
-	err := p.writeMessage(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_SubscriptionPermissionUpdate{
-			SubscriptionPermissionUpdate: &livekit.SubscriptionPermissionUpdate{
-				ParticipantSid: string(publisherID),
-				TrackSid:       string(trackID),
-				Allowed:        allowed,
-			},
-		},
-	})
-	if err != nil {
-		p.subLogger.Errorw("could not send subscription permission update", err)
-	}
-}
-
 func (p *ParticipantImpl) UpdateMediaRTT(rtt uint32) {
 	now := time.Now()
 	p.lock.Lock()
@@ -1993,11 +1973,7 @@ func (p *ParticipantImpl) onSubscriberOffer(offer webrtc.SessionDescription, off
 		"offer", offer,
 		"offerId", offerId,
 	)
-	return p.writeMessage(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_Offer{
-			Offer: ToProtoSessionDescription(offer, offerId),
-		},
-	})
+	return p.sendSdpOffer(offer, offerId)
 }
 
 func (p *ParticipantImpl) removePublishedTrack(track types.MediaTrack) {
@@ -2489,11 +2465,7 @@ func (p *ParticipantImpl) onStreamStateChange(update *streamallocator.StreamStat
 		})
 	}
 
-	return p.writeMessage(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_StreamStateUpdate{
-			StreamStateUpdate: streamStateUpdate,
-		},
-	})
+	return p.sendStreamStateUpdate(streamStateUpdate)
 }
 
 func (p *ParticipantImpl) onSubscribedMaxQualityChange(
@@ -2549,11 +2521,7 @@ func (p *ParticipantImpl) onSubscribedMaxQualityChange(
 		"qualities", subscribedQualities,
 		"max", maxSubscribedQualities,
 	)
-	return p.writeMessage(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_SubscribedQualityUpdate{
-			SubscribedQualityUpdate: subscribedQualityUpdate,
-		},
-	})
+	return p.sendSubscribedQualityUpdate(subscribedQualityUpdate)
 }
 
 func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *livekit.TrackInfo {
@@ -2704,18 +2672,6 @@ func (p *ParticipantImpl) HasConnected() bool {
 	return p.TransportManager.HasSubscriberEverConnected() || p.TransportManager.HasPublisherEverConnected()
 }
 
-func (p *ParticipantImpl) sendTrackPublished(cid string, ti *livekit.TrackInfo) {
-	p.pubLogger.Debugw("sending track published", "cid", cid, "trackInfo", logger.Proto(ti))
-	_ = p.writeMessage(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_TrackPublished{
-			TrackPublished: &livekit.TrackPublishedResponse{
-				Cid:   cid,
-				Track: ti,
-			},
-		},
-	})
-}
-
 func (p *ParticipantImpl) SetTrackMuted(trackID livekit.TrackID, muted bool, fromAdmin bool) *livekit.TrackInfo {
 	// when request is coming from admin, send message to current participant
 	if fromAdmin {
@@ -3344,14 +3300,7 @@ func (p *ParticipantImpl) onSubscriptionError(trackID livekit.TrackID, fatal boo
 		signalErr = livekit.SubscriptionError_SE_TRACK_NOTFOUND
 	}
 
-	_ = p.writeMessage(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_SubscriptionResponse{
-			SubscriptionResponse: &livekit.SubscriptionResponse{
-				TrackSid: string(trackID),
-				Err:      signalErr,
-			},
-		},
-	})
+	p.sendSubscriptionResponse(trackID, signalErr)
 
 	if p.params.ReconnectOnSubscriptionError && fatal {
 		p.subLogger.Infow("issuing full reconnect on subscription error", "trackID", trackID)

pkg/rtc/participant_signal.go

@@ -407,3 +407,75 @@ func (p *ParticipantImpl) sendLeaveRequest(
 
 	return nil
 }
+
+func (p *ParticipantImpl) sendSdpAnswer(answer webrtc.SessionDescription, answerId uint32) error {
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_Answer{
+			Answer: ToProtoSessionDescription(answer, answerId),
+		},
+	})
+}
+
+func (p *ParticipantImpl) sendSdpOffer(offer webrtc.SessionDescription, offerId uint32) error {
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_Offer{
+			Offer: ToProtoSessionDescription(offer, offerId),
+		},
+	})
+}
+
+func (p *ParticipantImpl) sendStreamStateUpdate(streamStateUpdate *livekit.StreamStateUpdate) error {
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_StreamStateUpdate{
+			StreamStateUpdate: streamStateUpdate,
+		},
+	})
+}
+
+func (p *ParticipantImpl) sendSubscribedQualityUpdate(subscribedQualityUpdate *livekit.SubscribedQualityUpdate) error {
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_SubscribedQualityUpdate{
+			SubscribedQualityUpdate: subscribedQualityUpdate,
+		},
+	})
+}
+
+func (p *ParticipantImpl) sendTrackPublished(cid string, ti *livekit.TrackInfo) error {
+	p.pubLogger.Debugw("sending track published", "cid", cid, "trackInfo", logger.Proto(ti))
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_TrackPublished{
+			TrackPublished: &livekit.TrackPublishedResponse{
+				Cid:   cid,
+				Track: ti,
+			},
+		},
+	})
+}
+
+func (p *ParticipantImpl) sendSubscriptionResponse(trackID livekit.TrackID, subErr livekit.SubscriptionError) error {
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_SubscriptionResponse{
+			SubscriptionResponse: &livekit.SubscriptionResponse{
+				TrackSid: string(trackID),
+				Err:      subErr,
+			},
+		},
+	})
+}
+
+func (p *ParticipantImpl) SendSubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool) error {
+	p.subLogger.Debugw("sending subscription permission update", "publisherID", publisherID, "trackID", trackID, "allowed", allowed)
+	err := p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_SubscriptionPermissionUpdate{
+			SubscriptionPermissionUpdate: &livekit.SubscriptionPermissionUpdate{
+				ParticipantSid: string(publisherID),
+				TrackSid:       string(trackID),
+				Allowed:        allowed,
+			},
+		},
+	})
+	if err != nil {
+		p.subLogger.Errorw("could not send subscription permission update", err)
+	}
+	return err
+}

pkg/rtc/subscriptionmanager.go

@@ -556,7 +556,7 @@ func (m *SubscriptionManager) subscribe(s *trackSubscription) error {
 
 	permChanged := s.setHasPermission(res.HasPermission)
 	if permChanged {
-		m.params.Participant.SubscriptionPermissionUpdate(s.getPublisherID(), trackID, res.HasPermission)
+		m.params.Participant.SendSubscriptionPermissionUpdate(s.getPublisherID(), trackID, res.HasPermission)
 	}
 	if !res.HasPermission {
 		return ErrNoTrackPermission

pkg/rtc/types/interfaces.go

@@ -444,7 +444,7 @@ type LocalParticipant interface {
 	SendDataMessageUnlabeled(data []byte, useRaw bool, sender livekit.ParticipantIdentity) error
 	SendRoomUpdate(room *livekit.Room) error
 	SendConnectionQualityUpdate(update *livekit.ConnectionQualityUpdate) error
-	SubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool)
+	SendSubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool) error
 	SendRefreshToken(token string) error
 	SendRequestResponse(requestResponse *livekit.RequestResponse) error
 	HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -996,6 +996,19 @@ type FakeLocalParticipant struct {
 	sendSpeakerUpdateReturnsOnCall map[int]struct {
 		result1 error
 	}
+	SendSubscriptionPermissionUpdateStub        func(livekit.ParticipantID, livekit.TrackID, bool) error
+	sendSubscriptionPermissionUpdateMutex       sync.RWMutex
+	sendSubscriptionPermissionUpdateArgsForCall []struct {
+		arg1 livekit.ParticipantID
+		arg2 livekit.TrackID
+		arg3 bool
+	}
+	sendSubscriptionPermissionUpdateReturns struct {
+		result1 error
+	}
+	sendSubscriptionPermissionUpdateReturnsOnCall map[int]struct {
+		result1 error
+	}
 	SetAttributesStub        func(map[string]string)
 	setAttributesMutex       sync.RWMutex
 	setAttributesArgsForCall []struct {
@@ -1122,13 +1135,6 @@ type FakeLocalParticipant struct {
 		result1 *livekit.SubscriptionPermission
 		result2 utils.TimedVersion
 	}
-	SubscriptionPermissionUpdateStub        func(livekit.ParticipantID, livekit.TrackID, bool)
-	subscriptionPermissionUpdateMutex       sync.RWMutex
-	subscriptionPermissionUpdateArgsForCall []struct {
-		arg1 livekit.ParticipantID
-		arg2 livekit.TrackID
-		arg3 bool
-	}
 	SupportsCodecChangeStub        func() bool
 	supportsCodecChangeMutex       sync.RWMutex
 	supportsCodecChangeArgsForCall []struct {
@@ -6569,6 +6575,69 @@ func (fake *FakeLocalParticipant) SendSpeakerUpdateReturnsOnCall(i int, result1
 	}{result1}
 }
 
+func (fake *FakeLocalParticipant) SendSubscriptionPermissionUpdate(arg1 livekit.ParticipantID, arg2 livekit.TrackID, arg3 bool) error {
+	fake.sendSubscriptionPermissionUpdateMutex.Lock()
+	ret, specificReturn := fake.sendSubscriptionPermissionUpdateReturnsOnCall[len(fake.sendSubscriptionPermissionUpdateArgsForCall)]
+	fake.sendSubscriptionPermissionUpdateArgsForCall = append(fake.sendSubscriptionPermissionUpdateArgsForCall, struct {
+		arg1 livekit.ParticipantID
+		arg2 livekit.TrackID
+		arg3 bool
+	}{arg1, arg2, arg3})
+	stub := fake.SendSubscriptionPermissionUpdateStub
+	fakeReturns := fake.sendSubscriptionPermissionUpdateReturns
+	fake.recordInvocation("SendSubscriptionPermissionUpdate", []interface{}{arg1, arg2, arg3})
+	fake.sendSubscriptionPermissionUpdateMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2, arg3)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeLocalParticipant) SendSubscriptionPermissionUpdateCallCount() int {
+	fake.sendSubscriptionPermissionUpdateMutex.RLock()
+	defer fake.sendSubscriptionPermissionUpdateMutex.RUnlock()
+	return len(fake.sendSubscriptionPermissionUpdateArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) SendSubscriptionPermissionUpdateCalls(stub func(livekit.ParticipantID, livekit.TrackID, bool) error) {
+	fake.sendSubscriptionPermissionUpdateMutex.Lock()
+	defer fake.sendSubscriptionPermissionUpdateMutex.Unlock()
+	fake.SendSubscriptionPermissionUpdateStub = stub
+}
+
+func (fake *FakeLocalParticipant) SendSubscriptionPermissionUpdateArgsForCall(i int) (livekit.ParticipantID, livekit.TrackID, bool) {
+	fake.sendSubscriptionPermissionUpdateMutex.RLock()
+	defer fake.sendSubscriptionPermissionUpdateMutex.RUnlock()
+	argsForCall := fake.sendSubscriptionPermissionUpdateArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
+}
+
+func (fake *FakeLocalParticipant) SendSubscriptionPermissionUpdateReturns(result1 error) {
+	fake.sendSubscriptionPermissionUpdateMutex.Lock()
+	defer fake.sendSubscriptionPermissionUpdateMutex.Unlock()
+	fake.SendSubscriptionPermissionUpdateStub = nil
+	fake.sendSubscriptionPermissionUpdateReturns = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeLocalParticipant) SendSubscriptionPermissionUpdateReturnsOnCall(i int, result1 error) {
+	fake.sendSubscriptionPermissionUpdateMutex.Lock()
+	defer fake.sendSubscriptionPermissionUpdateMutex.Unlock()
+	fake.SendSubscriptionPermissionUpdateStub = nil
+	if fake.sendSubscriptionPermissionUpdateReturnsOnCall == nil {
+		fake.sendSubscriptionPermissionUpdateReturnsOnCall = make(map[int]struct {
+			result1 error
+		})
+	}
+	fake.sendSubscriptionPermissionUpdateReturnsOnCall[i] = struct {
+		result1 error
+	}{result1}
+}
+
 func (fake *FakeLocalParticipant) SetAttributes(arg1 map[string]string) {
 	fake.setAttributesMutex.Lock()
 	fake.setAttributesArgsForCall = append(fake.setAttributesArgsForCall, struct {
@@ -7280,40 +7349,6 @@ func (fake *FakeLocalParticipant) SubscriptionPermissionReturnsOnCall(i int, res
 	}{result1, result2}
 }
 
-func (fake *FakeLocalParticipant) SubscriptionPermissionUpdate(arg1 livekit.ParticipantID, arg2 livekit.TrackID, arg3 bool) {
-	fake.subscriptionPermissionUpdateMutex.Lock()
-	fake.subscriptionPermissionUpdateArgsForCall = append(fake.subscriptionPermissionUpdateArgsForCall, struct {
-		arg1 livekit.ParticipantID
-		arg2 livekit.TrackID
-		arg3 bool
-	}{arg1, arg2, arg3})
-	stub := fake.SubscriptionPermissionUpdateStub
-	fake.recordInvocation("SubscriptionPermissionUpdate", []interface{}{arg1, arg2, arg3})
-	fake.subscriptionPermissionUpdateMutex.Unlock()
-	if stub != nil {
-		fake.SubscriptionPermissionUpdateStub(arg1, arg2, arg3)
-	}
-}
-
-func (fake *FakeLocalParticipant) SubscriptionPermissionUpdateCallCount() int {
-	fake.subscriptionPermissionUpdateMutex.RLock()
-	defer fake.subscriptionPermissionUpdateMutex.RUnlock()
-	return len(fake.subscriptionPermissionUpdateArgsForCall)
-}
-
-func (fake *FakeLocalParticipant) SubscriptionPermissionUpdateCalls(stub func(livekit.ParticipantID, livekit.TrackID, bool)) {
-	fake.subscriptionPermissionUpdateMutex.Lock()
-	defer fake.subscriptionPermissionUpdateMutex.Unlock()
-	fake.SubscriptionPermissionUpdateStub = stub
-}
-
-func (fake *FakeLocalParticipant) SubscriptionPermissionUpdateArgsForCall(i int) (livekit.ParticipantID, livekit.TrackID, bool) {
-	fake.subscriptionPermissionUpdateMutex.RLock()
-	defer fake.subscriptionPermissionUpdateMutex.RUnlock()
-	argsForCall := fake.subscriptionPermissionUpdateArgsForCall[i]
-	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
-}
-
 func (fake *FakeLocalParticipant) SupportsCodecChange() bool {
 	fake.supportsCodecChangeMutex.Lock()
 	ret, specificReturn := fake.supportsCodecChangeReturnsOnCall[len(fake.supportsCodecChangeArgsForCall)]
@@ -8613,6 +8648,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.sendRoomUpdateMutex.RUnlock()
 	fake.sendSpeakerUpdateMutex.RLock()
 	defer fake.sendSpeakerUpdateMutex.RUnlock()
+	fake.sendSubscriptionPermissionUpdateMutex.RLock()
+	defer fake.sendSubscriptionPermissionUpdateMutex.RUnlock()
 	fake.setAttributesMutex.RLock()
 	defer fake.setAttributesMutex.RUnlock()
 	fake.setICEConfigMutex.RLock()
@@ -8647,8 +8684,6 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.subscriberAsPrimaryMutex.RUnlock()
 	fake.subscriptionPermissionMutex.RLock()
 	defer fake.subscriptionPermissionMutex.RUnlock()
-	fake.subscriptionPermissionUpdateMutex.RLock()
-	defer fake.subscriptionPermissionUpdateMutex.RUnlock()
 	fake.supportsCodecChangeMutex.RLock()
 	defer fake.supportsCodecChangeMutex.RUnlock()
 	fake.supportsMovingMutex.RLock()