From 846121e781e043b5955b8d49be5035f4c8457083 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 29 Jan 2024 12:36:27 +0530 Subject: [PATCH] Revert "Cache data synchronously for processing in worker." (#2425) --- pkg/rtc/room.go | 84 +++++++++++++++----------------------- pkg/rtc/room_test.go | 2 +- pkg/service/roommanager.go | 8 ++-- 3 files changed, 37 insertions(+), 57 deletions(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index d95044ed9..745ae9f4f 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -118,7 +118,7 @@ type Room struct { trailer []byte - onParticipantChanged func(p types.LocalParticipant, pi *livekit.ParticipantInfo) + onParticipantChanged func(p types.LocalParticipant) onRoomUpdated func() onClose func() @@ -347,13 +347,11 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me pw := r.addParticipantWorkerLocked(participant) participant.OnStateChange(func(p types.LocalParticipant, state livekit.ParticipantInfo_State) { - ri := r.ToProto() - pi := p.ToProto() pw.eventsQueue.Enqueue(func() { if r.onParticipantChanged != nil { - r.onParticipantChanged(p, pi) + r.onParticipantChanged(p) } - r.broadcastParticipantState(p, pi, broadcastOptions{skipSource: true}) + r.broadcastParticipantState(p, broadcastOptions{skipSource: true}) if state == livekit.ParticipantInfo_ACTIVE { // subscribe participant to existing published tracks @@ -370,8 +368,8 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me } } r.telemetry.ParticipantActive(context.Background(), - ri, - pi, + r.ToProto(), + p.ToProto(), meta, false, ) @@ -385,29 +383,23 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me }) // it's important to set this before connection, we don't want to miss out on any published tracks participant.OnTrackPublished(func(p types.LocalParticipant, t types.MediaTrack) { - pi := p.ToProto() - ti := t.ToProto() pw.eventsQueue.Enqueue(func() { - r.onTrackPublished(p, pi, t, ti) + r.onTrackPublished(p, t) }) }) participant.OnTrackUpdated(func(p types.LocalParticipant, t types.MediaTrack) { - pi := p.ToProto() pw.eventsQueue.Enqueue(func() { - r.onTrackUpdated(p, pi, t) + r.onTrackUpdated(p, t) }) }) participant.OnTrackUnpublished(func(p types.LocalParticipant, t types.MediaTrack) { - pi := p.ToProto() - ti := t.ToProto() pw.eventsQueue.Enqueue(func() { - r.onTrackUnpublished(p, pi, t, ti) + r.onTrackUnpublished(p, t) }) }) participant.OnParticipantUpdate(func(p types.LocalParticipant) { - pi := p.ToProto() pw.eventsQueue.Enqueue(func() { - r.onParticipantUpdate(p, pi) + r.onParticipantUpdate(p) }) }) participant.OnDataPacket(r.onDataPacket) @@ -468,7 +460,7 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me r.participantRequestSources[participant.Identity()] = requestSource if r.onParticipantChanged != nil { - r.onParticipantChanged(participant, participant.ToProto()) + r.onParticipantChanged(participant) } time.AfterFunc(time.Minute, func() { @@ -645,11 +637,10 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek r.leftAt.Store(time.Now().Unix()) if sendUpdates { - pi := p.ToProto() if r.onParticipantChanged != nil { - r.onParticipantChanged(p, pi) + r.onParticipantChanged(p) } - r.broadcastParticipantState(p, pi, broadcastOptions{skipSource: true}) + r.broadcastParticipantState(p, broadcastOptions{skipSource: true}) } } @@ -835,7 +826,7 @@ func (r *Room) OnClose(f func()) { r.onClose = f } -func (r *Room) OnParticipantChanged(f func(p types.LocalParticipant, pi *livekit.ParticipantInfo)) { +func (r *Room) OnParticipantChanged(f func(participant types.LocalParticipant)) { r.onParticipantChanged = f } @@ -997,14 +988,9 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS } // a ParticipantImpl in the room added a new track, subscribe other participants to it -func (r *Room) onTrackPublished( - participant types.LocalParticipant, - pi *livekit.ParticipantInfo, - track types.MediaTrack, - ti *livekit.TrackInfo, -) { +func (r *Room) onTrackPublished(participant types.LocalParticipant, track types.MediaTrack) { // publish participant update, since track state is changed - r.broadcastParticipantState(participant, pi, broadcastOptions{skipSource: true}) + r.broadcastParticipantState(participant, broadcastOptions{skipSource: true}) r.lock.RLock() // subscribe all existing participants to this MediaTrack @@ -1033,7 +1019,7 @@ func (r *Room) onTrackPublished( r.lock.RUnlock() if onParticipantChanged != nil { - onParticipantChanged(participant, pi) + onParticipantChanged(participant) } r.trackManager.AddTrack(track, participant.Identity(), participant.ID()) @@ -1086,51 +1072,42 @@ func (r *Room) onTrackPublished( context.Background(), participant.ID(), participant.Identity(), - ti, + track.ToProto(), ) } -func (r *Room) onTrackUpdated( - p types.LocalParticipant, - pi *livekit.ParticipantInfo, - _ types.MediaTrack, -) { +func (r *Room) onTrackUpdated(p types.LocalParticipant, _ types.MediaTrack) { // send track updates to everyone, especially if track was updated by admin - r.broadcastParticipantState(p, pi, broadcastOptions{}) + r.broadcastParticipantState(p, broadcastOptions{}) if r.onParticipantChanged != nil { - r.onParticipantChanged(p, pi) + r.onParticipantChanged(p) } } -func (r *Room) onTrackUnpublished( - p types.LocalParticipant, - pi *livekit.ParticipantInfo, - track types.MediaTrack, - ti *livekit.TrackInfo, -) { +func (r *Room) onTrackUnpublished(p types.LocalParticipant, track types.MediaTrack) { r.telemetry.TrackUnpublished( context.Background(), p.ID(), p.Identity(), - ti, + track.ToProto(), !p.IsClosed(), ) r.trackManager.RemoveTrack(track) if !p.IsClosed() { - r.broadcastParticipantState(p, pi, broadcastOptions{skipSource: true}) + r.broadcastParticipantState(p, broadcastOptions{skipSource: true}) } if r.onParticipantChanged != nil { - r.onParticipantChanged(p, pi) + r.onParticipantChanged(p) } } -func (r *Room) onParticipantUpdate(p types.LocalParticipant, pi *livekit.ParticipantInfo) { +func (r *Room) onParticipantUpdate(p types.LocalParticipant) { r.protoProxy.MarkDirty(false) // immediately notify when permissions or metadata changed - r.broadcastParticipantState(p, pi, broadcastOptions{immediate: true}) + r.broadcastParticipantState(p, broadcastOptions{immediate: true}) if r.onParticipantChanged != nil { - r.onParticipantChanged(p, pi) + r.onParticipantChanged(p) } } @@ -1165,13 +1142,16 @@ func (r *Room) subscribeToExistingTracks(p types.LocalParticipant) { } // broadcast an update about participant p -func (r *Room) broadcastParticipantState(p types.LocalParticipant, pi *livekit.ParticipantInfo, opts broadcastOptions) { +func (r *Room) broadcastParticipantState(p types.LocalParticipant, opts broadcastOptions) { + pi := p.ToProto() + if p.Hidden() { if !opts.skipSource { // send update only to hidden participant err := p.SendParticipantUpdate([]*livekit.ParticipantInfo{pi}) if err != nil { - p.GetLogger().Errorw("could not send update to participant", err) + r.Logger.Errorw("could not send update to participant", err, + "participant", p.Identity(), "pID", p.ID()) } } return diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index ec564c4af..4990b64d4 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -127,7 +127,7 @@ func TestRoomJoin(t *testing.T) { t.Run("participant state change is broadcasted to others", func(t *testing.T) { rm := newRoomWithParticipants(t, testRoomOpts{num: numParticipants}) var changedParticipant types.Participant - rm.OnParticipantChanged(func(participant types.LocalParticipant, _pi *livekit.ParticipantInfo) { + rm.OnParticipantChanged(func(participant types.LocalParticipant) { changedParticipant = participant }) participants := rm.GetParticipants() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 1c010acbb..abfd29331 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -577,10 +577,10 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room } }) - newRoom.OnParticipantChanged(func(p types.LocalParticipant, pi *livekit.ParticipantInfo) { - if pi.State != livekit.ParticipantInfo_DISCONNECTED { - if err := r.roomStore.StoreParticipant(ctx, roomName, pi); err != nil { - p.GetLogger().Errorw("could not handle participant change", err) + newRoom.OnParticipantChanged(func(p types.LocalParticipant) { + if !p.IsDisconnected() { + if err := r.roomStore.StoreParticipant(ctx, roomName, p.ToProto()); err != nil { + newRoom.Logger.Errorw("could not handle participant change", err) } } })