AppSync Events を Amplify なしで使ってみた

AppSync に新しく WebSocket による Publish/Subscribe (Pub/Sub) API として AppSync Events がリリースされました。公式ドキュメントでは Amplify クライアントを使った接続のサンプルが紹介されていますが、今回 Amplify を使用しない接続方法を試してみました。

(あとから気づいたのですが、AWS 公式ブログでも Amplify を使わない方式で紹介されていました… 🙇‍♂️)

AppSync Events とは？

公式ドキュメントでは、コネクションやリソースのスケーリングを管理することなく、リアルタイムのイベントデータを何百万ものサブスクライバーにブロードキャストできる、安全でパフォーマンスの高いサーバーレス WebSocket API を作成できるとあります。

これまでも AppSync ではリアルタイムなイベントを扱うことができましたが、GraphQL でのやり取りが前提でした。また、API Gateway でも WebSocket を扱うことができましたが、コネクションの管理など実装側で考慮すべき点がいろいろありました。今回リリースされた AppSync Evnets は両者からいいとこ取りをしたようなシンプルな WebSocket のマネージドサービスとして利用でき、簡単に WebSocket での Publish / Subscribe が実装できそうです。

公式ドキュメントはこちら : https://docs.aws.amazon.com/appsync/latest/eventapi/event-api-welcome.html

2025-03-14 Update

これまで Publish は HTTP エンドポイントのみの対応でしたが、WebSocket エンドポイントからも Publish できるようになりました 🙌 https://aws.amazon.com/jp/about-aws/whats-new/2025/03/appsync-events-publishing-websocket-real-time-pub-sub/

(本記事内のこのあたりも修正しました。)

クライアントからの接続方法

公式ドキュメントの Getting started ではクライアントとしては Amplify を使った例が示されています。Amplify を使うことで WebSocket を使った Pub / Sub がシンプルに実装できるのですが、このためだけに Amplify を導入するのは躊躇われるケースもあるかと思い、Amplify を使わずにネイティブな WebSocket(url, protocols) で AppSync Event に接続してみようと思ったのが今回のモチベーションです。

動作確認したアーキテクチャ

絵にするほどではないですが今回は以下のような構成を作って動作を確認してみました。ブラウザ側の React アプリとマネジメントコンソールの間でメッセージをやり取りしてみます (マネジメントコンソールでも Publish / Subscribe のテストが IoT Core と同じようにできます)。

Architecture

今回作成したものは GitHub でも公開しています: https://github.com/msysh/aws-sample-appsync-events

動作確認環境の構築

動作確認のためのデモ環境を作りました。

1. Event API の作成

まずはじめに、AppSync Events の API を作っていきます。マネジメントコンソールから作成する場合は API の名前を指定するだけで作成できます。自動的に default という名前空間が作成され、API のデフォルトの認証として API Key が設定 & 作成されます。

~~CDK で作成したい場合、L1 コンストラクトですが GitHub にあります。~~
　→ L2 コンストラクトに対応しました (GitHub)

Select Event API

Create Event API

作成できたら、API 一覧のリストから作成した API の詳細を開き、“設定” タブを選択後、以下の値をメモっておきます。

DNS エンドポイントの “HTTP”
DNS エンドポイントの “リアルタイム”
“認可設定"のセクションから API キー

AWS CLI で API を作成する場合

CLI で作成する場合は以下のようになります。
エラーになる場合は新しい機能なので AWS CLI がバージョンアップされているか確認してみてください。私はバージョン “2.19.1” で実行しました。

# API の作成
aws appsync create-api --name "My AppSync App" --event-config '{
  "authProviders": [ { "authType": "API_KEY" } ],
  "connectionAuthModes": [ { "authType": "API_KEY" } ],
  "defaultPublishAuthModes": [ { "authType": "API_KEY" } ],
  "defaultSubscribeAuthModes": [ { "authType": "API_KEY" } ]
}'

# 名前空間の作成 (`default` という名前で作成する場合)
# ${API_ID} は `create-api` の結果で得られる `api.apiId` の値
aws appsync create-channel-namespace --api-id ${API_ID} --name default

# 認証方法を API Key とした場合、API Key を作成する
# ${API_ID} は `create-api` の結果で得られる `api.apiId` の値
aws appsync create-api-key --api-id ${API_ID}

1つ目の create-api の実行結果である JSON から以下をメモっておきます。

api.dns.HTTP : HTTP エンドポイント
api.dns.REALTIME : リアルタイムエンドポイント

3つ目の create-api-key の実行結果である JSON から以下をメモっておきます。

apiKey.id : API キーの値

API Key 認証の場合、最後の create-api-key コマンドで --expires オプションで API Key の有効期限を設定できます。現在時刻から Epoch Unix タイムスタンプで指定し、最低でも 1日以上有効期間を設ける必要があります。指定しない場合のデフォルトの有効期間は 7日間なので注意しましょう。

また、API キーは複数発行することができるのでローテーションなども対応しやすいのではないかと思います。

ログ出力や WAF による保護

今回設定しませんが、CloudWatch Logs へはリクエストレベルのログやハンドラーのログを出力することができます (ドキュメント)。また、AWS WAF による保護も可能です (ドキュメント)。

補足: 名前空間と認証方法

API を作成すると、デフォルトの認証モードが、接続時、Publish 時、Subscribe 時それぞれ用に設定できます。名前空間を作成するとそれぞれの認証モードを引き継ぎますが、上書きすることもできます。名前空間は Publish や Subscribe 時のチャネル名 (トピック名) のプレフィックスとして設定されます。

例えば、名前空間 guest と user を作成し、未ログインユーザは API Key 認証で /guest/hoge が Subscribe できる、ログイン済みユーザは Cognito ユーザプール認証で /user/fuga が Subscribe できる、といった分け方ができるのではないかと思います。

ついでに、AppSync Events API で使える認証方法は以下が使えるようです (ドキュメント)。

API Key
Lambda
IAM (SigV4)
OpenID Connect
Amazon Cognito User Pool

2. クライアント側 (React アプリ) の作成

demo-app というフォルダを作成し、Vite で React プロジェクトを作っていきます。途中の選択肢は “React” と “TypeScript + SWC” を指定しました。

mkdir demo-app && cd $_
pnpm create vite@latest . -- --template react-ts

必要なものをインストールし、ローカルで React アプリを実行します。

pnpm install
pnpm run dev

続いて、src/App.tsx を以下のように書き換えます。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219


import { useState, useRef, } from 'react'

// 手順 1. でメモした値を設定
const AppSyncHttpEndpoint = 'HTTP エンドポイント';
const AppSyncRealTimeEndpoint = 'リアルタイムエンドポイント';
const AppSyncApiKey = 'API キー';

let instance: WebSocket | null = null;

// WebSocket オブジェクトを生成。すでに作成済みであればそちらを返す
const getWebSocketInstance = (url: string, protocols?: string | string[]): WebSocket => {
  if (!(instance instanceof WebSocket)) {
    instance = new WebSocket(url, protocols);
  }
  else if (instance.readyState === WebSocket.CLOSED || instance.readyState === WebSocket.CLOSING) {
    instance = new WebSocket(url, protocols);
  }
  return instance;
};

function App() {

  // Publish、Subscribe に使用するチャネル名
  const [channel, setChannel] = useState<string>('/default/test');

  // Subscribe した際の ID
  const [subscriptionId, setSubscriptionId] = useState<string>('N/A');

  // Publish するメッセージ
  const [publishMessage, setPublishMessage] = useState<string>('');

  // Subscribe により受信したメッセージのリスト
  const [receivedEvents, setReceivedEvents] = useState<object[]>([]);

  // WebSocket サブプロトコルに設定するヘッダコンテンツ
  // 詳細 : https://docs.aws.amazon.com/appsync/latest/eventapi/event-api-websocket-protocol.html#authorization-formatting-by-mode
  const headerInfo = {
    "host": AppSyncHttpEndpoint,
    "x-api-key": AppSyncApiKey,
  };
  // 認証情報は Base64URL 形式でエンコードする
  // https://docs.aws.amazon.com/appsync/latest/eventapi/event-api-websocket-protocol.html#websocket-connection-handshake
  const encodedHeaderInfo = btoa(JSON.stringify(headerInfo)).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');

  // WebSocket 接続を生成・取得。URL はリアルタイムエンドポイントを指定。
  // 認証情報はサブプロトコルとして指定する。
  const ws = getWebSocketInstance(
    `wss://${AppSyncRealTimeEndpoint}/event/realtime`,
    [
      'aws-appsync-event-ws',
      `header-${encodedHeaderInfo}`
    ]
  );

  // https://docs.aws.amazon.com/appsync/latest/eventapi/event-api-websocket-protocol.html#message-details
  // WebSocket が確立したら init メッセージを送信
  ws.onopen = (event) => {
    ws.send(JSON.stringify({ "type": "connection_init" }));
  };

  ws.onmessage = (event) => {
    console.log(event);
    const message = JSON.parse(event.data);

    switch (message.type) {
      case 'connection_ack':
        // init メッセージの送信結果として接続確認メッセージが返ってきた時
        break;

      case 'subscribe_success':
        // Subscribe に成功した時
        // Subscribe を解除 (Unsubscribe) する際に必要になるため保管しておく
        setSubscriptionId(message.id);
        break;

      case 'subscribe_error':
        // Subscribe に失敗した時
        break;

      case 'data':
        // Subscribe によりメッセージを受信した時
        setReceivedEvents((currentEvents) => {return [... currentEvents, message.event]});
        break;

      case 'broadcast_error':
        // ブロードキャストエラーなどが発生した場合、クライアントでエラーを受信することがある
        break;

      case 'unsubscribe_success':
        // Subscribe 解除に成功した時
        // 保管していた Subscription ID をリセット
        setSubscriptionId('N/A');
        break;

      case 'unsubscribe_error':
        // Subscribe 解除でエラーが発生した時
        break;

      case 'publish_success':
        // Publish に成功した時 (リアルタイムエンドポイント、すなわち WebSocket 経由で Publish した時の応答)
        // `id` に Publish した際の id が設定されている (`successful[].identifier` は別の固有の値)
        break;

      case 'publish_error':
        // Publish でエラーが発生した場合 (リアルタイムエンドポイント、すなわち WebSocket 経由で Publish した時の応答)
        break;

      case 'ka':
        // 接続維持のため、AppSync から Keep-Alive メッセージ (60秒間隔) を受信した時
        // 初期化時に得られる `connectionTimeoutMs` のタイマーをリセットしたりすると良さそう
        break;

      default:
        break;
    }
  };

  const wsRef = useRef(ws);

  const onClickSubscribe = () => {
    // Subscribe を開始する
    wsRef.current.send(JSON.stringify({
      "type": "subscribe",
      "id": crypto.randomUUID(),
      "channel": channel,
      "authorization": {
        "host": AppSyncHttpEndpoint,
        "x-api-key": AppSyncApiKey,
      }
    }));
  };

  const onClickUnsubscribe = () => {
    // Subscribe を解除 (Unsubscribe) する
    wsRef.current.send(JSON.stringify({
      "type": "unsubscribe",
      "id": subscriptionId,
    }));
  };

  const onClickPublishViaRealtime = () => {
    // メッセージを Publish する (リアルタイムエンドポイントを使用した WebSocket による Publish)
    // https://docs.aws.amazon.com/appsync/latest/eventapi/publish-websocket.html
    wsRef.current.send(JSON.stringify({
      "type": "publish",
      "id": crypto.randomUUID(),
      "channel": channel,
        "events": [ JSON.stringify(publishMessage) ],
      "authorization": {
        "host": AppSyncHttpEndpoint,
        "x-api-key": AppSyncApiKey,
      }
    }));
  };

  const onClickPublishViaHttp = () => {
    // メッセージを Publish する (HTTP エンドポイントを使用した Publish)
    // https://docs.aws.amazon.com/appsync/latest/eventapi/publish-http.html
    fetch(`https://${AppSyncHttpEndpoint}/event`, {
      "method": "POST",
      "headers": {
        "content-type": "application/json",
        "x-api-key": AppSyncApiKey,
      },
      "body": JSON.stringify({
        "channel": channel,
        "events": [ JSON.stringify(publishMessage) ]
      })
    })
    .then((value) => {
      console.log(value);
    }, (reason) => {
      console.error(reason);
    });
  };

  return (
    <>
      <h1>AppSync Event without Amplify</h1>
      <div>
        <label>Channel: </label>
        <input type="text" value={ channel } onChange={(event) => setChannel(event.target.value)} />
        <button onClick={ onClickSubscribe }>
          Subscribe
        </button>
      </div>
      <hr />
      <div>
        <label>Subscription ID: </label>
        <span>{subscriptionId}</span>
        <button onClick={ onClickUnsubscribe }>
          Unsubscribe
        </button>
      </div>
      <hr />
      <div>
        <input type="text" value={ publishMessage } onChange={ (event) => setPublishMessage(event.target.value) } />
        <button onClick={ onClickPublishViaHttp }>
          Publish (HTTP Endpoint)
        </button>
        <button onClick={ onClickPublishViaRealtime }>
          Publish (Realtime Endpoint)
        </button>
      </div>
      <hr />
      <div>
        <ul>
          {
            receivedEvents.map((message, index) => {
              return <li key={ index }>{ message.toString() }</li>
            })
          }
        </ul>
      </div>
    </>
  )
}

export default App

以上で確認環境としては準備完了です。解説の前に先に動かしてみたい方は下の方の動作確認をご覧ください。

(解説) Amplify を利用しない AppSync Events の利用方法

Amplify を使わずにネイティブな WebSocket で AppSync Events に接続する方法を順を追って解説したいと思います。

1. WebSocket オブジェクトの生成

WebSocket オブジェクトを生成する際、必須 1つ、任意 1つの引数を指定することができます (参考: MDN)。

ws = new WebSocket(url, protocols);

url には AppSync Event API のリアルタイムエンドポイントを使った、wss://${AppSyncRealTimeEndpoint}/event/realtime を指定します。/event/realtime のところは固定値です
protocols にはプロトコル文字列を設定できます。AppSync Event API では以下の 2つを設定します。
- aws-appsync-event-ws
- 認証情報などをまとめ、Base64URL 形式でエンコードした文字列

サブプロトコルに指定する認証情報

protocols に指定する認証情報は認証モードに応じて JSON オブジェクトから作成します。例えば API Key 認証の場合は下記になります。その他の認証モードについてはドキュメントを参照ください。host は HTTP エンドポイントである点に注意してください。 (行番号は前述の src/App.tsx での行番号と一致しています。)

37
38
39
40


const headerInfo = {
  "host": AppSyncHttpEndpoint,
  "x-api-key": AppSyncApiKey,
};

ドキュメントによると、この JSON オブジェクトを Base64URL 形式でエンコード (さらにいくつかの文字を置換)し、、、

43

const encodedHeaderInfo = btoa(JSON.stringify(headerInfo)).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');

さらに header- を前に付与したものをサブプロトコルとして指定する必要があるようです。

まとめると、前述の src/App.tsx では以下のようになります (getWebSocketInstance()の中で new WebSocket(url, protocols) を呼び出しています。)

47
48
49
50
51
52
53


const ws = getWebSocketInstance(
  `wss://${AppSyncRealTimeEndpoint}/event/realtime`,
  [
    'aws-appsync-event-ws',
    `header-${encodedHeaderInfo}`
  ]
);

2. 初期化

接続を確立するだけでは AppSync Events を利用できません。いくつかの手続きを踏む必要があるので、ドキュメントに倣って実行していきます。

まず初期化として init メッセージを送信します。動作確認では、WebSocket 接続確立後、onopen のタイミングで送信しました。
前述の src/App.tsx ですと以下の箇所です。

57
58
59


ws.onopen = (event) => {
  ws.send(JSON.stringify({ "type": "connection_init" }));
};

init メッセージのレスポンス

初期化が成功すると、以下のようなメッセージが返ってきます。今回実装していないですがクライアントが接続タイムアウト期間内 (connectionTimeoutMs) にキープアライブメッセージを受信しない場合、クライアント側から接続を閉じてあげる必要があるとドキュメントにはあります。

{
  "type": "connection_ack",
  "connectionTimeoutMs": 300000
}

動作確認では init のレスポンスに対しては何もしていませんが WebSocket.onmessage で、受信したメッセージの event.data.type が connection_ack の時として判別することができます。
前述の src/App.tsx ですと以下の箇所でハンドリングできます。

61
62
63
64
65
66
67
68
69


ws.onmessage = (event) => {
  console.log(event);
  const message = JSON.parse(event.data);

  switch (message.type) {
    case 'connection_ack':
      // init メッセージの送信結果として接続確認メッセージが返ってきた時
      break;
      //  :

初期化が完了すると、Subscribe が可能になります。

Subscribe するには以下のような JSON を WebSocket から送信することで可能になります。

{
  "type": "subscribe",
  "id": "abcdef12-abcd-1234-abcd-1234567890ab",
  "channel": "/YOUR_NAME_SPACE/SUB_A/SUB_B",
  "authorization": {
    "x-api-key": "da2-12345678901234567890123456",
    "host": "example1234567890000.appsync-api.us-east-1.amazonaws.com"
   }
}

id はクライアントコネクションごとに一意である必要があります。重複するとエラーが返ります。また、Subscribe を解除 (Unsubscribe) する時に必要になります。
authorization は WebSocket の接続時の認証情報と同じものを指定します (src/App.tsx の 37 - 40行目)
authorization.host はリアルタイムエンドポイントではなく HTTP エンドポイント である点にご注意ください

今回の動作確認では、ボタンクリック時に Subscribe を開始するようにしました。
前述の src/App.tsx ですと以下の箇所です。

120
121
122
123
124
125
126
127
128
129
130
131


const onClickSubscribe = () => {
  // Subscribe を開始する
  wsRef.current.send(JSON.stringify({
    "type": "subscribe",
    "id": crypto.randomUUID(),
    "channel": channel,
    "authorization": {
      "host": AppSyncHttpEndpoint,
      "x-api-key": AppSyncApiKey,
    }
  }));
};

Subscribe に成功すると event.data.type に subscribe_success が設定されたメッセージが返ります。event.data.id は Subscribe する際に指定した ID と同じで、Subscribe を解除 (Unsubscribe) する時に必要になります。エラーが発生すると subscribe_error が返されます (メッセージフォーマットなどはドキュメントを参照してください)。
前述の src/App.tsx ですと以下の箇所でハンドリングできます。

70
71
72
73
74
75
76
77
78


  case 'subscribe_success':
    // Subscribe に成功した時
    // Subscribe を解除 (Unsubscribe) する際に必要になるため保管しておく
    setSubscriptionId(message.id);
    break;

  case 'subscribe_error':
    // Subscribe に失敗した時
    break;

4. データメッセージの受信

Subscribe が成功し、Subscribe しているチャネルにイベントが Publish されると、データメッセージとして以下のフォーマットで受信できます。

{
  "type": "data",
  "id": "abcdef12-abcd-1234-abcd-1234567890ab",
  "event": ["\"my published message \""]
}

前述の src/App.tsx では event.data.type が data の時にハンドリングできます。
今回の動作確認では受信したメッセージを単純にリストに追加していきました。

80
81
82
83


case 'data':
  // Subscribe によりメッセージを受信した時
  setReceivedEvents((currentEvents) => {return [... currentEvents, message.event]});
  break;

ブロードキャストエラーが発生すると event.data.type に broadcast_error が設定されてクライアント側に届くそうです (メッセージフォーマットなどはドキュメントを参照してください)。
前述の src/App.tsx ですと以下の箇所でハンドリングできます。

85
86
87


case 'broadcast_error':
  // ブロードキャストエラーなどが発生した場合、クライアントでエラーを受信することがある
  break;

Subscribe を解除するには Subscribe 時に指定した ID (= Subscribe が成功した時の応答メッセージの ID) を指定する必要があり、以下のようなフォーマットになります。

{
  "type": "unsubscribe",
  "id": "abcdef12-abcd-1234-abcd-1234567890ab"
}

今回の動作確認では、“Unsubscribe” ボタンを押した際に Unsubscribe するようにしてみました。

133
134
135
136
137
138
139


const onClickUnsubscribe = () => {
  // Subscribe を解除 (Unsubscribe) する
  wsRef.current.send(JSON.stringify({
    "type": "unsubscribe",
    "id": subscriptionId,
  }));
};

Unsubscribe に成功した場合は、event.data.type に unsubscribe_success が設定されたメッセージが届きます。今回の動作試験では保管しておいた Subscription ID をリセットするようにしました。
前述の src/App.tsx ですと以下のようにしています。

89
90
91
92
93


case 'unsubscribe_success':
  // Subscribe 解除に成功した時
  // 保管していた Subscription ID をリセット
  setSubscriptionId('N/A');
  break;

Unsubscribe にエラーがあった場合、event.data.type に unsubscribe_error が設定されたメッセージが届きます (メッセージフォーマットなどはドキュメントを参照してください)。
前述の src/App.tsx ですと以下の箇所でハンドリングできます。

95
96
97


case 'unsubscribe_error':
  // Subscribe 解除でエラーが発生した時
  break;

6. Publish

Publish は従来 HTTP エンドポイントを使用した方法しかサポートされていませんでしたが、2025年3月14日にリアルタイムエンドポイントを使用した WebSocket による Publish がサポートされました。リアルタイムエンドポイントと HTTP エンドポイントそれぞれを使った Publish をやってみます。

6-1. WebSocket による Publish (リアルタイムエンドポイントを使用)

アップデートにより、単一の WebSocket 接続を使用してメッセージを Publish できるようになりました。

WebSocket 経由で Publish するには以下のような JSON を送信します。

{
  "type": "publish",
  "id": "abcdef12-abcd-1234-abcd-1234567890ab",
  "channel": "/YOUR_NAME_SPACE/SUB_A/SUB_B",
  "event": ["\"my published message \""],
  "authorization": {
    "x-api-key": "da2-12345678901234567890123456",
    "host": "example1234567890000.appsync-api.us-east-1.amazonaws.com"
   }
}

authorization は WebSocket の接続時の認証情報と同じものを指定します (src/App.tsx の 37 - 40行目)
authorization.host はリアルタイムエンドポイントではなく HTTP エンドポイント である点にご注意ください

今回の動作確認では、“Publish (Realtime Endpoint)” というボタンをクリックした時に Publish するようにしました。
前述の src/App.tsx ですと以下の箇所です。

141
142
143
144
145
146
147
148
149
150
151
152
153
154


const onClickPublishViaRealtime = () => {
  // メッセージを Publish する (リアルタイムエンドポイントを使用した WebSocket による Publish)
  // https://docs.aws.amazon.com/appsync/latest/eventapi/publish-websocket.html
  wsRef.current.send(JSON.stringify({
    "type": "publish",
    "id": crypto.randomUUID(),
    "channel": channel,
      "events": [ JSON.stringify(publishMessage) ],
    "authorization": {
      "host": AppSyncHttpEndpoint,
      "x-api-key": AppSyncApiKey,
    }
  }));
};

Publish に成功すると event.data.type に publish_success が設定された以下のようなメッセージが返ります。
id は Publish 時に指定した id と同じ値が設定されてきます。

{
  "id": "abcdef12-abcd-1234-abcd-1234567890ab",
  "type": "publish_success",
  "successful": [
    {
      "identifier": "221b6b3f-46fd-4ac1-a327-bd2623b7402e",
      "index": 0
    }
  ],
  "failed": []
}

Publish 時にエラーがあった場合は event.data.type に publish_error が設定されたメッセージが返ってきます (メッセージフォーマットなどはドキュメントを参照してください)。

今回は Publish の成功時も失敗時も特に処理は記述しませんでしたが、前述の src/App.tsx では以下の箇所でハンドリングできます。

 99
100
101
102
103
104
105
106


case 'publish_success':
  // Publish に成功した時 (リアルタイムエンドポイント、すなわち WebSocket 経由で Publish した時の応答)
  // `id` に Publish した際の id が設定されている (`successful[].identifier` は別の固有の値)
  break;

case 'publish_error':
  // Publish でエラーが発生した場合 (リアルタイムエンドポイント、すなわち WebSocket 経由で Publish した時の応答)
  break;

6-2. Publish (HTTP エンドポイントを使用)

Publish は HTTP エンドポイントから POST メソッドでも実行できます。ドキュメントはこちら。
今回の動作確認では “Publish (HTTP Endpoint)” というボタンをクリックした時に処理として以下のようにしてみました。

156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175


const onClickPublishViaHttp = () => {
  // メッセージを Publish する (HTTP エンドポイントを使用した Publish)
  // https://docs.aws.amazon.com/appsync/latest/eventapi/publish-http.html
  fetch(`https://${AppSyncHttpEndpoint}/event`, {
    "method": "POST",
    "headers": {
      "content-type": "application/json",
      "x-api-key": AppSyncApiKey,
    },
    "body": JSON.stringify({
      "channel": channel,
      "events": [ JSON.stringify(publishMessage) ]
    })
  })
  .then((value) => {
    console.log(value);
  }, (reason) => {
    console.error(reason);
  });
}

~~リアルタイムエンドポイントから Publish はできないのか~~ 2025年 3月 14日のアップデートで可能になりました。

ドキュメントによるとリアルタイムのエンドポイントでも対応してそうに読み取れたのですが、UnknownOperation となってしまい Publish できませんでした (私の指定の仕方が悪かったのかもしれません…)。

type:
The type of operation being performed. Supported client operations are subscribe, unsubscribe, publish. The property is a string and must be one of the message types defined in the next section, Configuring message details.

~~リアルタイムエンドポイントを使って以下のように Publish してみたのですが (いろいろ試してみたのですが)…~~

wsRef.current.send(JSON.stringify({
  "type": "publish",
  "id": crypto.randomUUID(),
  "channel": channel,
  "events": [ "message" ],
  "authorization": {
    "host": AppSyncHttpEndpoint,
    "x-api-key": AppSyncApiKey,
  }
}));

~~以下のようなエラーになってしまいました。~~

{
  "id": "abcdef12-abcd-1234-abcd-1234567890ab",
  "type": "publish_error",
  "errors": [
    {
      "errorType": "UnknownOperationException",
      "message": "Unknown Operation Request."
    }
  ]
}

7. Keep-Alive の受信

AppSync Events から 60秒間隔で Keep-Alive メッセージが届きます。event.data.type が ka の時として受信できます。
src/App.tsx では以下の箇所でハンドリングできます。

 99
100
101
102


case 'ka':
  // 接続維持のため、AppSync から Keep-Alive メッセージ (60秒間隔) を受信した時
  // 初期化時に得られる `connectionTimeoutMs` のタイマーをリセットしたりすると良さそう
  break;

今回は実装していないのですが、2. の初期化時で得られる connectionTimeoutMs のタイマーをリセットする実装を入れると良さそうです。

動作確認

それでは実際に動かして確認してみましょう (動作確認の環境構築が完了しており、pnpm run dev でローカルのサーバが起動していることが前提です)。

まずは React アプリから Subscribe し、マネジメントコンソールから Publish したメッセージが届くか確認してみます。

ブラウザを開き、http://localhost:5173 に接続すると以下のようになるかと思います。"Subscribe” ボタンを押して Subscribe します。

Subscribe from React App

続いてマネジメントコンソールから作成した API の詳細を開き、"Pub / Sub エディタ" タブを選択します。

“パブリッシュ” セクションで認証タイプを合わせ、"チャンネル" を React アプリと合わせます (React アプリの Channel が /default/test であれば、名前空間に default、次のテキストボックスに /test と入れます)。テキストエリアに適当にメッセージを入れ “パブリッシュ” します (HTTP と WebSocket どちらからでも OK です)。

Publish from Management Console

以下のように React アプリ側でメッセージが受信できていれば OK です。

Receive message at React App

“Unsubscribe” ボタンを押すと、Subscribe を解除します。

Unsubscribe from React App

続いて逆向き、マネジメントコンソールから Subscribe し、React アプリから Publish したメッセージが届くかも確認してみます。

マネジメントコンソールで作成した API の詳細を開き、"Pub / Sub エディタ" タブを選択後、下の方に “サブスクライブ” セクションがあります。

“チャンネル” はこちらではワイルドカード (/*) を指定してみます。"接続" 後、"サブスクライブ" します。

Subscribe from Management Console

続いて React アプリ側から Publish します。WebSocket を使ったリアルタイムエンドポイント経由と、HTTP POST メソッドを使った HTTP エンドポイント経由、それぞれ確認してみます。

Publish from React App

マネジメントコンソール側でもメッセージが受信できていれば OK です。

Receive message at Management Console

まとめ

AppSync Events という使いやすい WebSocket のマネージドサービスが登場しました。Amplify を利用しないネイティブな WebSocket で利用する方法を確認してみました。~~Publish に関しては HTTP エンドポイントを使った POST になるというところはポイントかもしれません。~~ (2025年 3月 14日のアップデートでリアルタイムエンドポイントを使用した WebSocket 経由の Publish にも対応しました)。

ご参考になれば幸いです。

(今回作成したものは GitHub でも公開しています: https://github.com/msysh/aws-sample-appsync-events )

最後に・・・

この投稿は個人的なものであり、所属組織を代表するものではありません。ご了承ください。