Live status subscription

sub.go

// StartStatusSubscription ... ライブ動画の放送状況の監視
func (s *liveProgramService) StartStatusSubscription(ctx context.Context) {
  s.subLock.Do(func() {
    sub, err := s.sub.GetSubscriber()
    if err != nil {
      panic(err)
    }
    go func(sub *redis.PubSub) {
      defer sub.Close()
      for {
        m, err := sub.ReceiveMessage()
        if err != nil {
          log.Error(err)
          continue
        }
        if m.Payload != "" {
          // 動画一覧のキャッシュを更新
          vs, err := s.repo.GetLiveVideos(ctx)
          if err != nil {
            log.Error(err)
          }
          err = s.crepo.SetLiveVideos(ctx, vs)
          if err != nil {
            log.Error(err)
          }

          for _, v := range vs {
            err = s.crepo.SetLiveVideo(ctx, v)
            log.Error(err)
          }

          // セル一覧のキャッシュを更新
          cells, err := s.repo.GetLiveVideoCells(ctx)
          if err != nil {
            log.Error(err)
          }
          err = s.crepo.SetLiveVideoCells(ctx, cells)
          if err != nil {
            log.Error(err)
          }
        }
      }
    }(sub)
  })
}