diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 2fa92af00..9e8a1997e 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -110,16 +110,20 @@ func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit. } // SendSpeakerUpdate notifies participant changes to speakers. only send members that have changed since last update -func (p *ParticipantImpl) SendSpeakerUpdate(speakers []*livekit.SpeakerInfo) error { +func (p *ParticipantImpl) SendSpeakerUpdate(speakers []*livekit.SpeakerInfo, force bool) error { if !p.IsReady() { return nil } var scopedSpeakers []*livekit.SpeakerInfo - for _, s := range speakers { - participantID := livekit.ParticipantID(s.Sid) - if p.IsSubscribedTo(participantID) || participantID == p.ID() { - scopedSpeakers = append(scopedSpeakers, s) + if force { + scopedSpeakers = speakers + } else { + for _, s := range speakers { + participantID := livekit.ParticipantID(s.Sid) + if p.IsSubscribedTo(participantID) || participantID == p.ID() { + scopedSpeakers = append(scopedSpeakers, s) + } } } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index a3f009e60..1e70a0893 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -305,7 +305,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions Level: float32(level), Active: active, }, - }) + }, false) } update := &livekit.ConnectionQualityUpdate{} @@ -320,7 +320,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions Level: 0, Active: false, }, - }) + }, true) } }) @@ -911,7 +911,7 @@ func (r *Room) sendActiveSpeakers(speakers []*livekit.SpeakerInfo) { func (r *Room) sendSpeakerChanges(speakers []*livekit.SpeakerInfo) { for _, p := range r.GetParticipants() { if p.ProtocolVersion().SupportsSpeakerChanged() { - _ = p.SendSpeakerUpdate(speakers) + _ = p.SendSpeakerUpdate(speakers, false) } } } diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index 488d20239..392508a55 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -411,7 +411,7 @@ func TestActiveSpeakers(t *testing.T) { var updates [][]*livekit.SpeakerInfo numCalls := p.SendSpeakerUpdateCallCount() for i := 0; i < numCalls; i++ { - infos := p.SendSpeakerUpdateArgsForCall(i) + infos, _ := p.SendSpeakerUpdateArgsForCall(i) updates = append(updates, infos) } return updates diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 4eef964d7..43ca2d259 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -288,7 +288,7 @@ type LocalParticipant interface { // server sent messages SendJoinResponse(joinResponse *livekit.JoinResponse) error SendParticipantUpdate(participants []*livekit.ParticipantInfo) error - SendSpeakerUpdate(speakers []*livekit.SpeakerInfo) error + SendSpeakerUpdate(speakers []*livekit.SpeakerInfo, force bool) error SendDataPacket(packet *livekit.DataPacket, data []byte) error SendRoomUpdate(room *livekit.Room) error SendConnectionQualityUpdate(update *livekit.ConnectionQualityUpdate) error diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 3baa5c697..1d2f4bb19 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -597,10 +597,11 @@ type FakeLocalParticipant struct { sendRoomUpdateReturnsOnCall map[int]struct { result1 error } - SendSpeakerUpdateStub func([]*livekit.SpeakerInfo) error + SendSpeakerUpdateStub func([]*livekit.SpeakerInfo, bool) error sendSpeakerUpdateMutex sync.RWMutex sendSpeakerUpdateArgsForCall []struct { arg1 []*livekit.SpeakerInfo + arg2 bool } sendSpeakerUpdateReturns struct { result1 error @@ -3963,7 +3964,7 @@ func (fake *FakeLocalParticipant) SendRoomUpdateReturnsOnCall(i int, result1 err }{result1} } -func (fake *FakeLocalParticipant) SendSpeakerUpdate(arg1 []*livekit.SpeakerInfo) error { +func (fake *FakeLocalParticipant) SendSpeakerUpdate(arg1 []*livekit.SpeakerInfo, arg2 bool) error { var arg1Copy []*livekit.SpeakerInfo if arg1 != nil { arg1Copy = make([]*livekit.SpeakerInfo, len(arg1)) @@ -3973,13 +3974,14 @@ func (fake *FakeLocalParticipant) SendSpeakerUpdate(arg1 []*livekit.SpeakerInfo) ret, specificReturn := fake.sendSpeakerUpdateReturnsOnCall[len(fake.sendSpeakerUpdateArgsForCall)] fake.sendSpeakerUpdateArgsForCall = append(fake.sendSpeakerUpdateArgsForCall, struct { arg1 []*livekit.SpeakerInfo - }{arg1Copy}) + arg2 bool + }{arg1Copy, arg2}) stub := fake.SendSpeakerUpdateStub fakeReturns := fake.sendSpeakerUpdateReturns - fake.recordInvocation("SendSpeakerUpdate", []interface{}{arg1Copy}) + fake.recordInvocation("SendSpeakerUpdate", []interface{}{arg1Copy, arg2}) fake.sendSpeakerUpdateMutex.Unlock() if stub != nil { - return stub(arg1) + return stub(arg1, arg2) } if specificReturn { return ret.result1 @@ -3993,17 +3995,17 @@ func (fake *FakeLocalParticipant) SendSpeakerUpdateCallCount() int { return len(fake.sendSpeakerUpdateArgsForCall) } -func (fake *FakeLocalParticipant) SendSpeakerUpdateCalls(stub func([]*livekit.SpeakerInfo) error) { +func (fake *FakeLocalParticipant) SendSpeakerUpdateCalls(stub func([]*livekit.SpeakerInfo, bool) error) { fake.sendSpeakerUpdateMutex.Lock() defer fake.sendSpeakerUpdateMutex.Unlock() fake.SendSpeakerUpdateStub = stub } -func (fake *FakeLocalParticipant) SendSpeakerUpdateArgsForCall(i int) []*livekit.SpeakerInfo { +func (fake *FakeLocalParticipant) SendSpeakerUpdateArgsForCall(i int) ([]*livekit.SpeakerInfo, bool) { fake.sendSpeakerUpdateMutex.RLock() defer fake.sendSpeakerUpdateMutex.RUnlock() argsForCall := fake.sendSpeakerUpdateArgsForCall[i] - return argsForCall.arg1 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeLocalParticipant) SendSpeakerUpdateReturns(result1 error) {