オリジナル配信サイトに視聴者数を取得して表示する機能をコマンド一発で構築する

AWS Cloud9 を使えば、 Web ブラウザ経由で専用の開発環境を構築・利用できるので、誰でも簡単に同じスタートラインで開発を始められます。今回は Cloud9 を使ってハンズオンを実施するので、下記の手順に沿ってまずは Cloud9 環境のセットアップをしましょう。

*注: 以下の手順は全て IVS のリソースに合わせて バージニア北部 (us-east-1)リージョン で実行していきます。

AWS Cloud9 が利用できない場合、こちらのブログ をご参考に AWS IDE Toolkits または AWS CloudShell をご利用ください。

AWS Cloud9 から AWS IDE Toolkits または AWS CloudShell に移行する方法 »

AWS マネジメントコンソール から、検索フォームに「cloud9」と入力し、 「Cloud9」をクリックします。

「Create environment」をクリックします。

Name に任意の環境名(今回は ivs-cdk-environment としています) を入力し、 「Next step」をクリックします。

Instance type は t3.small を選択し、それ以外はデフォルトのまま 「Next step」をクリックします。

「Create environment」をクリックします。

今回のアーキテクチャの構成と、実行する Lambda のコードを含むリポジトリをクローンし、解説していきます。

下記コマンドを実行し、リポジトリをクローンします。

git clone https://github.com/matsuihidetoshi/ivs-viewers-count-cdk.git

まず、リソース構築スクリプトの、 ivs-viewers-count-cdk/blob/main/lib/ivs-viewers-count-cdk-stack.ts から解説していきます。

import { Duration, Stack, StackProps, RemovalPolicy } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import { Table, BillingMode, AttributeType } from 'aws-cdk-lib/aws-dynamodb';
import { Rule, Schedule } from 'aws-cdk-lib/aws-events';
import { LambdaFunction } from 'aws-cdk-lib/aws-events-targets';
import { PolicyStatement } from 'aws-cdk-lib/aws-iam';
import { NodejsFunction } from "aws-cdk-lib/aws-lambda-nodejs";

export class IvsViewersCountCdkStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    // ①視聴者数保存用の DynamoDB テーブルを作成
    const ivsViewersCountTable = new Table(this, "ivsViewersCountTable", {
      billingMode: BillingMode.PAY_PER_REQUEST,
      partitionKey: { name: 'channel', type: AttributeType.STRING },
      sortKey: { name: 'time', type: AttributeType.NUMBER },
      removalPolicy: RemovalPolicy.DESTROY
    });

    // ②Lambda 関数を作成
    const ivsViewersCountFunction = new NodejsFunction(this, "ivsViewersCountFunction", {
      entry: "src/ivs-viewers-count-function.handler.ts",
      environment: {
          TABLE_NAME: ivsViewersCountTable.tableName
      }
    });

    // ③Lambda 関数に、　DynamoDB の R/W 権限を付与
    ivsViewersCountTable.grantReadWriteData(ivsViewersCountFunction);
    
    // ④Lambda 関数に、 IVS へのアクセス権限を付与
    ivsViewersCountFunction.addToRolePolicy(new PolicyStatement({
      resources: [
        'arn:aws:ivs:us-east-1:*:channel/*'
      ],
      actions: [
        'ivs:ListChannels',
        'ivs:ListStreams',
        'ivs:PutMetadata'
      ]
    }));

    // ⑤Lambda 関数の定期呼び出し用の EventBridge ルールを定義
    new Rule(this, "ivsViewersCountRule", {
      schedule: Schedule.rate(Duration.minutes(1)),
      targets: [
        new LambdaFunction(ivsViewersCountFunction)
      ]
    });
  }
}

• ①で、視聴者数保存用の DynamoDB テーブルを作成しています。

  • 集計等に便利なように、下記の Key を指定しています。

    • partitionKey に、 IVS のチャンネルの ARN を保存するためのフィールド、 channel を指定

    • sortKey に、視聴者数取得時刻を保存するためのフィールド、 time を指定

  • 今回の CDK で構築するリソース全体の作成・削除に合わせて、テーブルも削除されるように, removalPolicy: RemovalPolicy.DESTROY を指定しています。

• ②で、 Lambda 関数を作成しています。

  • NodejsFunction クラスを使用して、 Lambda の関数コード src/ivs-viewers-count-function.handler.ts を entry に指定しています。これにより、 TypeScript のコードをデプロイ時に自動的にトランスパイルしてくれます。

  • environment オプションに DynamoDB のテーブル名を指定し、環境変数から DynamoDB テーブル名を読み込み、関数からアクセスできる様にしています。

• ③で、 Lambda 関数に DynamoDB の R/W 権限を付与しています。

• ④で、 Lambda 関数に IVS へのアクセス権限を付与しています。

  • resources に 'arn:aws:ivs:us-east-1:*:channel/*' を指定していますが、これにより us-east-1 リージョンの全ての IVS のチャンネルにアクセスできます　(今回デプロイする関数コードの仕様上、上限は 50 チャンネルまでです)。

• ⑤で、 Lambda 関数を定期呼び出しする EventBridge のルールを定義しています (1 分毎に実行)。

続いて、関数コード ivs-viewers-count-cdk/src/ivs-viewers-count-function.handler.ts を解説していきます。

import {
  IvsClient,
  ListChannelsCommand,
  ListStreamsCommand,
  PutMetadataCommand
} from '@aws-sdk/client-ivs'
import {
  DynamoDB,
  BatchWriteItemCommand,
  QueryCommand
} from '@aws-sdk/client-dynamodb'

let response: object;
const dynamodb = new DynamoDB({});
const ivs = new IvsClient({});

export const handler = async () => {
  const tableName = process.env.TABLE_NAME as string;
  const time = Date.now();
  const counts: { [key: string]: number } = {};

  try {
    // ①チャンネルを取得し、視聴者数を 0 で初期化
    const listChannels = new ListChannelsCommand({});
    const resListChannels = await ivs.send(listChannels);
    if (resListChannels.channels) {
      resListChannels.channels.forEach(channel => {
        if (channel.arn) {
          counts[channel.arn] = 0;
        }
      });
    };

    // ②実際に開始されたストリーミングを取得し、視聴者数を取得
    const listStreams = new ListStreamsCommand({});
    const resListStreams = await ivs.send(listStreams);
    if (resListStreams.streams) {
      resListStreams.streams.forEach((stream) => {
        if (stream.channelArn && stream.viewerCount) {
          counts[stream.channelArn] += stream.viewerCount
        };
      });
    };

    // ③ DynamoDB に書き込みするデータを整形
    const requestItems = Object.entries(counts).map((item) => ({
      PutRequest: {
        Item: {
          channel: { S: item[0] },
          time: { N: String(time) },
          count: { N: String(item[1]) },
        },
      },
    }));

    // ④ DynamoDB に書き込み
    const batchWriteItem = new BatchWriteItemCommand({
      RequestItems: {
        [tableName]: requestItems,
      },
    });
    await dynamodb.send(batchWriteItem);

    // ⑤ DynamoDB から直近5分間の視聴者数を取得
    const queries = requestItems.map(item => {
      return new QueryCommand({
        TableName: tableName,
        KeyConditionExpression: "channel = :channel",
        ExpressionAttributeValues: {
          ':channel': { S: item.PutRequest.Item.channel.S },
        },
        Limit: 5,
        ScanIndexForward: false
      });
    });

    const results = await Promise.all(
      queries.map(async (query) => {
        return await dynamodb.send(query);
      })
    );

    // ⑥視聴者数を Timed Metadata として IVS に送信
    results.forEach(result => {
      if (result.Items) {
        const input = {
          channelArn: result.Items[0].channel.S,
          metadata: result.Items.map(item => item.count.N).join(',')
        };

        const putMetadataCommand = new PutMetadataCommand(input);

        ivs.send(putMetadataCommand);
      }
    });

    response = {
      statusCode: 200,
      body: JSON.stringify(counts),
    };
  } catch(err) {
    console.log(err);
    return err;
  };

  return response;
};

• ①で、リージョン全体のチャンネルの ARN を取得しています。

• ②で、実際に開始されたストリーミングの視聴者数を取得しています。

• ③で、視聴者数を DynamoDB に保存するために整形しています。

• ④で、 DynamoDB に視聴者数を保存しています。

• ⑤で、保存された過去 5 分間の視聴者数を DynamoDB から取得しています。

• ⑥で、⑤で取得した過去 5 分間の視聴者数をカンマ (,) 区切りの文字列データに変換し、 Timed Metadata として IVS に送信しています。

前準備が完了したので、いよいよ実際にリソースをデプロイしていきます。

下記コマンドを実行し、プロジェクトのフォルダに移動し、パッケージをインストールします。

cd ivs-viewers-count-cdk
npm i

下記コマンドを実行し、CDK デプロイ用の Amazon S3 バケットと AWS CloudFormation スタックを作成します。

cdk bootstrap

下記コマンドを実行し、リソースをデプロイします。

cdk deploy

下記の通り質問されるので、 y を入力して「Enter」を押下してください。

Do you wish to deploy these changes (y/n)? y

下記の通り表示されれば、リソースのデプロイ成功です !

✅  IvsViewersCountCdkStack

今回の内容は「5. 各リソースのデプロイ」までで完了ですが、せっかくなので以前の 「Amazon Interactive Video Service (IVS) と AWS Amplify を使って自分だけのオリジナル配信サイトを作る !」 の記事で紹介したサイトに視聴者数を表示してみましょう。

まずは 以前の記事の 「3. Nuxt.js プロジェクトのセットアップ」までの手順通り、配信サイトを構築しましょう。

次に、以前の記事 のソースコードを、下記のブランチに切り替え、アプリケーションを再起動します。

# ctrl + c でアプリケーションを停止してから
git checkout feature/add-viewers-count
npm run dev

しばらく (EventBridge の関数呼び出しサイクルの関係上、最大 1 分) 待ってから、下記の通り視聴者数が表示されれば成功です !

せっかく動いたところで名残惜しいですが、リソースの削除方法も紹介します。

7-1. 今回の機能のリソースの削除

下記のコマンドを実行してください。

cdk destroy

下記の通り質問されるので、 y を入力して「Enter」を押下してください。

Are you sure you want to delete: IvsViewersCountCdkStack (y/n)? y

下記の通り表示されれば、削除成功です。

✅  IvsViewersCountCdkStack: destroyed

7-2. Cloud9 環境の削除

Cloud 9 環境の左上の雲の中に 9 のマークをクリックし、 「Go To Your Dashboard」をクリックします。

「Delete」と入力し、 「Delete」ボタンをクリックします。

これで、 Cloud 9 環境も削除されました。