ライブストリーミング時に必要なGoのAPIの番組キャッシュ機構

こんにちは。新規事業開発室、LUCRA開発チームの高橋（@__timakin__）です。

現在LUCRAでは、ライブ番組の放送を行なっております。

prtimes.jp

ライブストリーミングというのは無尽蔵に改善事項が生まれるタスク生成器のような開発分野の一つでありますが、今回はその中でも、APIの番組情報のキャッシュ機構について書きたいと思います。

LUCRAで行なっているライブ放送

LUCRAではライブ放送機能のリリース時点に、以下3つの番組を放送していました。

番組によってはクイズ形式で視聴者参加型のコンテンツもあり、企画に応じて様々なAPIが必要となります。

ライブ放送開始時に、APIは何をするのか？

みなさんがよく利用されるライブ放送のアプリなどは、放送開始時に何をしているのでしょうか？

放送までの時系列も含めて整理すると、大体は以下のような流れかと思います。

1. 番組放送までは、予告情報を表示する

2. 番組が開始したら予告 -> 放送中ステータスに切り替える

3. 放送終了後は表示を消す or 次回予告を出す（1に戻る）

今回書くのは、これらのステータスが切り替わるタイミングで、APIはどのようにしてキャッシュを切り替えているのか？という話です。

APIは番組データをどうキャッシュすべきか

どのようにしてキャッシュを切り替えているのか？ と書きましたが、なぜこれを気にしなければいけないのでしょうか？

それは番組の演出・ユーザー体験と深く関係しています。

弊社のアプリでは、前述の順序1のように、放送前に予告情報が表示されています。「22:00~ 放送開始」というような表記です。

加えて、放送直前にカウントダウンする、開始時刻を明確に意識させるような演出があります。

つまりユーザー体験の観点から内容のリフレッシュのスピードに対して注意しなくてはならない、ということです。

放送ステータス切り替えイベントの伝播

次に、LUCRAでは、「放送開始 -> 放送中 -> 放送終了」というステータス切り替えをどう伝播しているのか書いていきます。

結論から書くと、RedisのPubsub経由で送受信されるステータス情報に基づいて、複数の番組レコードのどれを参照するか決定しています。

手順としては以下の通りです。

1. LUCRAのライブ放送管理画面で、「放送開始ボタン」を押す

2. 管理画面のサーバー経由でRedisにリクエストが行き、ステータス更新のPublishイベントが発行される

3. 上記のイベントをSubscribeする番組情報APIの中で、最新ステータスにひもづくデータにキャッシュがすげ変わる

この流れですが、重要なのが 最新ステータスにひもづくデータにキャッシュがすげ変わる というポイントです。

イベントを受け取ったAPIがキャッシュを即時更新するには

上記の3で、Redisからの番組放送ステータスをSubscribeするわけですが、これは初期化処理の中で購読を開始します。

レイヤー化されたアーキテクチャの中で、DDDのアプリケーション層に近いところに、StartStatusSubscriptionと言うメソッドを生やしています。

liveProgramService.StartStatusSubscription(context.Background())

この実装は以下の通りです。

書き方が少しややこしいですが、やってるのは、「RedisのPublishイベントを受け取って、番組のURLなどを持ったデータと、番組をビューとして表示するために必要な、見た目に関わるデータのキャッシュを更新する」というバックグラウンドプロセスを、1度だけ起動することです。

DB内には、「予告」「放送中」のそれぞれにひもづくビュー情報やサムネイル画像の情報が格納されています。そのいずれを使うか、一番最後に受け取ったRedisのイベント情報から決め、キャッシュすると言う仕組みです。

	// 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)
	})
	}

view raw sub.go hosted with ❤ by GitHub

gist.github.com

GoのAPIに関していえば、細かくレイヤー化されたアーキテクチャに基づく事例をいくつか最近は見ますが、正直どのレイヤーにこのPubSubの処理を書くか、迷うところでした。

ただ、複数のリソースを呼び出してリポジトリ層に接続すると言う意味では、このようにアプリケーションレイヤにまとめるのが良いだろうと言う結論でした。

最終的にやりたかったのは、即時にキャッシュを切り替えると言うことでしたが、そのためにはあらかじめデータを複数パターン作っておいて、どのパターンを使うかというのを通知ベースで切り替える、と言うのが一つの解決策だと考えています。

まとめ

以上、ライブ動画の放送ステータスが変わったタイミングで、いかにして即時にキャッシュを切り替えるかという実装の紹介になります。

より大規模な基盤を持っているような企業様では、これとは少し違う実装かもしれませんが、初期リリース時点ではこのような実装も可能性の一個としてある、と参考にしていただけると幸いです。