Azure Digital Twinsの情報をSignalRを使ってリアルタイムにクライアント側への送信

実現したいこと

センサーから受信した値（温度・湿度・人間の座標情報・部屋の明るさなど）をDigital Twinを通してSignalRからリアルタイムにクライアント側へ送信する流れを作成します。

前提

アーキテクチャは説明しますが、インフラ構築は今回含めてませんので、あらかじめご了承ください...

アーキテクチャ

▪️ 構成図

---

▪️ データの流れ

① デバイス登録しているセンサーからのメッセージをIoTHubのDevice to Cloudで受け取ります。
　 また、クライアント側とSignalRを接続させるためにFunctionsに接続のリクエストを送ります。
② IoTHubがメッセージを受信したことをEvent Hubが検知します。
③ Event Hubトリガーが動いて、Azure Functionsを実行させます。
④ Azure FunctionsはIoTHubが受信したメッセージを見て、Digital Twins内の対象ツインのプロパティを更新します。
⑤ 別のEvent HubがDigital Twinsのツインが更新されたことを検知します。
⑥ Event Hubトリガーが動いて、Azure Functionsを実行させます。
⑦ Functionsは更新されたツインのプロパティをSignalRに渡します。
⑧ SignalRはFunctionsからツインの更新プロパティを受け取ったので、接続しているクライアントに送信します。

---

▪️ 使うリソース

1. Azure Functions

こちらは
・Event Hubトリガー二つ
・Httpトリガー一つ

計三つのFunctionを作成します。

2. IoTHub

センサーの管理、メッセージの受信などに利用します。

3. Azure Digital Twins

センサーから受信した情報（温度・湿度・明るさ・人の座標情報など）を管理します。

4. Event Hub

IoTHubがセンサーから受信したことを検知してFunctionsを実行させるために使用します。

5. SignalR

リアルタイム通信を簡単に構築できるサービスです。

実装

1 . HttpトリガーFunction（クライアント→SinglaR）

こちらのFunctionの名前は「negociate」にするようにしてください。

__init__.py

import logging

import azure.functions as func


def main(req: func.HttpRequest, connectionInfo) -> func.HttpResponse:
    logging.info("SignalRへの接続リクエスト受信")
    logging.info(f"connectionInfo: {connectionInfo}")
    return func.HttpResponse(connectionInfo)

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": ["post"]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    },
    {
      "type": "signalRConnectionInfo",
      "name": "connectionInfo",
      "hubName": "serverless",
      "connectionStringSetting": "<SignalRのエンドポイントが格納された環境変数>",
      "direction": "in"
    }
  ]
}

クライアント側では画面描画時もしくは任意のタイミングでリアルタイム通信を行いたいタイミングで、こちらのHttpトリガーFunctionにリクエストを実行します。

---

2 . EvetHubトリガーFunction（ADT→SingalR→クライアント）

こちらのFunctionの名前は何でも大丈夫です。

__init__.py

import json
import logging

import azure.functions as func
import requests

etag = ""


def main(event: func.EventHubEvent, signalRMessages: func.Out[str]):
    logging.info("SignalR Functionsを実行")
    # 引数のeventからメッセージ情報を抜き出す
    message = event.get_body().decode()
    logging.info(f"取得メッセージ: {message}")

    global etag
    headers = {"User-Agent": "serverless", "If-None-Match": etag}
    res = requests.get("https://api.github.com/repos/azure/azure-signalr", headers=headers)
    if res.headers.get("ETag"):
        etag = res.headers.get("ETag")

    logging.info("twinの更新情報を送信")
    signalRMessages.set(json.dumps({"target": "twinUpdate", "arguments": [message]}))

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "event",
      "direction": "in",
      "eventHubName": "<EventHubのリソース名>",
      "connection": "<EventHubの接続文字列が格納された環境変数>",
      "cardinality": "one",
      "consumerGroup": "$Default"
    },
    {
      "type": "signalR",
      "name": "signalRMessages",
      "hubName": "serverless",
      "connectionStringSetting": "<SignalRのエンドポイントが格納された環境変数>",
      "direction": "out"
    }
  ]
}

Azure Digital Twinのツインプロパティが更新されるとEventHubが検知し、Functionが実行されるようにします。
Function内ではツインプロパティの更新内容を取得して、function.json内で定義しているSignalRのエンドポイントにメッセージとして送信します。
既にクライアントと疎通されているSignalRはFunctionを経由して送られてきたメッセージを、クライアント側へ送ります。

3 . クライアント側（SignalR→Next.js製アプリ）

クライアント側は何でも良いですが、今回はNext.jsで作成したプロジェクトにします。

signalr.tsx

import React, { useEffect, useState } from "react";
import { HubConnection, HubConnectionBuilder } from "@microsoft/signalr";

const ConnectionPage = () => {
  const [signalRConnection, setSignalRConnection] = useState<HubConnection | null>(null);
  const [twinUpdateMessage, setTwinUpdateMessage] = useState<string>("");

  useEffect(() => {
    let ignore = false;

    const connection = new HubConnectionBuilder()
      .withUrl("<作成したFunctionのエンドポイント>/api")
      .withAutomaticReconnect()
      .build();

    connection
      .start()
      .then(() => {
        // 複数回レンダリングされても大丈夫のように対策（react.strictmodeのやつ）
        if (!ignore) {
          setSignalRConnection(connection);
        }
      })
      .catch((error) => console.log(error));

    return () => {
      ignore = true;
    };
  }, []);

  useEffect(() => {
    if (signalRConnection) {
      // twinUpdateというメッセージIDの時に受信する
      signalRConnection.on("twinUpdate", (message: string) => {
        console.log("twinUpdateMessage");
        setTwinUpdateMessage(message);
      });
    }
  }, [signalRConnection]);

  return (
    <div>
      <p style={{ marginBottom: 20 }}>
        SignalR is {signalRConnection ? "Connected!" : "Disconnected.."}
　　　　　　　　　　　</p>
      <div>
        <p>Message from SignalR</p>
        <p>{twinUpdateMessage}</p>
      </div>
    </div>
  );
};

export default ConnectionPage;

上手くいくか検証

HttpトリガーFunction（SignalR認証用）

クライアント側からのリクエストを受け付けられてます。

EventHubトリガーFunction(ADT→クライアント用)

ツインプロパティを更新されたタイミングでEventHubトリガーが動いて、Functionが実行できてます。

クライアント側

接続できた

メッセージも取得できました

最後に

EventHubやFunctionを経由するので、ちょっと大変でした。
インフラの構築も備忘録がてら書きたい