Fix unsubscribed speakers stuck as speaking to clients (#1475)

When we unsubscribe from a speaker, SendSpeakerUpdates will drop updates
from that speaker. This has the side effect of dropping the "clearing"
message that we are sending as well.

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

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

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

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

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