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