From 2a6a9b8a4a0842c84c734aea2d30ec5741a584f3 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 18 Jul 2025 15:24:52 +0530 Subject: [PATCH] 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 | 61 +-------- pkg/rtc/participant_signal.go | 72 +++++++++++ pkg/rtc/subscriptionmanager.go | 2 +- pkg/rtc/types/interfaces.go | 2 +- .../typesfakes/fake_local_participant.go | 121 +++++++++++------- 5 files changed, 157 insertions(+), 101 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 10e9baddc..bff233137 100644 --- a/pkg/rtc/participant.go +++ b/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) diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 75733e633..9cb0d70e8 100644 --- a/pkg/rtc/participant_signal.go +++ b/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 +} diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 84cc69524..0c89c0882 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/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 diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 0611a7b1f..ffdc60675 100644 --- a/pkg/rtc/types/interfaces.go +++ b/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 diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 4c9eb9a77..6a309360d 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/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()