API Gateway + WebSocketでさくっとお絵かきチャットを作る

この記事はモバイルファクトリー Advent Calendar 2020 17日目の記事です。

---

こんにちは、エンジニアのshioiyanです。

モバイルファクトリーには部活動制度があり、いくつもの部活動が存在しているのですが、自分はそのうちのゲームジャム部に所属しています。

今年2月から弊社はリモートワークになりましたが、ゲームジャム部はビデオ通話を使って活動を継続しています。

近頃、外出自粛している人が増えた中でも、ビデオ通話で話しながら楽しく遊べるサービスを作ろう！ということで部活を通じて、Web上でリアルタイムにそれぞれの画面が同期するお絵かきチャットの開発をしました。

仕様

今回作るリアルタイムお絵かきチャットの仕様はざっくり以下のようになります。

1. ユーザは部屋を選んで入室ができる
2. 部屋にはマウスやタップ操作で絵を描くことのできるキャンバスがある
3. 絵を描くと同じ部屋のメンバーのキャンバスがリアルタイムで更新される
4. 部屋を退室することができる

技術選定の背景

各クライアントのお絵かき画面をリアルタイムで同期するためにはWebSocket APIを用いることが思い浮かびました。

しかし、WebSocketサーバの構築・管理にはコストがかかるため、「動いて触れるものを素早く作りたい」という方針だったゲームジャム部で作るには少しネックでした。

調べていく中でAPI Gateway の WebSocket APIを用いればサーバレスでさくっと構築できてサーバの管理要らずで良さそう、ということがわかってきたので、公式で紹介されていたチャットの実装を参考にものは試しと作ってみました。

すると思った以上にさくっと要件を叶える実装をすることができたので、その知見を共有しようと思います。

利用した技術スタック

• リアルタイム通信: Amazon API Gateway WebSocket API
• バックエンド: AWS Lambda
• 接続/部屋情報の保持: Amazon DynamoDB
• フロントエンド: Nuxt.js (v2.14.11)

構成

まずはじめに今回実装したものの全体の構成を示しておきます。

クライアントはAPI GatewayとWebSocketで通信し、通信内容に応じて3種類のLambda関数を実行します。

DynamoDBはLambda関数を通じて接続しているクライアント情報の参照や更新を行います。

実装されたもの(canvasの同期のみ)

(左のブラウザのcanvasに描いた絵が右のブラウザにも同期されています)

実装

▶︎ フォルダ構成

$ tree -I node_modules -L 3
.
├── webSocket
│   ├── onConnect
│   │   └── app.js
│   ├── onDisconnect
│   │   └── app.js
│   ├── package.json
│   ├── sendMessage
│   │   └── app.js
│   ├── serverless.yml
│   └── yarn.lock
└── view
    ├── README.md
    ├── assets
    │   └── README.md
    ├── commitlint.config.js
    ├── components
    │   └── README.md
    ├── layouts
    │   ├── README.md
    │   └── default.vue
    ├── middleware
    │   └── README.md
    ├── nuxt.config.js
    ├── package.json
    ├── pages
    │   ├── README.md
    │   ├── room.vue
    │   └── index.vue
    ├── plugins
    │   └── README.md
    ├── static
    │   ├── README.md
    │   └── favicon.ico
    ├── store
    │   └── README.md
    ├── stylelint.config.js
    ├── utils
    │   └── webSocket.js
    └── yarn.lock

Serverless Framework

API Gateway WebSocket API + Lambda + DynamoDBの構成はServerless Frameworkで作成します。

コマンド1つで各種リソースの生成・更新・削除ができるのはとても便利です。

デプロイ
$ sls deploy -v --stage dev
...
Service Information
service: advent-calendar-2020
stage: dev
region: ap-northeast-1
stack: advent-calendar-2020-dev
resources: 25
api keys:
  None
endpoints:
  wss://xxxxx.execute-api.ap-northeast-1.amazonaws.com/dev
functions:
  connectHandler: advent-calendar-2020-dev-connectHandler
  disconnectHandler: advent-calendar-2020-dev-disconnectHandler
  sendMessageHandler: advent-calendar-2020-dev-sendMessageHandler
layers:
  None

Stack Outputs
ConnectHandlerLambdaFunctionQualifiedArn: arn:aws:lambda:ap-northeast-1:xxxxx:function:advent-calendar-2020-dev-connectHandler:10
DisconnectHandlerLambdaFunctionQualifiedArn: arn:aws:lambda:ap-northeast-1:xxxxx:function:advent-calendar-2020-dev-disconnectHandler:10
ServiceEndpointWebsocket: wss://xxxxx.execute-api.ap-northeast-1.amazonaws.com/dev
ServerlessDeploymentBucketName: advent-calendar-2020-dev-serverlessdeploymentbuck-xxxxx
SendMessageHandlerLambdaFunctionQualifiedArn: arn:aws:lambda:ap-northeast-1:xxxxx:function:advent-calendar-2020-dev-sendMessageHandler:10

✨  Done in 18.34s.

リソース一括削除
$ sls remove --stage dev

▶︎ webSocket/serverless.yml

service: advent-calendar-2020

provider:
  name: aws
  stage: ${opt:stage, 'dev'}
  region: ${opt:region, 'ap-northeast-1'}
  runtime: nodejs12.x
  iamRoleStatements:
    - Effect: Allow
      Action:
        - dynamodb:Query
        - dynamodb:GetItem
        - dynamodb:PutItem
        - dynamodb:DeleteItem
      Resource:
          - Fn::GetAtt: [ ConnectionsTable, Arn ]
    - Effect: Allow
      Action:
        - dynamodb:Query
      Resource: "arn:aws:dynamodb:${self:provider.region}:*:table/${self:service}-connections-${self:provider.stage}/index/*"
  environment:
    TABLE_NAME:
      Ref: ConnectionsTable
  websocketsApiName: ${self:service}-${self:provider.stage}
  websocketsApiRouteSelectionExpression: $request.body.message

functions:
  connectHandler:
    handler: onConnect/app.handler
    events:
      - websocket: $connect
  disconnectHandler:
    handler: onDisconnect/app.handler
    events:
      - websocket: $disconnect
  sendMessageHandler:
    handler: sendMessage/app.handler
    events:
      - websocket: sendMessage

resources:
  Resources:
    ConnectionsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: ${self:service}-connections-${self:provider.stage}
        AttributeDefinitions:
          - AttributeName: connectionId
            AttributeType: S
          - AttributeName: roomId
            AttributeType: S
        KeySchema:
          - AttributeName: connectionId
            KeyType: HASH
        ProvisionedThroughput:
          ReadCapacityUnits: 1
          WriteCapacityUnits: 1
        GlobalSecondaryIndexes:
          - IndexName: roomId-index
            KeySchema:
              - AttributeName: roomId
                KeyType: HASH
            ProvisionedThroughput:
              ReadCapacityUnits: 1
              WriteCapacityUnits: 1
            Projection:
              ProjectionType: ALL
        SSESpecification:
          SSEEnabled: False

DynamoDB

WebSocketのconnectionIdとその接続しているクライアントが今入室している部屋情報を保持するために、connectionIdとroomIdのカラムを作成しています。

また、同じ部屋に入っているメンバーのレコードを取得するためにroomIdを使用してクエリを投げたいのですが、パーテションキー(connectionId)の指定をせずにクエリを投げることはできません。（パーテションキーを指定せずにクエリを投げるとValidationException: Query condition missed key schema elementエラーになる）

そこでroomIdにグローバルセカンダリインデックスを貼って検索できるようにしています。

SSESpecification.SSEEnabledはDynamoDBに保存されたデータの暗号化を有効にするかの設定です。

有効にすると料金がかかるので、開発時には無効にしておくと良いでしょう。

API Gateway WebSocket API

API GatewayのWebSocket APIではserverless.ymlのwebsocketsApiRouteSelectionExpressionで指定された値(routeKey)がクライアントから渡されると、それに応じたルートと呼ばれるリソースタイプで処理が実行されます。

今回の実装だと、$request.body.messageの値によってルートが決定され、$request.body.messageがsendMessageだとsendMessageHandlerの関数が実行されることになります。

ただし、API Gatewayで最初からルートに使用できる3つの特別なrouteKey値が存在します。

• $connect: クライアントがWebSocket APIに最初に接続するときに使用される
  • 接続開始したときに実行したい処理にルーティングできる
  • 今回の実装だとconnectHandlerが実行される
• $disconnect: クライアントがWebSocket APIから切断するときに使用される
  • 接続を切断したときに実行したい処理にルーティングできる
  • 今回の実装だとdisconnectHandlerが実行される
• $default: websocketsApiRouteSelectionExpressionの値が他のrouteKeyに一致しない場合に使用される
  • 今回は使用しない(sendMessage以外の$request.body.messageは考慮しない)

Lambda関数

今回実装している3つの関数の大枠はこちらのドキュメントを参考にしています。

▶︎ webSocket/onConnect/app.js

const AWS = require('aws-sdk')
AWS.config.update({ region: process.env.AWS_REGION })
const DDB = new AWS.DynamoDB({ apiVersion: '2012-10-08' })

exports.handler = function (event, context, callback) {
  let roomId = ''
  if (event.queryStringParameters && event.queryStringParameters.roomId) {
    roomId = event.queryStringParameters.roomId
  }

  const putParams = {
    TableName: process.env.TABLE_NAME,
    Item: {
      connectionId: { S: event.requestContext.connectionId },
      roomId: { S: roomId }
    }
  }

  DDB.putItem(putParams, function (err) {
    callback(null, {
      statusCode: err ? 500 : 200,
      body: err ? 'Failed to connect: ' + JSON.stringify(err) : 'Connected.'
    })
  })
}

接続時にQuery ParameterでroomIdを渡してconnectionIdと共にDynamoDBに保持します。

これによってWebSocketの接続状態をDynamoDBに保持しつつ、接続しているクライアントがどの部屋に入っているかも参照できるようになります。

▶︎ webSocket/sendMessage/app.js

const AWS = require('aws-sdk')

const DDB = new AWS.DynamoDB.DocumentClient({ apiVersion: '2012-08-10' })

const { TABLE_NAME } = process.env

exports.handler = async (event, context) => {
  const roomId = JSON.parse(event.body).roomId

  // 自分が参加しているルームの参加者のレコードを取得
  const queryParams = {
    TableName: TABLE_NAME,
    KeyConditionExpression: "#ROOMID = :ROOMID",
    ExpressionAttributeNames: { "#ROOMID": "roomId" },
    ExpressionAttributeValues: { ":ROOMID": roomId },
    IndexName: 'roomId-index'
  }
  const connectionData = await DDB.query(queryParams).promise()

  const apigwManagementApi = new AWS.ApiGatewayManagementApi({
    apiVersion: '2018-11-29',
    endpoint: event.requestContext.domainName + '/' + event.requestContext.stage
  })

  const postData = JSON.parse(event.body).data
  const myConnectionId = event.requestContext.connectionId

  const postCalls = connectionData.Items.map(async ({ connectionId }) => {
    try {
      // 送信者には送らない
      if (myConnectionId !== connectionId) {
        await apigwManagementApi.postToConnection({ ConnectionId: connectionId, Data: postData }).promise()
      }
    } catch (e) {
      if (e.statusCode === 410) {
        console.log(`Found stale connection, deleting ${connectionId}`)
        await DDB.delete({ TableName: TABLE_NAME, Key: { connectionId } }).promise()
      } else {
        throw e
      }
    }
  })

  try {
    await Promise.all(postCalls)
  } catch (e) {
    return { statusCode: 500, body: e.stack }
  }

  return { statusCode: 200, body: 'Data sent.' }
}

グローバルセカンダリインデックスを貼ったroomIdで同じ部屋のクライアントのレコードを取得して、それらのconnectionIdに対してpostToConnectionでdataを送信します。

クライアントから送信されたdataを同じ部屋のクライアント全員に送信しています。

▶︎ webSocket/onDisconnect/app.js

const AWS = require('aws-sdk')
AWS.config.update({ region: process.env.AWS_REGION })
const DDB = new AWS.DynamoDB({ apiVersion: '2012-10-08' })

exports.handler = function (event, context, callback) {
  const deleteParams = {
    TableName: process.env.TABLE_NAME,
    Key: {
      connectionId: { S: event.requestContext.connectionId }
    }
  }

  DDB.deleteItem(deleteParams, function (err) {
    callback(null, {
      statusCode: err ? 500 : 200,
      body: err ? 'Failed to disconnect: ' + JSON.stringify(err) : 'Disconnected.'
    })
  })
}

切断時にはDynamoDBから切断したクライアントのレコードを削除します。

クライアント

WebSocketの接続にはJavaScriptのWebSocketのwrapperライブラリのSocketteを使用しています。

Socketteを使用することで再接続処理や、WebSocketの各種EventListenerで実行される関数が簡単に指定できます。

▶︎ view/utils/webSocket.js

import Sockette from 'sockette'

export function newConnection({ roomId, onReceivedMessage }) {
  return new Sockette(
    `wss://xxxxx.execute-api.ap-northeast-1.amazonaws.com/dev?roomId=${roomId}`,
    {
      timeout: 5e3,
      maxAttempts: 3,
      onmessage: (e) => onReceivedMessage(e),
      onerror: (e) => console.error(e),
    }
  )
}

vueコンポーネントの実装は、実際に絵を描く部分は割愛しますが以下のようになります。

▶︎ view/pages/room.vue

<template>
  <div>
    <canvas ref="canvas"></canvas>
  </div>
</template>

<script>
import { newConnection } from '~/utils/webSocket'

export default {
  data() {
    return {
      ws: null,
      roomId: 'roomA', // XXX: 実際は動的にする
    }
  },
  mounted() {
    this.connectWs()
  },
  beforeDestroy() {
    this.disconnectWs()
  },
  methods: {
    connectWs() {
      this.ws = newConnection({
        roomId: this.roomId,
        onReceivedMessage: this.onReceivedMessage,
      })
    },
    disconnectWs() {
      if (this.ws !== null) {
        this.ws.close()
      }
    },
    onReceivedMessage(event) { // WebSocketでメッセージを受信したときに実行される
      const data = JSON.parse(event.data)
      switch (data.actionType) { // actionTypeによって処理を分岐。画面クリア、1つ戻る/進むといったeventを増やしたりする
        case 'DRAW':
          this.draw(data.positions)
          break
        default:
          break
      }
    },
    draw({ fromX, fromY, toX, toY }) {
      // canvasにfromの座標からtoの座標に線を引く
      ...
    },
    sendDrawMessage(positions) {
      this.sendMessage({
        data: JSON.stringify({
          actionType: 'DRAW',
          positions,
        }),
      })
    },
    sendMessage({ data }) {
      if (this.ws !== null && this.roomId) {
        this.ws.json({ // WebSocketを介してobjを送信
          message: 'sendMessage',
          data,
          roomId: this.roomId,
        })
      }
    },
    // タップ&ドラッグ時にthis.sendDrawMessage(positions)や自分のcanvasに対してthis.draw(positions)を行う処理
    ...
  },
}
</script>

...

ページ表示後にWebSocketの接続を行い、接続した状態でcanvas上でタップ&ドラッグ操作をするとその座標をWebSocketを介して同じ部屋のクライアントに操作した内容を送信してcanvasの同期を行います。

送信するobjにactionTypeという値を持たせていますが、これは受信したメッセージの内容を識別するためのものです。

この値を変えることで、別のデータのやりとりとそれに応じた処理の分岐も簡単に行えます。

この記事では実装していませんが、例えばキャンバスクリアや描いた絵を1つ戻す(undo)/進める(redo)といったイベントの同期をできるようにすると、よりお絵かきチャットっぽくなるでしょう。

まとめ

API Gateway + WebSocket APIでお絵かきチャットを作ることができました。

今回はcanvasの同期に利用しましたが、様々なリアルタイム通信が必要な場面で便利に使っていけそうな機能だと感じました。

明日の記事は id:pikkaman さんです！