dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-17 01:12:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert "Cache data synchronously for processing in worker." (#2425)

Browse Source
This commit is contained in:
Raja Subramanian
2024-01-29 12:36:27 +05:30
committed by GitHub
parent 0be241eed8
commit 846121e781
3 changed files with 37 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/room.go
+32 -52
View File
@@ -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
pkg/rtc/room_test.go
+1 -1
View File
@@ -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()
pkg/service/roommanager.go
+4 -4
View File
@@ -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)
}
}
})