Amplify Gen2でリアルタイムサブスクリプションを実装してみた

はじめに

コンサル部の神野です。
Amplify Gen2を使ってリアルタイムイベントのやり取りを実装できると知っていましたが、理解が曖昧だったので実際にやってイメージをつかんでみました。

リアルタイムイベントのデータを取り扱う場合のやり方が公式ドキュメントに記載があり、下記を参考にし、今回の実装を行いました。

https://docs.amplify.aws/react/build-a-backend/data/subscribe-data/

構成するシステム構成図

Amplify、AppSync、DynamoDBを使ってリアルタイムイベントを検知します。
この際にAppSyncをアプリケーションからサブスクライブすることによってデータソースの変更を検知できるようにします。

サブスクリプションについて

公式ドキュメントに下記記載があります。

AWS AppSync では、サブスクリプションを使用して、ライブアプリケーションの更新、プッシュ通知などを実装できます。クライアントが GraphQL サブスクリプションオペレーションを呼び出すと、安全な WebSocket 接続が自動的に確立され、 によって維持されます AWS AppSync。その後、アプリケーションは、アプリケーションの接続とスケーリングの要件 AWS AppSync を継続的に管理しながら、データソースからサブスクライバーにデータをリアルタイムで配信できます。以下のセクションでは、 AWS AppSyncサブスクリプションの仕組みを示します。

まとめるとAppSyncのサブスクリプションは、WebSocketを使用してリアルタイムデータ通信を実現するサービスです。クライアントとサーバー間の接続管理やスケーリングをAWS側が自動で行うため、開発者はリアルタイム通信の実装に集中できるといった仕組みです。自前でリアルタイム通信の仕組みを実装しなくていいのは便利ですね。

ただリアルタイムで検知できるのは直接DynamoDBに更新ではなく、AppSync経由で更新があった場合に限ります。

AppSyncとの連携周りもAmplifyがうまくラップして実装してくれるイメージですね。

実装

今回はReact、UIフレームワークとしてMantineを使って簡単な画面を作りながらAmplifyのリアルタイムサブスクリプションを実装していきます。

使用したライブラリのバージョン一覧

React 環境作成

まずはビルドツールviteを使ってサクッと環境を作成します。

npm create vite@latest
Need to install the following packages:
create-vite@6.0.1
Ok to proceed? (y) y

> npx
> create-vite

✔ Project name: … amplify-realtime
✔ Select a framework: › React
✔ Select a variant: › TypeScript

Scaffolding project in /Users/jinno.yudai/dev/blog-amplify-realtime/amplify-realtime...

Done. Now run:

  cd amplify-realtime
  npm install
  npm run dev

cd amplify-realtime/

npm install

added 181 packages, and audited 182 packages in 30s

43 packages are looking for funding
  run `npm fund` for details

found 0 vulnerabilities

完了したら問題なく起動するか確認します。

npm run dev

起動画面

問題なく起動しましたね！
引き続きAmplifyの環境を構築していきます。

Amplify 環境作成

まずは必要なライブラリをインストールします。

npm install @aws-amplify/backend@latest @aws-amplify/backend-cli@latest @aws-amplify/ui-react@latest

インストールが完了したら、次に下記ディレクトリにAmplify用のリソースを作成していきます。
プロジェクトのディレクトリ直下にamplifyフォルダを作成し、その配下にauth,dataフォルダを作成します。
ディレクトリ構成のイメージとしては下記になります。

amplify/
├── auth/
│   └── resource.ts
├── data/
│   └── resource.ts
└── backend.ts

auth

resource.tsファイルを作成します。Cognitoでの認証を可能にします。
今回の主軸はリアルタイムサブスクリプションとはいえ、
ノーガードでデータリソースにアクセスされるのはよくないのでCognitoで認証・認可することとします。

import { defineAuth } from "@aws-amplify/backend";

export const auth = defineAuth({
  loginWith: {
    email: true,
  },
  userAttributes: {
    preferredUsername: {
      mutable: true,
      required: true,
    },
  },
});

data

resource.tsファイルを作成します。ここで今回扱うデータの定義をします。
今回はデバイスデータを取り扱うイメージでデータの定義を実装し、名前もDeviceStatusとします。

また、authorizationメソッドを使ってCognitoで認証されたユーザーのみがデータにアクセスできるようにします。

import { type ClientSchema, a, defineData } from "@aws-amplify/backend";

const schema = a.schema({
  DeviceStatus: a
    .model({
      device_Id: a.string(),
      status_code: a.string(),
      status_state: a.string(),
      status_description: a.string(),
      temperature: a.float(),
      humidity: a.float(),
      voltage: a.string(),
      last_updated: a.string(),
    })
    .authorization((allow) => [allow.authenticated()]),

export type Schema = ClientSchema<typeof schema>;

export const data = defineData({
  schema,
  authorizationModes: {
    defaultAuthorizationMode: "userPool"
  },
});

backend.ts

最後にバックエンド全体の定義を記載します。
作成したauth/resource.tsとdata/resourcce.tsをそれぞれimportして定義します。

import { defineBackend } from "@aws-amplify/backend";
import { auth } from "./auth/resource";
import { data } from "./data/resource";

defineBackend({
  auth,
  data,
});

ここまでで一旦、sandbox環境を作成します。
sandbox環境はローカルで画面をホストする前提で、DynamoDBやCognitoなどのAWSリソースを作成できる機能です。
Amplifyにデプロイせずとも開発時はローカル環境からAWSリソースとの連携を確認できて便利ですね。
下記コマンドを実行します。

npx ampx sandbox

イメージとしては下記に当たります。

コマンドの実行結果

✨  Total time: 201.94s

[Sandbox] Watching for file changes...
File written: amplify_outputs.json

File written: amplify_outputs.jsonが表示されたら完了です。
次にログイン画面を実装します。

補足

Sandbox環境の詳細については下記公式ドキュメントをご参照ください。

https://docs.amplify.aws/react/deploy-and-host/sandbox-environments/setup/

画面実装

main.tsx

初期表示画面を作成していきます。
Amplifyの設定を追加すればOKです。

import { StrictMode } from 'react'
import { createRoot } from 'react-dom/client'
import './index.css'
import App from './App.tsx'
+ import '@aws-amplify/ui-react/styles.css'

+ import { Amplify } from 'aws-amplify'
+ import outputs from '../amplify_outputs.json'

+ Amplify.configure(outputs)

createRoot(document.getElementById('root')!).render(
  <StrictMode>
    <App />
  </StrictMode>,
)

App.tsx

認証用のコンポーネントAuthenticatorを記載します。この配下に記載されているものが、ログイン後に表示されます。今回ならLogin Success!!が表示される想定です。

import "./App.css";
+ import { Authenticator } from "@aws-amplify/ui-react";

function App() {
+  return <Authenticator>Login Success!!</Authenticator>;
}

export default App;

認証用コンポーネントを組み込めたかサーバーを立ち上げて動作確認をします。

npm run dev

http://localhost:5173/にアクセスするとログイン画面が表示されますね。

Craete Accountタブをクリックして、アカウントを作成します。

• Email：任意のメールアドレス
• Password：任意のパスワード
• Confirm Password：任意のパスワード
• Preferred Username：ユーザー名

上記を入力したら、Create Accountボタンを押下します。

認証コードが記載されたメールが届き、届いたコードを画面に入力します。

Confirmを押下して、先に進みます。

Login Success!!と表示されていますね！！
これで認証周りまでは実装が完了したので、リアルタイムサブスクリプションを実装していきます。

UIライブラリの準備

今回はUI ライブラリのMantineを使うのでその準備も行なっていきます。
Mantineは便利なコンポーネントが揃っているので、簡単に画面を作るときに便利ですね。

npm install @mantine/core @mantine/hooks @tabler/icons-react
npm install --save-dev postcss postcss-preset-mantine postcss-simple-vars

postcss.config.cjsファイルをプロジェクト直下に作成し、下記設定を記載します。

module.exports = {
  plugins: {
    "postcss-preset-mantine": {},
    "postcss-simple-vars": {
      variables: {
        "mantine-breakpoint-xs": "36em",
        "mantine-breakpoint-sm": "48em",
        "mantine-breakpoint-md": "62em",
        "mantine-breakpoint-lg": "75em",
        "mantine-breakpoint-xl": "88em",
      },
    },
  },
};

次にApp.tsxにMantineProviderとstyles.cssのimportを追記します。

+ import '@mantine/core/styles.css';

+ import { MantineProvider } from '@mantine/core';

export default function App() {
+   return <MantineProvider>{/*ログイン処理*/}</MantineProvider>;
}

これで準備は完了です！

リアルタイムサブスクリプション処理を追加

いよいよリアルタイムサブスクリプション更新処理部分を作成します。
App.tsxに処理を追記します。

App.tsx全体

import { useState, useEffect, useRef } from "react";
import { generateClient } from "aws-amplify/data";
import type { Schema } from "../amplify/data/resource";
import { Authenticator } from "@aws-amplify/ui-react";
import { Table, Text, Group, Badge, ScrollArea } from "@mantine/core";
import {
  IconClock,
  IconThermometer,
  IconDroplet,
  IconBolt,
} from "@tabler/icons-react";
import "@mantine/core/styles.css";
import { MantineProvider } from "@mantine/core";

type DeviceStatus = Schema["DeviceStatus"]["type"];

const client = generateClient<Schema>();

function App() {
  const [devices, setDevices] = useState<DeviceStatus[]>([]);
  const [updatedIds, setUpdatedIds] = useState<Set<string>>(new Set());
  const previousDevices = useRef<DeviceStatus[]>([]);

  const getStatusColor = (status: string | null | undefined) => {
    return status === "NORMAL" ? "green" : "red";
  };

  const formatDateTime = (dateString: string) => {
    return new Date(dateString).toLocaleString();
  };

  useEffect(() => {
    const sub = client.models.DeviceStatus.observeQuery().subscribe({
      next: ({ items }) => {
        const sortedItems = [...items].sort(
          (a, b) =>
            new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()
        );

        const newUpdatedIds = new Set<string>();

        // 初回読み込み時は全てのアイテムをハイライト対象とする
        if (previousDevices.current.length === 0) {
          sortedItems.forEach((device) => {
            newUpdatedIds.add(device.id);
          });
        } else {
          // 2回目以降は変更があったアイテムのみをハイライト
          sortedItems.forEach((device) => {
            if (
              !previousDevices.current.find(
                (prev) =>
                  prev.id === device.id &&
                  prev.temperature === device.temperature &&
                  prev.humidity === device.humidity &&
                  prev.status_state === device.status_state
              )
            ) {
              newUpdatedIds.add(device.id);
            }
          });
        }

        setDevices(sortedItems);
        setUpdatedIds(newUpdatedIds);
        previousDevices.current = sortedItems;

        setTimeout(() => {
          setUpdatedIds(new Set());
        }, 3000);
      },
    });
    return () => sub.unsubscribe();
  }, []);

  const rows = devices.map((device) => (
    <Table.Tr
      key={device.id}
      style={{
        backgroundColor: updatedIds.has(device.id)
          ? "rgba(255, 255, 0, 0.1)"
          : undefined,
        transition: "background-color 0.5s ease",
      }}
    >
      <Table.Td>
        <Group gap="xs">
          <IconClock size={16} />
          <Text size="sm" fw={500}>
            {formatDateTime(device.createdAt)}
          </Text>
        </Group>
      </Table.Td>
      <Table.Td>
        <Text size="sm">{device.device_Id}</Text>
      </Table.Td>
      <Table.Td>
        <Badge color={getStatusColor(device.status_state)}>
          {device.status_state}
        </Badge>
      </Table.Td>
      <Table.Td>
        <Group gap="xs">
          <Text size="sm" c="dimmed">
            {device.status_code}
          </Text>
          <Text size="sm">-</Text>
          <Text size="sm">{device.status_description}</Text>
        </Group>
      </Table.Td>
      <Table.Td>
        <Group gap="xs">
          <IconThermometer size={16} />
          <Text size="sm">{device.temperature}°C</Text>
        </Group>
      </Table.Td>
      <Table.Td>
        <Group gap="xs">
          <IconDroplet size={16} />
          <Text size="sm">{device.humidity}%</Text>
        </Group>
      </Table.Td>
      <Table.Td>
        <Group gap="xs">
          <IconBolt size={16} />
          <Text size="sm">{device.voltage}V</Text>
        </Group>
      </Table.Td>
    </Table.Tr>
  ));

  return (
    <Authenticator>
      <MantineProvider>
        <ScrollArea>
          <Table striped highlightOnHover>
            <Table.Thead>
              <Table.Tr>
                <Table.Th>Timestamp</Table.Th>
                <Table.Th>Device ID</Table.Th>
                <Table.Th>State</Table.Th>
                <Table.Th>Status</Table.Th>
                <Table.Th>Temperature</Table.Th>
                <Table.Th>Humidity</Table.Th>
                <Table.Th>Voltage</Table.Th>
              </Table.Tr>
            </Table.Thead>
            <Table.Tbody>{rows}</Table.Tbody>
          </Table>
        </ScrollArea>
      </MantineProvider>
    </Authenticator>
  );
}

export default App;

要点

コード全体が長いので処理の要点をピックアップして説明します。

• useEffect フック内で observeQuery() メソッドを使用

  • DeviceStatusテーブルのデータに変更があった際に自動的に検知

  useEffect(() => {
      const sub = client.models.DeviceStatus.observeQuery().subscribe({
        next: ({ items }) => {
          // データ更新時の処理
        },
      });
      return () => sub.unsubscribe();
    }, []);
  

• 最新のデータが上に表示されるよう時系列でソート

  const sortedItems = [...items].sort((a, b) => 
    new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()
  );
  

• 更新されたデータを一時的にハイライト表示

  const newUpdatedIds = new Set<string>();
  
  // 初回読み込み時は全てのアイテムをハイライト対象とする
  if (previousDevices.current.length === 0) {
    sortedItems.forEach((device) => {
      newUpdatedIds.add(device.id);
    });
  } else {
    // 2回目以降は変更があったアイテムのみをハイライト
    sortedItems.forEach((device) => {
      if (
        !previousDevices.current.find(
          (prev) =>
            prev.id === device.id &&
            prev.temperature === device.temperature &&
            prev.humidity === device.humidity &&
            prev.status_state === device.status_state
        )
      ) {
        newUpdatedIds.add(device.id);
      }
    });
  }
  
  setDevices(sortedItems);
  setUpdatedIds(newUpdatedIds);
  previousDevices.current = sortedItems;
  
  setTimeout(() => {
    setUpdatedIds(new Set());
  }, 3000);
  

ここまでで実装は完了です。
下記のように空のテーブルが画面に表示されていればOKです。

まだデータがないのでデータを追加してみます。

データ登録

AppSyncにはコンソール上からGraphQLのクエリを実行できるクライアントが存在します。
そのクライアントを使用して今回使用するデバイスのデータを登録します。

AWSコンソール上からAppSyncの画面に遷移します。
画面から自動生成されたAPIのクエリを選択し、Cognitoで作成したユーザーでログインして、Mutationを実行して新規データを登録します。

まずはユーザープールでログインボタンを押下します。

先ほど登録したユーザーのメールアドレスとパスワードを入力してログインします。

ログイン成功後はExploerのAdd new operationからMutationを選択します。

クエリをコンソール上から作成できますが今回は下記クエリをコピペして実行するものとします。
クエリの内容はcreateDeviceStatusといった名前で、デバイスの情報を新規で登録するものとなります。

mutation MyMutation {
  createDeviceStatus(input: {device_Id: "device_003", humidity: 45.7, id: "status_1", last_updated: "2024-01-20T15:30:22Z", status_code: "200", status_description: "Normal operation", status_state: "ACTIVE", temperature: 23.4, voltage: "12.3"}) {
    id
    createdAt
    device_Id
    humidity
    last_updated
    status_code
    status_description
    status_state
    temperature
    updatedAt
    voltage
  }
}

クエリをペーストして、実行するボタンを押下します。
このタイミングで、作った画面も開いてリアルタイムで更新されるか確認します。

Mutationに成功したら下記のようにハイライトでデータが追加されていればOKです！

同じクエリでデータを再度追加したい場合は、id　を重複することはできないため、クエリのidを変更することで追加可能となります。

おわりに

Amplify Gen2でリアルタイムサブスクリプションを実装する方法はいかがでしたか。
Amplifyがラップしてくれている箇所も多く、迷わずシンプルに実装できましたね。
Amplifyを使用しているかつ、リアルタイムでアップデートを検知したい処理などがあった際は検討されてもいいのではと思いました。

本記事が少しでも参考になりましたら幸いです。ご覧いただきありがとうございました！