diff --git a/go.mod b/go.mod index ec383e2e8..e64a54ef9 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26 - github.com/livekit/protocol v1.5.5-0.20230422131440-eb3ef0f4bf36 + github.com/livekit/protocol v1.5.5 github.com/livekit/psrpc v0.3.0 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.14.0 diff --git a/go.sum b/go.sum index fa24f3bd0..dd6b0e056 100644 --- a/go.sum +++ b/go.sum @@ -121,8 +121,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26 h1:QlQFyMwCDgjyySsrgmrMcVbEBA6KZcyTzvK+z346tUA= github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26/go.mod h1:eDA41kiySZoG+wy4Etsjb3w0jjLx69i/vAmSjG4bteA= -github.com/livekit/protocol v1.5.5-0.20230422131440-eb3ef0f4bf36 h1:Qrl0N7dAeR2iLOk6u4WNelFhTh2OetyZzd/h8kbxLSI= -github.com/livekit/protocol v1.5.5-0.20230422131440-eb3ef0f4bf36/go.mod h1:iZ289+6H5xn/9kP2iqpRvVWxuc8GXBMqN0qI7LdN9HI= +github.com/livekit/protocol v1.5.5 h1:vuSU3TI/w58WnAWnyC59nMzY/JE+ZznU6W/iRgWw4JQ= +github.com/livekit/protocol v1.5.5/go.mod h1:iZ289+6H5xn/9kP2iqpRvVWxuc8GXBMqN0qI7LdN9HI= github.com/livekit/psrpc v0.3.0 h1:giBZsfM3CWA0oIYXofsMITbVQtyW7u/ES9sQmVspHPM= github.com/livekit/psrpc v0.3.0/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index e9d8b65d8..5a58ebe3c 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -37,6 +37,7 @@ const ( var ( // var to allow unit test override RoomDepartureGrace uint32 = 20 + roomUpdateInterval = 5 * time.Second // frequency to update room participant counts ) type broadcastOptions struct { @@ -47,9 +48,10 @@ type broadcastOptions struct { type Room struct { lock sync.RWMutex - protoRoom *livekit.Room - internal *livekit.RoomInternal - Logger logger.Logger + protoRoom *livekit.Room + internal *livekit.RoomInternal + protoProxy *utils.ProtoProxy[*livekit.Room] + Logger logger.Logger config WebRTCConfig audioConfig *config.AudioConfig @@ -76,7 +78,7 @@ type Room struct { closed chan struct{} onParticipantChanged func(p types.LocalParticipant) - onMetadataUpdate func(metadata string) + onRoomUpdated func() onClose func() } @@ -110,6 +112,7 @@ func NewRoom( batchedUpdates: make(map[livekit.ParticipantIdentity]*livekit.ParticipantInfo), closed: make(chan struct{}), } + r.protoProxy = utils.NewProtoProxy[*livekit.Room](roomUpdateInterval, r.updateProto) if r.protoRoom.EmptyTimeout == 0 { r.protoRoom.EmptyTimeout = DefaultEmptyTimeout } @@ -119,16 +122,13 @@ func NewRoom( go r.audioUpdateWorker() go r.connectionQualityWorker() - go r.subscriberBroadcastWorker() + go r.changeUpdateWorker() return r } func (r *Room) ToProto() *livekit.Room { - r.lock.RLock() - defer r.lock.RUnlock() - - return proto.Clone(r.protoRoom).(*livekit.Room) + return r.protoProxy.Get() } func (r *Room) Name() livekit.RoomName { @@ -244,14 +244,13 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me } if r.protoRoom.MaxParticipants > 0 && !participant.IsRecorder() { - participantCount := 0 + numParticipants := uint32(0) for _, p := range r.participants { if !p.IsRecorder() { - participantCount++ + numParticipants++ } } - - if participantCount >= int(r.protoRoom.MaxParticipants) { + if numParticipants >= r.protoRoom.MaxParticipants { return ErrMaxParticipantsExceeded } } @@ -259,9 +258,6 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me if r.FirstJoinedAt() == 0 { r.joinedAt.Store(time.Now().Unix()) } - if !participant.Hidden() { - r.protoRoom.NumParticipants++ - } // it's important to set this before connection, we don't want to miss out on any published tracks participant.OnTrackPublished(r.onTrackPublished) @@ -340,7 +336,9 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me if participant.IsRecorder() && !r.protoRoom.ActiveRecording { r.protoRoom.ActiveRecording = true - r.sendRoomUpdateLocked() + r.protoProxy.MarkDirty(true) + } else { + r.protoProxy.MarkDirty(false) } r.participants[participant.Identity()] = participant @@ -419,10 +417,7 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, requestSource routing return err } - r.lock.RLock() - p.SendRoomUpdate(r.protoRoom) - r.lock.RUnlock() - + p.SendRoomUpdate(r.ToProto()) p.ICERestart(nil) return nil } @@ -445,6 +440,7 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek } } + immediateChange := false if (p != nil && p.IsRecorder()) || r.protoRoom.ActiveRecording { activeRecording := false for _, op := range r.participants { @@ -456,10 +452,12 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek if r.protoRoom.ActiveRecording != activeRecording { r.protoRoom.ActiveRecording = activeRecording - r.sendRoomUpdateLocked() + immediateChange = true + } } r.lock.Unlock() + r.protoProxy.MarkDirty(immediateChange) if !ok { return @@ -634,6 +632,7 @@ func (r *Room) Close() { for _, p := range r.GetParticipants() { _ = p.Close(true, types.ParticipantCloseReasonRoomClose) } + r.protoProxy.Stop() if r.onClose != nil { r.onClose() } @@ -661,32 +660,26 @@ func (r *Room) SetMetadata(metadata string) { r.lock.Lock() r.protoRoom.Metadata = metadata r.lock.Unlock() - - r.lock.RLock() - r.sendRoomUpdateLocked() - r.lock.RUnlock() - - if r.onMetadataUpdate != nil { - r.onMetadataUpdate(metadata) - } + r.protoProxy.MarkDirty(true) } -func (r *Room) sendRoomUpdateLocked() { +func (r *Room) sendRoomUpdate() { + roomInfo := r.ToProto() // Send update to participants - for _, p := range r.participants { + for _, p := range r.GetParticipants() { if !p.IsReady() { continue } - err := p.SendRoomUpdate(r.protoRoom) + err := p.SendRoomUpdate(roomInfo) if err != nil { r.Logger.Warnw("failed to send room update", err, "participant", p.Identity()) } } } -func (r *Room) OnMetadataUpdate(f func(metadata string)) { - r.onMetadataUpdate = f +func (r *Room) OnRoomUpdated(f func()) { + r.onRoomUpdated = f } func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScenario *livekit.SimulateScenario) error { @@ -760,11 +753,9 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS } return &livekit.JoinResponse{ - Room: r.protoRoom, + Room: r.ToProto(), Participant: participant.ToProto(), OtherParticipants: otherParticipants, - ServerVersion: r.serverInfo.Version, - ServerRegion: r.serverInfo.Region, IceServers: iceServers, // indicates both server and client support subscriber as primary SubscriberPrimary: participant.SubscriberAsPrimary(), @@ -1003,15 +994,39 @@ func (r *Room) pushAndDequeueUpdates(pi *livekit.ParticipantInfo, isImmediate bo return updates } -func (r *Room) subscriberBroadcastWorker() { - ticker := time.NewTicker(subscriberUpdateInterval) - defer ticker.Stop() +func (r *Room) updateProto() *livekit.Room { + r.lock.RLock() + room := proto.Clone(r.protoRoom).(*livekit.Room) + r.lock.RUnlock() + + room.NumPublishers = 0 + room.NumParticipants = 0 + for _, p := range r.GetParticipants() { + if !p.IsRecorder() { + room.NumParticipants++ + } + if p.IsPublisher() { + room.NumPublishers++ + } + } + + return room +} + +func (r *Room) changeUpdateWorker() { + subTicker := time.NewTicker(subscriberUpdateInterval) + defer subTicker.Stop() for !r.IsClosed() { select { case <-r.closed: return - case <-ticker.C: + case <-r.protoProxy.Updated(): + if r.onRoomUpdated != nil { + r.onRoomUpdated() + } + r.sendRoomUpdate() + case <-subTicker.C: r.batchedUpdatesMu.Lock() updatesMap := r.batchedUpdates r.batchedUpdates = make(map[livekit.ParticipantIdentity]*livekit.ParticipantInfo) diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index 8d93367bf..f95fb5e2c 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -35,6 +35,7 @@ func init() { }) // allow immediate closure in testing RoomDepartureGrace = 1 + roomUpdateInterval = defaultDelay } var iceServersForRoom = []*livekit.ICEServer{{Urls: []string{"stun:stun.l.google.com:19302"}}} @@ -645,7 +646,7 @@ func TestHiddenParticipants(t *testing.T) { require.Len(t, res.OtherParticipants, 2) require.Len(t, rm.GetParticipants(), 4) require.NotEmpty(t, res.IceServers) - require.Equal(t, "testregion", res.ServerRegion) + require.Equal(t, "testregion", res.ServerInfo.Region) }) t.Run("hidden participant subscribes to tracks", func(t *testing.T) { @@ -665,15 +666,35 @@ func TestHiddenParticipants(t *testing.T) { } func TestRoomUpdate(t *testing.T) { + t.Run("updates are sent when participant joined", func(t *testing.T) { + rm := newRoomWithParticipants(t, testRoomOpts{num: 1}) + defer rm.Close() + + p1 := rm.GetParticipants()[0].(*typesfakes.FakeLocalParticipant) + require.Equal(t, 0, p1.SendRoomUpdateCallCount()) + + p2 := newMockParticipant("p2", types.CurrentProtocol, false, false) + require.NoError(t, rm.Join(p2, nil, nil, iceServersForRoom)) + + // p1 should have received an update + time.Sleep(2 * defaultDelay) + require.Equal(t, 1, p1.SendRoomUpdateCallCount()) + require.EqualValues(t, 2, p1.SendRoomUpdateArgsForCall(0).NumParticipants) + }) + t.Run("participants should receive metadata update", func(t *testing.T) { rm := newRoomWithParticipants(t, testRoomOpts{num: 2}) defer rm.Close() rm.SetMetadata("test metadata...") + // callbacks are updated from goroutine + time.Sleep(2 * defaultDelay) + for _, op := range rm.GetParticipants() { fp := op.(*typesfakes.FakeLocalParticipant) - require.Equal(t, 1, fp.SendRoomUpdateCallCount()) + // room updates are now sent for both participant joining and room metadata + require.GreaterOrEqual(t, fp.SendRoomUpdateCallCount(), 1) } }) } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 83b22b833..1163cc1d2 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -181,6 +181,7 @@ type Participant interface { SetName(name string) SetMetadata(metadata string) + IsPublisher() bool GetPublishedTrack(sid livekit.TrackID) MediaTrack GetPublishedTracks() []MediaTrack RemovePublishedTrack(track MediaTrack, willBeResumed bool, shouldClose bool) @@ -284,7 +285,6 @@ type LocalParticipant interface { // returns list of participant identities that the current participant is subscribed to GetSubscribedParticipants() []livekit.ParticipantID IsSubscribedTo(sid livekit.ParticipantID) bool - IsPublisher() bool GetAudioLevel() (smoothedLevel float64, active bool) GetConnectionQuality() *livekit.ConnectionQualityInfo diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 29c8cb032..404af0a17 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -95,6 +95,16 @@ type FakeParticipant struct { identityReturnsOnCall map[int]struct { result1 livekit.ParticipantIdentity } + IsPublisherStub func() bool + isPublisherMutex sync.RWMutex + isPublisherArgsForCall []struct { + } + isPublisherReturns struct { + result1 bool + } + isPublisherReturnsOnCall map[int]struct { + result1 bool + } IsRecorderStub func() bool isRecorderMutex sync.RWMutex isRecorderArgsForCall []struct { @@ -637,6 +647,59 @@ func (fake *FakeParticipant) IdentityReturnsOnCall(i int, result1 livekit.Partic }{result1} } +func (fake *FakeParticipant) IsPublisher() bool { + fake.isPublisherMutex.Lock() + ret, specificReturn := fake.isPublisherReturnsOnCall[len(fake.isPublisherArgsForCall)] + fake.isPublisherArgsForCall = append(fake.isPublisherArgsForCall, struct { + }{}) + stub := fake.IsPublisherStub + fakeReturns := fake.isPublisherReturns + fake.recordInvocation("IsPublisher", []interface{}{}) + fake.isPublisherMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) IsPublisherCallCount() int { + fake.isPublisherMutex.RLock() + defer fake.isPublisherMutex.RUnlock() + return len(fake.isPublisherArgsForCall) +} + +func (fake *FakeParticipant) IsPublisherCalls(stub func() bool) { + fake.isPublisherMutex.Lock() + defer fake.isPublisherMutex.Unlock() + fake.IsPublisherStub = stub +} + +func (fake *FakeParticipant) IsPublisherReturns(result1 bool) { + fake.isPublisherMutex.Lock() + defer fake.isPublisherMutex.Unlock() + fake.IsPublisherStub = nil + fake.isPublisherReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeParticipant) IsPublisherReturnsOnCall(i int, result1 bool) { + fake.isPublisherMutex.Lock() + defer fake.isPublisherMutex.Unlock() + fake.IsPublisherStub = nil + if fake.isPublisherReturnsOnCall == nil { + fake.isPublisherReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.isPublisherReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeParticipant) IsRecorder() bool { fake.isRecorderMutex.Lock() ret, specificReturn := fake.isRecorderReturnsOnCall[len(fake.isRecorderArgsForCall)] @@ -1118,6 +1181,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.iDMutex.RUnlock() fake.identityMutex.RLock() defer fake.identityMutex.RUnlock() + fake.isPublisherMutex.RLock() + defer fake.isPublisherMutex.RUnlock() fake.isRecorderMutex.RLock() defer fake.isRecorderMutex.RUnlock() fake.removePublishedTrackMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 0244abb84..4a63ea4e1 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -451,7 +451,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room newRoom.Logger.Infow("room closed") }) - newRoom.OnMetadataUpdate(func(metadata string) { + newRoom.OnRoomUpdated(func() { if err := r.roomStore.StoreRoom(ctx, newRoom.ToProto(), newRoom.Internal()); err != nil { newRoom.Logger.Errorw("could not handle metadata update", err) }