Momento でリアルタイムチャットアプリを構築する手順

リアルタイムアプリケーションはますます普及し、多くの人が WhatsApp など、一つ以上のリアルタイムチャットアプリを使っています。Momento の SDK ドキュメントも参考にしながら、その構築方法を見ていきましょう。

こうしたアプリケーションの大きな課題の一つが、ユーザーの接続を効率よく管理することです。多数のユーザーが接続していても、その接続の大半はほとんどの時間、何も送受信していません。

ところが、WebSockets に対応する多くのサービス、たとえば API Gateway、AppSync、IoT Core は、接続時間に対して課金します。ユーザーが実際にメッセージを送受信していなくても、接続を維持するだけで料金が発生します。

アプリケーションの規模が大きくなると、この方式は非常に非効率になり得ます。そこで役立つのが Momento Topics [1] です。

Momento Topics は WebSockets の複雑な部分を抽象化し、実際に使った分だけを支払うオンデマンド料金を提供します。

このガイドでは、Momento Topics [1] を使ってリアルタイムチャットアプリを構築する方法を紹介します。

デモのコードはすべてこちらのリポジトリ [2] で公開しています。

前提条件

このガイドを進めるには、以下が必要です。

Momento アカウント。console.gomomento.com [3] で無料アカウントを作成できます。
AWS アカウント。
Node.js。
CDK v2。

Momento Topics についての注意点

Momento Topics の重要な特徴は、トピックが仮想的であることです。使用する前に、明示的にトピックを作成する必要はありません。

以下の手順を進める際は、この点を覚えておいてください。

使い捨てトークンを理解する

Momento には、認証と認可のために API キーと使い捨てトークンという二つの仕組みがあります。

バックエンドの処理には API キーを使います。

フロントエンドでは、Momento SDK [4] を使って使い捨てトークンを生成できます。このトークンの権限は、特定のリソースに対する特定の操作に限定できます。

たとえば、特定のトピックへの配信と購読を許可するトークンを発行できます。チャットルームごとにトピックを用意すれば、使い捨てトークンでユーザーがアクセスできるルームを制御できます。

アーキテクチャの概要

このデモでは、次の動作をするフルスタックアプリケーションを構築します。

ログインしたユーザーは、利用可能なチャットルームの一覧を確認できます。
ユーザーは新しいチャットルームを作成するか、既存のルームに参加できます。
チャットルームの一覧は DynamoDB に保存します。
各チャットルームには、対応する Momento のトピックがあります。
ユーザーがルームに参加すると、フロントエンドがそのルームのトピックを購読します。
ユーザーがルームにメッセージを送ると、トピックに配信され、そのルームにいるすべてのユーザーへ届きます。

API には三つのルートがあります。

POST /chats: 新しいチャットルームを作成します。
GET /chats: 利用可能なチャットルームを一覧表示します。
GET /token: フロントエンドが Momento のトピックを購読するための使い捨てトークンを生成します。

全体のアーキテクチャは次のとおりです。

手順 1: 新しいキャッシュを作成する

Momento Topics は Momento Cache と同じインフラを使います。そのため、少し不思議に思えるかもしれませんが、トピックを使うには最初にキャッシュを作成します。

Momento Console にログインし、「Caches」を開きます。

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

新しいキャッシュの名前を「chat」とし、「us-east-1」リージョンに作成します。

Momento CLI [5] でも同じ操作ができます。

手順 2: API キーを生成する

Momento には二種類の API キーがあります。

Super User Key: キャッシュやトピックの管理、および使い捨てトークンの生成に使います。
Fine-Grained Access Key: Momento のキャッシュやトピックを操作するために使います。

使い捨てトークンの生成には Super User Key が必要です。今回はこのキーを作成します。

Momento Console の「API keys」を開き、「us-east-1」リージョンで新しい「Super User Key」を生成します。

「Generate Api Key」をクリックします。

手順 3: API キーを安全に保管する

API キーを安全に保管するため、SSM Parameter Store に保存します。

AWS の「AWS Systems Manager」コンソールを開き、「Parameter Store」をクリックします。

「/chat-api/dev/momento-api-key」という名前のパラメーターを作成します。これは、私が普段 SSM パラメーターに使っている命名規則 /{service-name}/{environment}/{parameter-name} に従っています。

パラメーターの種類は、必ず「SecureString」にします。これにより、API キーが保存時に暗号化されます。

手順 4: CDK アプリを作成する

このデモでは CDK を使いますが、ほかの Infrastructure as Code ツールでも同じように構築できます。

CDK アプリには、次の要件を持たせます。

サーバーレス開発で特に効果の高い実践方法の一つである、一時的な環境 [6] に対応すること。
一時的な環境で、dev などの主要な環境の SSM パラメーターを再利用 [7] できること。

そこで、stageName と ssmStageName という二つのコンテキスト変数を受け取ります。

stageName は、名前の衝突を避けるため、作成するすべての AWS リソースの名前に含めます。

ssmStageName は、参照するすべての SSM パラメーターで stageName の代わりに使います。

これを踏まえた CDK アプリが、次のコードです。

#!/usr/bin/env node

const cdk = require('aws-cdk-lib');
const { ChatApiStack } = require('./constructs/chat-api-stack');

const app = new cdk.App();

let stageName = app.node.tryGetContext('stageName');
let ssmStageName = app.node.tryGetContext('ssmStageName');

if (!stageName) {
  console.log('Defaulting stage name to dev');
  stageName = 'dev';
}

if (!ssmStageName) {
  console.log(`Defaulting SSM stage name to "stageName": ${stageName}`);
  ssmStageName = stageName;
}

const serviceName = 'chat-api';

new ChatApiStack(app, `ChatApiStack-${stageName}`, {
  serviceName,
  stageName,
  ssmStageName,
});

ChatApiStack は次のとおりです。

const { Stack, CfnOutput } = require('aws-cdk-lib');
const { Runtime } = require('aws-cdk-lib/aws-lambda');
const { NodejsFunction } = require('aws-cdk-lib/aws-lambda-nodejs');
const { Table, BillingMode, AttributeType } = require('aws-cdk-lib/aws-dynamodb');
const { RestApi, LambdaIntegration, CfnAuthorizer, AuthorizationType } = require('aws-cdk-lib/aws-apigateway');
const iam = require('aws-cdk-lib/aws-iam');
const { UserPool, UserPoolClient } = require('aws-cdk-lib/aws-cognito');

const MOMENTO_CACHE_NAME = 'chat';

class ChatApiStack extends Stack {
  constructor(scope, id, props) {
    super(scope, id, props);

    const api = new RestApi(this, `${props.stageName}-ChatApi`, {
      deployOptions: {
        stageName: props.stageName,
        tracingEnabled: true
      }
    });

    const userPool = new UserPool(this, 'CognitoUserPool', {
      userPoolName: `${props.serviceName}-${props.stageName}-UserPool`,
      selfSignUpEnabled: true,
      signInAliases: { email: true }
    });

    const webUserPoolClient = new UserPoolClient(this, 'WebUserPoolClient', {
      userPool,
      authFlows: {
        userSrp: true
      },
      preventUserExistenceErrors: true
    });

    new CfnOutput(this, 'UserPoolId', { value: userPool.userPoolId });
    new CfnOutput(this, 'UserPoolClientId', { value: webUserPoolClient.userPoolClientId });

    this.momentoApiKeyParamName = `/${props.serviceName}/${props.ssmStageName}/momento-api-key`;
    this.momentoApiKeyParamArn = `arn:aws:ssm:${this.region}:${this.account}:parameter${this.momentoApiKeyParamName}`;

    this.chatsTable = this.createChatsTable();

    const listChatsFunction = this.createListChatsFunction(props);
    const newChatFunction = this.createNewChatFunction(props);
    const tokenVendingMachineFunction = this.createTokenVendingMachineFunction(props);

    this.createApiEndpoints(api, userPool, {
      listChats: listChatsFunction,
      newChat: newChatFunction,
      tokenVendingMachine: tokenVendingMachineFunction
    });
  }

  createChatsTable() {
    return new Table(this, 'ChatsTable', {
      billingMode: BillingMode.PAY_PER_REQUEST,
      partitionKey: {
        name: 'chatName',
        type: AttributeType.STRING
      }
    });
  }

  createListChatsFunction(props) {
    const func = new NodejsFunction(this, 'ListChatsFunction', {
      runtime: Runtime.NODEJS_20_X,
      handler: 'handler',
      entry: 'functions/list-chats.js',
      memorySize: 1024,
      environment: {
        SERVICE_NAME: props.serviceName,
        STAGE_NAME: props.stageName,
        POWERTOOLS_LOG_LEVEL: props.stageName === 'prod' ? 'INFO' : 'DEBUG',
        CHATS_TABLE_NAME: this.chatsTable.tableName
      }
    });

    this.chatsTable.grantReadData(func);

    return func;
  }

  createNewChatFunction(props) {
    const func = new NodejsFunction(this, 'NewChatFunction', {
      runtime: Runtime.NODEJS_20_X,
      handler: 'handler',
      entry: 'functions/new-chat.js',
      memorySize: 1024,
      environment: {
        SERVICE_NAME: props.serviceName,
        STAGE_NAME: props.stageName,
        POWERTOOLS_LOG_LEVEL: props.stageName === 'prod' ? 'INFO' : 'DEBUG',
        CHATS_TABLE_NAME: this.chatsTable.tableName
      }
    });

    this.chatsTable.grantWriteData(func);

    return func;
  }

  createTokenVendingMachineFunction(props) {
    const func = new NodejsFunction(this, 'TokenVendingMachineFunction', {
      runtime: Runtime.NODEJS_20_X,
      handler: 'handler',
      entry: 'functions/token-vending-machine.js',
      memorySize: 1024,
      environment: {
        SERVICE_NAME: props.serviceName,
        STAGE_NAME: props.stageName,
        MOMENTO_API_KEY_PARAM_NAME: this.momentoApiKeyParamName,
        MOMENTO_CACHE_NAME,
        POWERTOOLS_LOG_LEVEL: props.stageName === 'prod' ? 'INFO' : 'DEBUG'
      }
    });

    func.role.attachInlinePolicy(new iam.Policy(this, 'TokenVendingMachineFunctionSsmPolicy', {
      statements: [
        new iam.PolicyStatement({
          effect: iam.Effect.ALLOW,
          actions: [ 'ssm:GetParameter*' ],
          resources: [ this.momentoApiKeyParamArn ]
        })
      ]
    }));

    return func;
  }

  /**
   *
   * @param {RestApi} api
   * @param {UserPool} userPool
   */
  createApiEndpoints(api, userPool, functions) {
    const authorizer = new CfnAuthorizer(this, 'CognitoAuthorizer', {
      name: 'CognitoAuthorizer',
      type: 'COGNITO_USER_POOLS',
      identitySource: 'method.request.header.Authorization',
      providerArns: [userPool.userPoolArn],
      restApiId: api.restApiId,
    });

    const chatsResource = api.root.addResource('chats');

    // POST /chats
    chatsResource.addMethod('POST', new LambdaIntegration(functions.newChat), {
      authorizer: {
        authorizationType: AuthorizationType.COGNITO,
        authorizerId: authorizer.ref
      }
    });

    // GET /chats
    chatsResource.addMethod('GET', new LambdaIntegration(functions.listChats), {
      authorizer: {
        authorizationType: AuthorizationType.COGNITO,
        authorizerId: authorizer.ref
      }
    });

    chatsResource.addCorsPreflight({
      allowHeaders: ['*'],
      allowMethods: ['OPTIONS', 'GET', 'POST'],
      allowCredentials: true,
      allowOrigins: ['*']
    });

    // GET /token
    const tokenResource = api.root.addResource('token');
    tokenResource.addMethod('GET', new LambdaIntegration(functions.tokenVendingMachine), {
      authorizer: {
        authorizationType: AuthorizationType.COGNITO,
        authorizerId: authorizer.ref
      }
    });

    tokenResource.addCorsPreflight({
      allowHeaders: ['*'],
      allowMethods: ['OPTIONS', 'POST'],
      allowCredentials: true,
      allowOrigins: ['*']
    });
  }
}

module.exports = { ChatApiStack }

ここでは API Gateway に API を作成し、先ほどのルートを実装する三つの Lambda 関数を用意しています。

TokenVendingMachineFunction
NewChatFunction
ListChatsFunction

認証と認可

API は Cognito オーソライザーで保護します。

const authorizer = new CfnAuthorizer(this, 'CognitoAuthorizer', {
  name: 'CognitoAuthorizer',
  type: 'COGNITO_USER_POOLS',
  identitySource: 'method.request.header.Authorization',
  providerArns: [userPool.userPoolArn],
  restApiId: api.restApiId,
});

SSM パラメーターを安全に読み込む

関数の環境変数には Momento API キーを含めていない点に注目してください。代わりに、パラメーターの名前を渡しています。

environment: {
  SERVICE_NAME: props.serviceName,
  STAGE_NAME: props.stageName,
  MOMENTO_API_KEY_PARAM_NAME: this.momentoApiKeyParamName,
  MOMENTO_CACHE_NAME,
  POWERTOOLS_LOG_LEVEL: props.stageName === 'prod' ? 'INFO' : 'DEBUG'
}

そして、実行時にパラメーターを取得するための IAM 権限を関数に与えます。

func.role.attachInlinePolicy(new iam.Policy(this, 'TokenVendingMachineFunctionSsmPolicy', {
  statements: [
    new iam.PolicyStatement({
      effect: iam.Effect.ALLOW,
      actions: [ 'ssm:GetParameter*' ],
      resources: [ this.momentoApiKeyParamArn ]
    })
  ]
}));

このようにする理由は、次の二つです。

侵害された依存ライブラリが環境変数から情報を盗むリスクを抑えるため。
API キーの有効期間を短くし、アプリケーションを再デプロイせずにローテーションできるようにするため。

コールドスタート時に、関数は SSM パラメーターを取得して復号し、その値を数分間キャッシュします。キャッシュの有効期限が切れると、次の呼び出し時に SSM Parameter Store から更新された値の取得を試みます。

これにより、呼び出しのたびに SSM にアクセスする必要がなくなります。バックグラウンドの cron ジョブで API キーをローテーションすると、関数はキャッシュの期限切れ後、自動的に新しいキーを取得します。

Middy の ssm ミドルウェア [8] は、この処理に標準で対応しています。詳しくは後述しますが、複雑な処理はミドルウェアに任せられます。

ワークフローの例

JIRA チケット「ABP-1734」の作業を始める場合、私は次のように進めます。

機能ブランチ ABP-1734 を作成します。
cdk deploy --context stageName=FEAT-ABP-1734 --context ssmStageName=dev を実行して、一時的な環境を作成します。チャット API の新しいインスタンスが作られ、ほかと切り離して変更を進められます。この環境は dev の SSM パラメーターを使いますが、すべてのリソース名に FEAT-ABP-1734 のサフィックスが付きます。
変更を加えてテストし、PR を作成します。
cdk destroy --context stageName=FEAT-ABP-1734 --context ssmStageName=dev を実行して、一時的な環境を削除します。

このような短期間だけ存在する環境は、機能開発だけでなく、CI/CD パイプラインでのテストにも役立ちます。従量課金により、環境を増やしたこと自体で追加料金が発生することなく、必要な数の環境を用意できます。

理想的には、Momento のキャッシュも環境ごとに一つ用意します。その場合、キャッシュ名には stageName をプレフィックスまたはサフィックスとして付けます。

手順 5: TokenVendingMachine 関数を実装する

GET /token ルートの背後にある TokenVendingMachine 関数は、次のコードです。

const { initAuthClient, generateToken } = require('../lib/momento');
const middy = require('@middy/core');
const cors = require('@middy/http-cors');
const ssm = require('@middy/ssm');

module.exports.handler = middy(async (event, context) => {
  await initAuthClient(context.MOMENTO_API_KEY);

  const tokenResult = await generateToken();

  return {
    statusCode: 200,
    body: JSON.stringify(tokenResult)
  }
})
.use(cors())
.use(ssm({
  cache: true,
  cacheExpiry: 5 * 60 * 1000,
  setToContext: true,
  fetchData: {
    MOMENTO_API_KEY: process.env.MOMENTO_API_KEY_PARAM_NAME
  }
}));

ここでは、Middy の ssm ミドルウェア [8] を使って、SSM Parameter Store から Momento API キーを取得し、キャッシュしています。

.use(ssm({
  cache: true,
  cacheExpiry: 5 * 60 * 1000,
  setToContext: true,
  fetchData: {
    MOMENTO_API_KEY: process.env.MOMENTO_API_KEY_PARAM_NAME
  }
}));

このミドルウェアは、デフォルトでは取得したデータを環境変数に挿入します。しかし、先ほど説明したとおり、暗号化されていない API キーを Lambda 関数の環境変数に置くことは避けるべきです。攻撃者は機密情報を探すために、環境変数を調べることがよくあります。

そこで、取得したデータを Lambda の呼び出し時の context オブジェクトに設定するように指定します。そのため、Momento クライアントの初期化時には、context.MOMENTO_API_KEY から API キーを取得します。

await initAuthClient(context.MOMENTO_API_KEY);

共通処理をまとめる

先ほどのコードのとおり、Momento に関連する操作はすべて、共通の momento.js モジュールにまとめています。

Momento 認証クライアントの初期化などの共通処理も、ここに含まれます。

const {
  CredentialProvider,
  AuthClient,
  DisposableTokenScopes,
  ExpiresIn,
  GenerateDisposableTokenResponse,
  AllTopics,
} = require('@gomomento/sdk');

const { Logger } = require('@aws-lambda-powertools/logger');
const logger = new Logger({ serviceName: 'leaderboard-api' });

const { MOMENTO_CACHE_NAME } = global.process.env;

let authClient;

async function initAuthClient(apiKey) {
  if (!authClient) {
    logger.info('Initializing Momento auth client');

    authClient = new AuthClient({
      credentialProvider: CredentialProvider.fromString(apiKey)
    });

    logger.info('Initialized Momento auth client');
  }
};

async function generateToken() {
  const result = await authClient.generateDisposableToken(
    DisposableTokenScopes.topicPublishSubscribe(MOMENTO_CACHE_NAME, AllTopics),
    ExpiresIn.minutes(30)
  );

  return {
    endpoint: result.endpoint,
    token: result.authToken,
    cacheName: MOMENTO_CACHE_NAME,
    expiresAt: result.expiresAt
  };
}

module.exports = {
  initAuthClient,
  generateToken,
};

ここでは、Lambda の実行環境が再利用されることを利用しています。

コールドスタートで新しい実行環境が作られると、authClient 変数が設定されます。同じ実行環境での次回以降の呼び出しでは、initAuthClient 関数は初期化を省略し、すぐに処理を返します。

きめ細かな認可

次の行に注目してください。

const result = await authClient.generateDisposableToken(
  DisposableTokenScopes.topicPublishSubscribe(MOMENTO_CACHE_NAME, AllTopics),
  ExpiresIn.minutes(30)
);

ここでは、「chat」キャッシュ内のすべてのトピックに対して、配信と購読ができる短寿命のトークンを生成しています。これは広い権限であり、ユーザーはどのチャットルームにも参加してメッセージを送れます。

今回のデモはすべてのルームが公開なので問題ありません。ただし、実際のアプリケーションには非公開のルームもあるでしょう。その場合は、ユーザーの権限に応じて DisposableTokenScopes をトピックの一部に限定する必要があります。

手順 6: ListChats 関数を実装する

GET /chats ルートの背後にある ListChats 関数は、次のコードです。

const { DynamoDB } = require("@aws-sdk/client-dynamodb");
const { DynamoDBDocumentClient, ScanCommand } = require("@aws-sdk/lib-dynamodb");
const dynamodbClient = new DynamoDB();
const dynamodb = DynamoDBDocumentClient.from(dynamodbClient);
const middy = require('@middy/core');
const cors = require('@middy/http-cors');

module.exports.handler = middy(async () => {
  const { Items } = await dynamodb.send(new ScanCommand({
    TableName: process.env.CHATS_TABLE_NAME
  }))

  const chats = Items.map(x => ({ chatName: x.chatName }))
  return {
    statusCode: 200,
    body: JSON.stringify(chats)
  }
})
.use(cors());

この関数は DynamoDB の ChatsTable テーブルをスキャンし、利用可能なチャットルームをすべて取得します。

手順 7: NewChat 関数を実装する

POST /chats ルートの背後にある NewChat 関数は、次のコードです。

const { DynamoDB } = require("@aws-sdk/client-dynamodb");
const { DynamoDBDocumentClient, PutCommand } = require("@aws-sdk/lib-dynamodb");
const dynamodbClient = new DynamoDB();
const dynamodb = DynamoDBDocumentClient.from(dynamodbClient);
const middy = require('@middy/core');
const cors = require('@middy/http-cors');

module.exports.handler = middy(async (event) => {
  const { chatName } = JSON.parse(event.body);
  const chat = { chatName };

  await dynamodb.send(new PutCommand({
    TableName: process.env.CHATS_TABLE_NAME,
    Item: chat,
    ConditionExpression: "attribute_not_exists(chatName)"
  }));

  return {
    statusCode: 201,
    body: JSON.stringify(chat)
  };
})
.use(cors());

説明を簡潔にするため、DynamoDB の ChatsTable はシンプルな構成にしています。

テーブルを作成する CDK コードを、もう一度確認しましょう。

createChatsTable() {
  return new Table(this, 'ChatsTable', {
    billingMode: BillingMode.PAY_PER_REQUEST,
    partitionKey: {
      name: 'chatName',
      type: AttributeType.STRING
    }
  });
}

テーブルのハッシュキーは chatName です。そのため、同じ名前のチャットルームを重複して保存することはできません。

一般的には、作成日時や作成者など、チャットルームに関するメタデータも保存します。

このデモではそのメタデータを省略しています。ただし、実際の用途を少し考慮し、同じ名前のチャットルームがすでにある場合に PutCommand が上書きしないようにします。そのために ConditionExpression を指定しています。

await dynamodb.send(new PutCommand({
  TableName: process.env.CHATS_TABLE_NAME,
  Item: chat,
  ConditionExpression: "attribute_not_exists(chatName)"
}));

手順 8: フロントエンドでトピックを購読する

フロントエンドのコードの大半はリアルタイムメッセージングとは直接関係しないため、ここではすべてを詳しく説明しません。コード全体は、デモリポジトリ [2] の frontend ディレクトリにあります。

重要な部分を見ていきます。

ユーザーがサインインした後、次の処理を行います。

1. ユーザーの ID トークンを使って GET /token エンドポイントを呼び出し、使い捨てトークンを取得します。

2. GET /chats エンドポイントを呼び出し、利用可能なチャットルームを一覧表示します。

3. ユーザーがチャットルームをクリックすると、以下の joinChat 関数を使って、そのルームのトピックを購読します。

import { subscribeToTopic } from '@/lib/momento'

let subscription = null

const chatMessages = ref([])
const isConnected = ref(false)

...

const joinChat = async (chatName) => {
  if (subscription) {
    console.log('Unsubscribing from existing chat...')
    await subscription.unsubscribe()

    isConnected.value = false
    chatMessages.value = []
  }

  console.log('Subscribing to chat:', chatName)
  subscription = await subscribeToTopic(
    momentoToken,
    momentoCacheName,
    chatName,
    (jsonMsg) => {
      const { sender, message } = JSON.parse(jsonMsg)
      chatMessages.value.push({ sender, message })
    })

  isConnected.value = true
}

ここで、二つの点に注意してください。

まず、一度に接続するルームを一つに保つため、必ず現在のトピックの購読を解除してから、新しいトピックを購読します。

if (subscription) {
  console.log('Unsubscribing from existing chat...')
  await subscription.unsubscribe()

  isConnected.value = false
  chatMessages.value = []
}

次に、Momento のトピックを購読するときには、新しいメッセージを受信した際の処理を担う関数を渡します。

subscription = await subscribeToTopic(
  momentoToken,
  momentoCacheName,
  chatName,
  (jsonMsg) => {
    const { sender, message } = JSON.parse(jsonMsg)
    chatMessages.value.push({ sender, message })
  })

ここでは、次の前提を置いています。

メッセージは JSON 文字列である。
JSON オブジェクトには、sender (メールアドレス) と message が含まれる。

そのため、トピックへ送信するメッセージには、必ずこの二つを含めます。

では、subscription オブジェクトは何でしょうか。理解するために、subscribeToTopic ヘルパー関数を見てみましょう。

import {
  TopicClient,
  TopicConfigurations,
  CredentialProvider
} from '@gomomento/sdk-web'

async function subscribeToTopic(authToken, cacheName, topicName, onMessage) {
  console.log('Initializing Momento topic client', authToken)

  const topicClient = new TopicClient({
    configuration: TopicConfigurations.Browser.latest(),
    credentialProvider: CredentialProvider.fromString({
      authToken
    })
  })

  console.log('Initialized Momento topic client')
  console.log('Subscribing to Momento topic:', { cacheName, topicName })

  const resp = await topicClient.subscribe(cacheName, topicName, {
    onItem: (item => onMessage(item.value()))
  })

  if (!resp.isSubscribed) {
    const error = resp.innerException()
    console.error(`Failed to subscribe to Momento topic [${topicName}]: `, error)
    throw error
  }

  return {
    send: async (message) => {
      await topicClient.publish(cacheName, topicName, message)
    },
    unsubscribe: async () => await resp.unsubscribe()
  }
}

最も重要なのは、次の部分です。

const resp = await topicClient.subscribe(cacheName, topicName, {
  onItem: (item => onMessage(item.value()))
})

これがトピックの購読処理です。メッセージを受信すると、先ほどの処理関数が呼ばれます。

一方、トピックへメッセージを送信するには topicClient が必要で、トピックから unsubscribe するには購読時のレスポンスが必要です。

実装の詳細が UI 側へ漏れないように、このヘルパー関数は、メッセージの送信と購読解除を行えるオブジェクトを返します。

return {
  send: async (message) => {
    await topicClient.publish(cacheName, topicName, message)
  },
  unsubscribe: async () => await resp.unsubscribe()
}

これが、先ほど登場した subscription オブジェクトです。

手順 9: トピックへメッセージを送信する

トピックに接続したら、先ほどの subscription オブジェクトを使ってメッセージを送信できます。

const sendMessage = async () => {
  await subscription.send(JSON.stringify({
    sender: currentUserEmail.value,
    message: newMessage.value
  }))

  newMessage.value = ''
}

説明したとおり、送信者のメールアドレスとメッセージを JSON に含める必要があります。購読側の関数が、その形式のデータを受け取る前提になっているためです。

これで完成です。ユーザー同士がリアルタイムでやり取りできるチャットアプリができました。会話はチャットルームごとに分かれ、ユーザーはルームに参加したり、新しいルームを作成したりできます。

異なるブラウザーやシークレットウィンドウで、アプリケーションの複数のインスタンスを起動してみてください。一方のユーザーがルームへ送ったメッセージが、もう一方のウィンドウに即座に表示されることを確認できます。

手順 10: デプロイしてテストする

最後に cdk deploy を実行し、動作を試してください。

実際の動きを確認したい場合は、こちらの講座全編 [9] をご覧ください。

Yan Cui による Momento でリアルタイムアプリを構築する講座の画面

クラウド

セルフホスト型

サポート

プラットフォームエンジニアリング向け

業界別

開発

学ぶ

コミュニティ

クラウド

セルフホスト型

サポート

プラットフォームエンジニアリング向け

業界別

開発

学ぶ

コミュニティ

Momento でリアルタイムチャットアプリを構築する手順

前提条件

Momento Topics についての注意点

使い捨てトークンを理解する

アーキテクチャの概要

手順 1: 新しいキャッシュを作成する

手順 2: API キーを生成する

手順 3: API キーを安全に保管する

手順 4: CDK アプリを作成する

認証と認可

SSM パラメーターを安全に読み込む

ワークフローの例

手順 5: TokenVendingMachine 関数を実装する

共通処理をまとめる

きめ細かな認可

手順 6: ListChats 関数を実装する

手順 7: NewChat 関数を実装する

手順 8: フロントエンドでトピックを購読する

手順 9: トピックへメッセージを送信する

手順 10: デプロイしてテストする

関連リンク

Momento、Amazon EventBridge、Amazon DynamoDBによるリアルタイム気象更新システムの構築

Momento でリアルタイム通知システムを構築する方法: ステップバイステップガイド

リアルタイム・データ処理とは何か？

前提条件

Momento Topics についての注意点

使い捨てトークンを理解する

アーキテクチャの概要

手順 1: 新しいキャッシュを作成する

手順 2: API キーを生成する

手順 3: API キーを安全に保管する

手順 4: CDK アプリを作成する

認証と認可

SSM パラメーターを安全に読み込む

ワークフローの例

手順 5: TokenVendingMachine 関数を実装する

共通処理をまとめる

きめ細かな認可

手順 6: ListChats 関数を実装する

手順 7: NewChat 関数を実装する

手順 8: フロントエンドでトピックを購読する

手順 9: トピックへメッセージを送信する

手順 10: デプロイしてテストする

関連リンク

あわせて読みたい

Momento、Amazon EventBridge、Amazon DynamoDBによるリアルタイム気象更新システムの構築

Momento でリアルタイム通知システムを構築する方法: ステップバイステップガイド

リアルタイム・データ処理とは何か？