Update egress info
@@ -35,7 +35,7 @@ func NewEgressService(bus utils.MessageBus, store ServiceStore, rs livekit.RoomS
func (s *EgressService) Start() {
if s.bus != nil {
- go s.resultsWorker()
+ go s.updateListener()
}
}
@@ -221,7 +221,7 @@ func (s *EgressService) StopEgress(ctx context.Context, req *livekit.StopEgressR
return info, nil
}
-func (s *EgressService) resultsWorker() {
+func (s *EgressService) updateListener() {
sub, err := s.bus.SubscribeQueue(context.Background(), egress.ResultsChannel)
if err != nil {
logger.Errorw("failed to subscribe to results channel", err)
@@ -240,6 +240,13 @@ func (s *EgressService) resultsWorker() {
continue
}
+ switch res.Status {
+ case livekit.EgressStatus_EGRESS_ACTIVE, livekit.EgressStatus_EGRESS_ENDING:
+ // save updated info to store
+ err = s.store.UpdateEgress(context.Background(), res)
+
+ case livekit.EgressStatus_EGRESS_COMPLETE:
+ // delete from store
err = s.store.DeleteEgress(context.Background(), res)
if err != nil {
logger.Errorw("could not delete egress from store", err)
@@ -253,6 +260,7 @@ func (s *EgressService) resultsWorker() {
}
s.telemetry.EgressEnded(context.Background(), res)
+ }
case <-s.shutdown:
_ = sub.Close()
return
