[AWS IoT - Direct Messaging] MQTTのトピック設計を止めて、デバイスのグルーピングをサーバーサイドで実施してみました

1 はじめに

製造ビジネステクノロジー部の平内（SIN）です。

2026年5月、AWS IoT Core に Direct Messaging（SendDirectMessage API）という機能が追加されました。

参照: Introducing AWS IoT Core's Direct Messaging

これを使用すると、デバイスの分類をトピック階層に埋め込まず、配信対象をサーバーサイドの条件クエリで決める構成が書けるようになりそうです。

本稿では、それを試すために、100台の仮想デバイスをダッシュボードに並べ、条件を指定して送信すると該当デバイスだけが受信して色が変わるデモを作ってみました。まずは動作をご覧ください。

すべてのメッセージは、同じトピック名で送信していますが、宛先デバイスを選別できているようすが確認できると思います。

https://www.youtube.com/watch?v=co9aodZaZO4

※ 本稿は、「MQTTのトピック設計」を否定するものではありません。後段の「トレードオフ」にも、ご注意ください。

2 トピック設計の課題と Direct Messaging

(1) Direct Messaging とは

従来のように「デバイスが購読しているトピックに Publish し、購読マッチングで届く」のではなく、AWS IoT Core が clientId 宛にポイントツーポイントでルーティングします。送信側は IoT データエンドポイントに HTTP POST するだけです。

POST https://{IoT_data_endpoint}/connections/{clientId}/messages?topic={topic}&confirmation=true&timeout=10

主な仕様は次の通りです。

API レスポンスと CloudWatch Logs で配信の成否と失敗理由を追える点も特徴です。

参照: Direct Messaging - AWS IoT Core Developer Guide

(2) トピック階層に分類をハードコードする設計の限界

従来の MQTT では、「誰に送るか」をトピック階層で表現するのが定石でした。

factory/tokyo/line-a/robot/0001
region/jp/tier/premium/device/0001

これは便利な反面、デバイスの分類がトピック文字列にハードコードされている状態でもあり、次のような課題もあります。

• 分類変更のコスト: ライン再編で line-a から line-b になるとデバイス側の設定変更が必要になる
• 多軸の絞り込み不可: 階層は木構造なので1軸しか表現できず、「東京 かつ プレミアム契約」が表現しづらい
• 軸の後付け不可: 設計時に想定しなかった分類軸を後から足すのは難しい
• 分類ミスの回復困難: 誤った分類で出荷したデバイスは現地対応やファーム更新が要る

(3) 提案 — フラットな識別子 + サーバーサイド宛先解決

Direct Messaging は clientId で直接送れるため、トピックに分類を持たせる必要がなくなります。デバイスはフラットな識別子だけを持って出荷し、

device/0001
device/0002

「誰に送るか」は送信時にサーバー側で決めます。その一つの応用が、トピックを単なるアドレスとして扱い、デバイスのグルーピングロジックをサーバーサイドに寄せる設計です。今回はその一例として、デバイスのメタデータ（リージョン・ライン・ファームウェア・契約種別など）を DynamoDB に持たせ、配信対象を DB クエリで決めました。

# トピック階層では表現しにくい多軸クエリが、DB クエリになる
# DBからdeviceIdを検索し、それを送信対象にする
targets = query(region="tokyo", firmware="< 1.2.0", contract="premium")
for d in targets:
    send_direct_message(clientId=d.deviceId, payload=...)

「東京にある、ファーム 1.2.0 未満の、プレミアム契約デバイス」という絞り込みが、素直なクエリで表現できます。

3 構成図・アーキテクチャ

構成は、以下のとおりです。

1. 画面の取得は、CloudFront経由でS3から配信されます。
2. ブラウザで選択した条件は、Lambda（関数URL）に送信されます。
3. DynamoDBをScan+Filterし、宛先を解決します。
4. SendDirectMessage を使用して fan-out します。
5. Direct Messaging で配信されます(confirmation=true → PUBACK)。
6. 送信の結果は、dashboard/events をSubscribeしているブラウザに伝達されます (MQTT over WSS)。

「デバイスランナー」 ローカルNodeプロセスで仮想デバイスをシミュレートし、100本のMQTT接続（clientId=device-0001〜0100）を表現しています。

(1) 送信結果の伝達

confirmation=true の場合、SendDirectMessage は PUBACK を受け取ってから 200 を返します。つまり Lambda が 200 を得た時点で「そのデバイスが実際に受信した」ことが確定します。

そこで、Lambda が配信結果を1台ずつ dashboard/events トピックへ publish し、ブラウザがそれをSubscribeしてタイルを緑にする流れになっています。デバイスからダッシュボードへ報告する経路は、特に用意していません。

(2) 認証

ブラウザもデバイスランナーも、Cognito Identity Pool の未認証（ゲスト）ID を使用して一時クレデンシャルを取得し、IoT に接続しています（MQTT over WebSocket / SigV4）。デモを簡略化するため、X.509 証明書は使用していません。

IDに付与したのは、最低限の権限となっています。

• ブラウザ: iot:Connect / iot:Subscribe / iot:Receive（dashboard/events のみ）
• デバイス: iot:Connect / iot:Receive（commands/*。購読は不要）

4 実装 — CDK + Lambda + DynamoDB + 100台可視化ダッシュボード

Github aws-iot-direct-messaging-demo

(1) DynamoDB

deviceId を主キーに、分類軸を属性として持たせています。

デモは100件規模なので、宛先解決は Scan + FilterExpression となっています。

(2) dispatcher Lambda — 宛先解決と fan-out

Github cdk/lambda/dispatcher/index.ts

条件で DynamoDB を絞り込み、該当 clientId へ1台ずつ SendDirectMessage を送り、結果を dashboard/events に publish します。

// 宛先解決: DynamoDB を Scan + フィルタ
const scan = await ddb.send(new ScanCommand({ TableName: TABLE }));
const devices = (scan.Items || []).map(toDevice);
const targets = devices.filter((d) => matches(d, cond));

// 各 clientId へ SendDirectMessage を fan-out（1台ずつ結果をイベント配信）
for (const d of targets) {
  let status = "ok";
  try {
    await iot.send(new SendDirectMessageCommand({
      clientId: d.deviceId,
      topic: CMD_TOPIC,
      confirmation: confirm,   // true なら PUBACK 待ち
      timeout: 10,
      payload: enc.encode(JSON.stringify({ action: "highlight" })),
    }));
  } catch (e) {
    const code = e?.$metadata?.httpStatusCode;
    status = code === 404 ? "offline" : code === 429 ? "throttled" : "error";
  }
  await publishEvent({ type: "delivery", deviceId: d.deviceId, status });
}

404（オフライン）や 429（スロットリング）を状態として拾えるため、配信の成否を1台ずつ追跡できます。

なお SendDirectMessage は新しい API のため、 CDK の NodejsFunction（esbuild）で新しい @aws-sdk/client-iot-data-plane をバンドル同梱しています。

(3) CDK — 送信側・受信側の IAM

Github cdk/lib/stack.ts

以下が、Direct Messaging を送信・受信するための権限です。

// 送信側(Lambda): 対象 client への SendDirectMessage と、イベント publish
dispatcher.addToRolePolicy(new iam.PolicyStatement({
  actions: ["iot:SendDirectMessage"],
  resources: [`arn:aws:iot:${region}:${account}:client/*`],
}));

// 受信側(ゲストロール): 購読なしで受け取るため iot:Receive のみ（iot:Subscribe 不要）
guestRole.addToPolicy(new iam.PolicyStatement({
  actions: ["iot:Receive"],
  resources: [`arn:aws:iot:${region}:${account}:topic/commands/*`],
}));

(4) デバイスランナー

Github devices/runner.mjs

Direct Messaging は購読なしでメッセージが届きます。ランナーは clientId を変えて N 本つなぎ、受信ハンドラを持つだけです（QoS1 の PUBACK は mqtt.js が自動送信します）。

for (let i = 1; i <= N; i++) {
  const id = "device-" + String(i).padStart(4, "0");
  const c = mqtt.connect(url, { clientId: id, reconnectPeriod: 5000 });
  c.on("connect", () => { up.add(id); render(); });
  // Direct Messaging は購読なしで届く。全メッセージを処理する
  c.on("message", () => { received++; render(); });
}

受信側の注意点として、購読していないトピックのメッセージをクライアントライブラリが破棄しない実装であることが前提になります。AWS IoT Core は購読の有無に関わらず配信しますが、クライアント側で破棄すると（購読不要という利点が失われ）そのトピックでは受信できません。加えて、多くのライブラリは MQTT プロトコル層で PUBACK を返してからアプリのハンドラに渡すため、「PUBACK は返るがアプリは破棄する」実装だと、送信側は成功（200）を受け取るのにデバイスは何もしていない、というズレが起き得ます。デモの mqtt.js は購読状態に関わらず受信メッセージをイベントに流すため、そのまま処理できます。

(5) ダッシュボード — イベント購読で色を変える

Github web/app.js

ブラウザは Cognito ゲスト ID で IoT に WSS 接続し、dashboard/events を購読して、届いた delivery イベントのタイルを緑にします。

const creds = await getCreds(cfg.region, cfg.identityPoolId);       // Cognito 未認証
const url = await presignWssUrl(cfg.iotEndpoint, cfg.region, creds); // SigV4 署名
const client = mqtt.connect(url, { clientId: "dashboard-" + rand });
client.on("connect", () => client.subscribe(cfg.eventsTopic));
client.on("message", (_t, payload) => onEvent(JSON.parse(payload.toString())));

条件パネルの送信ボタンは、条件を Function URL に POST するだけです。あとは Lambda が配信し、その結果イベントで画面が更新されます。

5 デモの利用方法

(1) CDK デプロイ

git clone https://github.com/furuya02/aws-iot-direct-messaging-demo.git
cd aws-iot-direct-messaging-demo/cdk
pnpm install
pnpm cdk bootstrap  # 初回のみ
pnpm cdk deploy -c bucket_suffix=$(date +%Y%m%d) --outputs-file ../outputs.json

デプロイが終わると outputs.json にダッシュボードURL・IoTエンドポイント・Cognito Identity Pool ID などが出力されます。

(2) デバイスメタデータの投入と仮想デバイスの起動

cd ..
pnpm install
pnpm seed          # DynamoDB に100件投入

別ターミナルで100台の仮想デバイスを起動

$ pnpm devices 100

> aws-iot-direct-messaging-demo@0.1.0 devices aws-iot-direct-messaging-demo
> node devices/runner.mjs 100

起動中: 100 台の仮想デバイス（Ctrl+C で終了）
接続: 100/100  受信累計: 51

接続: 100/100 と表示されれば、100台がオンラインです。Ctrl+Cで停止します。

(3) ブラウザで表示・操作

outputs.json の DashboardUrl（CloudFront）を開きます。

1. 右上のバッジが「接続済み」になる（Cognito から IoT WSS 購読の成功）
2. 左パネルで条件（例: region = tokyo）を選んで送信する
3. 該当タイルが橙（送信中）から緑（受信完了）に変化し、「宛先ヒット / 受信完了」のカウントが動く
4. プリセットで各シナリオも試せる（後付け軸で絞り込み、カナリアリリース、全台一斉配信）

条件はサーバー側のクエリを変えるだけで、デバイスは一切変更していません。それでも配信対象が柔軟に変わることを確認できます。

(4) 片付け

検証が終わったら削除します。(常時課金の要素はありません)

cd cdk
pnpm cdk destroy

6 トレードオフ

「トピック設計から解放される」とは、書きましたが、トレードオフもあります。採用するかどうかは、この辺の検討が必要です。

(1) 配信効率

Pub/Sub のブロードキャストはブローカー側でファンアウトが完結し、送信側は1回 Publish するだけです。対して Direct Messaging は N台に送るなら N回の API 呼び出しが要ります。

少なくとも、ブロードキャスト用トピックを Subscribe（通常の Pub/Sub）しておき、一斉配信は、そちらを使用するという並行運用が良いと思いました。

(2) 状態管理の責務が送信側に移る

「誰が受け取るか」をブローカーではなく自前の DB で持つことになります。

• DB とデバイス実態の乖離（移設したのに DB が古い → 誤配信）→ 定期ハートビートで突き合わせる
• プロビジョニング時の登録漏れ → Fleet Provisioning のフックで DB 登録を必須化する
• メタデータストアが単一障害点 → グローバルテーブルと宛先解決のキャッシュ、緊急用に全台向け Pub/Sub 経路を残す

(3) 移行・共存のコスト

• Direct Message は IoT Rules を通らないため、Rule で加工・転送していた処理はそのまま流用できません。
• 全台通知は Pub/Sub、条件配信は Direct と使い分けると2経路の運用・監視が必要になります。ペイロードに配信経路を持たせるなどして、ログで追えるようにする必要がありそうです。

(4) その他

• メタ情報を管理する仕組み（デモではDynamoDB）が、単一障害点になります。
• 今回は、デバイスが100台ということで、メタ情報の管理をDynamoDB(scan)としましたが、規模が大きくなると、GISや、OpenSearchなども必要になるかも知れません。
• ペンディングキュー、リトライ管理、バージョン管理は、従来ブローカーや MQTT 仕様が担当していた部分ですが、配信の状態管理を自前で引き受ける責務を認識する必要があります。

7 最後に

AWS IoT Core の Direct Messaging は、トピックに分類をハードコードする前提を外し、配信対象をサーバー側のクエリで決める設計を現実的にしてくれます。今回のデモでは、条件を変えるだけで該当デバイスだけが色を変える様子を確認できました。

色々なメリットがありそうですが、トレードオフの検討も重要だと感じました。

Github aws-iot-direct-messaging-demo

• 参照記事: Introducing AWS IoT Core's Direct Messaging
• ドキュメント: Direct Messaging - AWS IoT Core Developer Guide