From 939ed10ca6be1d9852cb6ef7ef13136920b6929a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Sun, 3 Oct 2021 07:08:29 +0200 Subject: [PATCH] Room Metadata Implementation (#126) --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/participant.go | 10 +++ pkg/rtc/room.go | 26 +++++++ pkg/rtc/room_test.go | 15 ++++ pkg/rtc/types/interfaces.go | 1 + pkg/rtc/types/typesfakes/fake_participant.go | 74 ++++++++++++++++++++ pkg/service/roommanager.go | 9 +++ pkg/service/roomservice.go | 31 ++++++++ 9 files changed, 169 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index a8af55278..c4b42ff47 100644 --- a/go.mod +++ b/go.mod @@ -14,7 +14,7 @@ require ( github.com/google/wire v0.5.0 github.com/gorilla/websocket v1.4.2 github.com/jxskiss/base62 v0.0.0-20191017122030-4f11678b909b - github.com/livekit/protocol v0.9.3 + github.com/livekit/protocol v0.9.4 github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 diff --git a/go.sum b/go.sum index 6d3911f1e..a3ed57b53 100644 --- a/go.sum +++ b/go.sum @@ -246,8 +246,8 @@ github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= github.com/livekit/ion-sfu v1.20.9 h1:Ih+tE5CXqdL1o6iXfZn+uBoddBXWceaOKVIun6aVuZM= github.com/livekit/ion-sfu v1.20.9/go.mod h1:g8hwobZI5fvX1RXvayf4ZXkgP7spV5YGE4yTSsumpB4= -github.com/livekit/protocol v0.9.3 h1:rzJwes4AddFF/h47G6kU1/GQF8/meqZ3uihFIKPPxz0= -github.com/livekit/protocol v0.9.3/go.mod h1:MEKn847Iu/2U8ClZyUmEm2oHn8k9fnSHy81Wv8kkSDo= +github.com/livekit/protocol v0.9.4 h1:1vA49MwawrriOS3VvPqYfLiFI94XWvYGaWfkHN6ah6Y= +github.com/livekit/protocol v0.9.4/go.mod h1:MEKn847Iu/2U8ClZyUmEm2oHn8k9fnSHy81Wv8kkSDo= github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ= github.com/lucsky/cuid v1.0.2/go.mod h1:QaaJqckboimOmhRSJXSx/+IT+VTfxfPGSo/6mfgUfmE= github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index da0c5149a..ebbe3129c 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -578,6 +578,16 @@ func (p *ParticipantImpl) SendDataPacket(dp *livekit.DataPacket) error { return dc.Send(data) } +func (p *ParticipantImpl) SendRoomUpdate(room *livekit.Room) error { + return p.writeMessage(&livekit.SignalResponse{ + Message: &livekit.SignalResponse_RoomUpdate{ + RoomUpdate: &livekit.RoomUpdate{ + Room: room, + }, + }, + }) +} + func (p *ParticipantImpl) SetTrackMuted(trackId string, muted bool, fromAdmin bool) { isPending := false p.lock.RLock() diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index d525510e1..afac66dff 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -47,6 +47,7 @@ type Room struct { statsReporter *stats.RoomStatsReporter onParticipantChanged func(p types.Participant) + onMetadataUpdate func(metadata string) onClose func() } @@ -73,6 +74,7 @@ func NewRoom(room *livekit.Room, config WebRTCConfig, iceServers []*livekit.ICES } r.statsReporter.RoomStarted() go r.audioUpdateWorker() + return r } @@ -374,6 +376,30 @@ func (r *Room) SendDataPacket(up *livekit.UserPacket, kind livekit.DataPacket_Ki r.onDataPacket(nil, dp) } +func (r *Room) SetMetadata(metadata string) { + r.Room.Metadata = metadata + + // Send update to participants + for _, p := range r.GetParticipants() { + if !p.IsReady() { + continue + } + + err := p.SendRoomUpdate(r.Room) + if err != nil { + logger.Warnw("failed to send room update", err, "room", r.Room.Name, "participant", p.Identity()) + } + } + + if r.onMetadataUpdate != nil { + r.onMetadataUpdate(metadata) + } +} + +func (r *Room) OnMetadataUpdate(f func(metadata string)) { + r.onMetadataUpdate = f +} + // checks if participant should be autosubscribed to new tracks, assumes lock is already acquired func (r *Room) autoSubscribe(participant types.Participant) bool { if !participant.CanSubscribe() { diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index 273b99045..ea581e54a 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -520,6 +520,20 @@ func TestHiddenParticipants(t *testing.T) { }) } +func TestRoomUpdate(t *testing.T) { + t.Run("participants should receive metadata update", func(t *testing.T) { + rm := newRoomWithParticipants(t, testRoomOpts{num: 2}) + defer rm.Close() + + rm.SetMetadata("test metadata...") + + for _, op := range rm.GetParticipants() { + fp := op.(*typesfakes.FakeParticipant) + require.Equal(t, 1, fp.SendRoomUpdateCallCount()) + } + }) +} + type testRoomOpts struct { num int numHidden int @@ -548,6 +562,7 @@ func newRoomWithParticipants(t *testing.T, opts testRoomOpts) *rtc.Room { participant := newMockParticipant(identity, opts.protocol, i >= opts.num) err := rm.Join(participant, &rtc.ParticipantOptions{AutoSubscribe: true}) participant.StateReturns(livekit.ParticipantInfo_ACTIVE) + participant.IsReadyReturns(true) require.NoError(t, err) } return rm diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index ea0bd57d1..6f88b7fcc 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -50,6 +50,7 @@ type Participant interface { SendParticipantUpdate(participants []*livekit.ParticipantInfo) error SendSpeakerUpdate(speakers []*livekit.SpeakerInfo) error SendDataPacket(packet *livekit.DataPacket) error + SendRoomUpdate(room *livekit.Room) error SetTrackMuted(trackId string, muted bool, fromAdmin bool) GetAudioLevel() (level uint8, active bool) diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 2ec93faa9..bd47e5b7f 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -325,6 +325,17 @@ type FakeParticipant struct { sendParticipantUpdateReturnsOnCall map[int]struct { result1 error } + SendRoomUpdateStub func(*livekit.Room) error + sendRoomUpdateMutex sync.RWMutex + sendRoomUpdateArgsForCall []struct { + arg1 *livekit.Room + } + sendRoomUpdateReturns struct { + result1 error + } + sendRoomUpdateReturnsOnCall map[int]struct { + result1 error + } SendSpeakerUpdateStub func([]*livekit.SpeakerInfo) error sendSpeakerUpdateMutex sync.RWMutex sendSpeakerUpdateArgsForCall []struct { @@ -2117,6 +2128,67 @@ func (fake *FakeParticipant) SendParticipantUpdateReturnsOnCall(i int, result1 e }{result1} } +func (fake *FakeParticipant) SendRoomUpdate(arg1 *livekit.Room) error { + fake.sendRoomUpdateMutex.Lock() + ret, specificReturn := fake.sendRoomUpdateReturnsOnCall[len(fake.sendRoomUpdateArgsForCall)] + fake.sendRoomUpdateArgsForCall = append(fake.sendRoomUpdateArgsForCall, struct { + arg1 *livekit.Room + }{arg1}) + stub := fake.SendRoomUpdateStub + fakeReturns := fake.sendRoomUpdateReturns + fake.recordInvocation("SendRoomUpdate", []interface{}{arg1}) + fake.sendRoomUpdateMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) SendRoomUpdateCallCount() int { + fake.sendRoomUpdateMutex.RLock() + defer fake.sendRoomUpdateMutex.RUnlock() + return len(fake.sendRoomUpdateArgsForCall) +} + +func (fake *FakeParticipant) SendRoomUpdateCalls(stub func(*livekit.Room) error) { + fake.sendRoomUpdateMutex.Lock() + defer fake.sendRoomUpdateMutex.Unlock() + fake.SendRoomUpdateStub = stub +} + +func (fake *FakeParticipant) SendRoomUpdateArgsForCall(i int) *livekit.Room { + fake.sendRoomUpdateMutex.RLock() + defer fake.sendRoomUpdateMutex.RUnlock() + argsForCall := fake.sendRoomUpdateArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeParticipant) SendRoomUpdateReturns(result1 error) { + fake.sendRoomUpdateMutex.Lock() + defer fake.sendRoomUpdateMutex.Unlock() + fake.SendRoomUpdateStub = nil + fake.sendRoomUpdateReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeParticipant) SendRoomUpdateReturnsOnCall(i int, result1 error) { + fake.sendRoomUpdateMutex.Lock() + defer fake.sendRoomUpdateMutex.Unlock() + fake.SendRoomUpdateStub = nil + if fake.sendRoomUpdateReturnsOnCall == nil { + fake.sendRoomUpdateReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.sendRoomUpdateReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeParticipant) SendSpeakerUpdate(arg1 []*livekit.SpeakerInfo) error { var arg1Copy []*livekit.SpeakerInfo if arg1 != nil { @@ -2675,6 +2747,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.sendJoinResponseMutex.RUnlock() fake.sendParticipantUpdateMutex.RLock() defer fake.sendParticipantUpdateMutex.RUnlock() + fake.sendRoomUpdateMutex.RLock() + defer fake.sendRoomUpdateMutex.RUnlock() fake.sendSpeakerUpdateMutex.RLock() defer fake.sendSpeakerUpdateMutex.RUnlock() fake.setMetadataMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 1148433b8..32bb96973 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -304,6 +304,12 @@ func (r *LocalRoomManager) getOrCreateRoom(ctx context.Context, roomName string) "outgoingStats", room.GetOutgoingStats().Copy(), ) }) + room.OnMetadataUpdate(func(metadata string) { + err := r.StoreRoom(ctx, room.Room) + if err != nil { + logger.Errorw("could not handle metadata update", err) + } + }) room.OnParticipantChanged(func(p types.Participant) { var err error if p.State() == livekit.ParticipantInfo_DISCONNECTED { @@ -482,6 +488,9 @@ func (r *LocalRoomManager) handleRTCMessage(ctx context.Context, roomName, ident DestinationSids: rm.SendData.DestinationSids, } room.SendDataPacket(up, rm.SendData.Kind) + case *livekit.RTCNodeMessage_UpdateRoomMetadata: + logger.Debugw("updating room", "room", roomName) + room.SetMetadata(rm.UpdateRoomMetadata.Metadata) } } diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index c5e9ff782..6aa62b585 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -215,6 +215,37 @@ func (s *RoomService) SendData(ctx context.Context, req *livekit.SendDataRequest return &livekit.SendDataResponse{}, nil } +func (s *RoomService) UpdateRoomMetadata(ctx context.Context, req *livekit.UpdateRoomMetadataRequest) (*livekit.Room, error) { + if err := EnsureAdminPermission(ctx, req.Room); err != nil { + return nil, twirpAuthError(err) + } + + room, err := s.roomStore.LoadRoom(ctx, req.Room) + if err != nil { + return nil, err + } + + room.Metadata = req.Metadata + + participants, err := s.roomStore.ListParticipants(ctx, req.Room) + if err != nil { + return nil, err + } + + if len(participants) > 0 { + err := s.writeMessage(ctx, req.Room, participants[0].Identity, &livekit.RTCNodeMessage{ + Message: &livekit.RTCNodeMessage_UpdateRoomMetadata{ + UpdateRoomMetadata: req, + }, + }) + if err != nil { + return nil, err + } + } + + return room, nil +} + func (s *RoomService) writeMessage(ctx context.Context, room, identity string, msg *livekit.RTCNodeMessage) error { if err := EnsureAdminPermission(ctx, room); err != nil { return twirpAuthError(err)