DataChannels

Realtime SFU DataChannels を使うと、WebRTC 経由で低遅延のアプリケーションデータを送信できます。よくあるペイロードは、チャットメッセージ、ゲームの状態、センサーの更新、制御イベントです。

音声と映像は DataChannels ではなく、Realtime SFU のメディアトラックで送信します。

graph LR
    A[パブリッシャー] -->|アプリケーションデータ| B[Cloudflare Realtime SFU]
    B -->|アプリケーションデータ| C@{ shape: procs, label: "サブスクライバー"}

各パブリッシャーは、名前付きの DataChannel を複数のサブスクライバーに送信できます。デフォルトでは、メッセージはパブリッシャーからサブスクライバーへ流れます。

DataChannels は、デフォルトで信頼性が高く順序付きの配信を使います。ゲームの状態やライブセンサー更新のように、遅延したデータより最新のデータが重要な場合は、部分的な信頼性や順序なし配信を選びます。

HTTPS API で DataChannel を作成するときに、次のオプションフィールドを設定します。

• ordered (boolean、デフォルト true): false にすると、メッセージが順不同で到着することを許可します。遅延したメッセージが、後続のメッセージをブロックしません。
• maxRetransmits (integer): 初回送信後の再送回数を制限します。再送なしにする場合は 0 を設定し、再送回数の上限なしにする場合は省略します。
• maxPacketLifeTime (integer): トランスポートが配信を試みる時間をミリ秒単位で制限します。寿命の上限なしにする場合は省略します。

maxRetransmits と maxPacketLifeTime は同時に使えません。同じチャネルで両方を設定しないでください。

順序と再試行の動作は独立しています。信頼性が高く順序なしの配信にするには、ordered: false を設定し、maxRetransmits と maxPacketLifeTime の両方を省略します。メッセージは順不同で到着することがありますが、トランスポートは失敗した配信の再試行を続けます。

パブリッシャー（location: "local"）、各サブスクライバー（location: "remote"）、各クライアントの createDataChannel() 呼び出しで、同じ値を使います。Realtime DataChannels はネゴシエート済み ID を使うため、ブラウザーはリモートピアからこれらの設定を受け取りません。

信頼性が低く順序なしのパブリッシャーチャネルを作成します。

{
	"dataChannels": [
		{
			"location": "local",
			"dataChannelName": "player-state",
			"ordered": false,
			"maxRetransmits": 0
		}
	]
}

次に、同じ信頼性フィールドで、サブスクライバー側に対応するリモートチャネルを作成します。

{
	"dataChannels": [
		{
			"location": "remote",
			"sessionId": "<PUBLISHER_SESSION_ID>",
			"dataChannelName": "player-state",
			"ordered": false,
			"maxRetransmits": 0
		}
	]
}

同じ設定で、ブラウザー側の対応する DataChannel を作成します。この例では、pc はアクティブな RTCPeerConnection、resp はそのチャネルの API レスポンスです。

const dc = pc.createDataChannel("player-state", {
	negotiated: true,
	id: resp.dataChannels[0].id,
	ordered: false,
	maxRetransmits: 0,
});

部分的な信頼性にする場合は、ペイロードが有効でいられる時間に応じて、再送回数の上限かパケット寿命を選びます。

リモート DataChannel で waitForAck: true を設定すると、サブスクライバーが準備完了を通知するまで配信を遅らせます。

• waitForAck は location: "remote" の DataChannels にのみ適用され、デフォルトは false です。
• ゲートが閉じているあいだ、SFU はそのサブスクライバーへの配信を保留します。
• DataChannel が開いたあと、サブスクライバーは "ack" などの任意のメッセージを送信します。SFU はこの最初のメッセージを消費し、ゲートを開いて、パブリッシャーのメッセージの転送を開始します。
• 確認応答は、リモート DataChannel を作成してから 30 秒以内に SFU に届く必要があります。届かない場合、SFU はゲート付きチャネルを破棄します。再試行するには、リモート DataChannel を再度作成します。

canReply がない場合、それ以降のサブスクライバーメッセージはパブリッシャーに転送されません。

サブスクライバーセッションで POST /apps/{appId}/sessions/{sessionId}/datachannels/new を呼び出し、ゲートを有効にしたリモート DataChannel を作成します。

{
	"dataChannels": [
		{
			"location": "remote",
			"sessionId": "<PUBLISHER_SESSION_ID>",
			"dataChannelName": "my-channel",
			"waitForAck": true
		}
	]
}

次に、サブスクライバー側で、DataChannel が開いたら確認応答を送信します。この例では、API_BASE、headers、pc を初期化済みで、waitForOpen() ヘルパーを定義済みであるとします。

const response = await fetch(
	`${API_BASE}/sessions/${subscriberId}/datachannels/new`,
	{
		method: "POST",
		headers,
		body: JSON.stringify({
			dataChannels: [
				{
					location: "remote",
					sessionId: publisherId,
					dataChannelName: "my-channel",
					waitForAck: true,
				},
			],
		}),
	},
);

if (!response.ok) {
	throw new Error(`Failed to create DataChannel: ${response.status}`);
}

const resp = await response.json();
const channelId = resp.dataChannels?.[0]?.id;
if (channelId === undefined) {
	throw new Error("DataChannel response did not include an id");
}

const dc = pc.createDataChannel("my-channel-subscribed", {
	negotiated: true,
	id: channelId,
});

await waitForOpen(dc);
dc.send("ack"); // The first message opens the gate.

デフォルトでは、メッセージはパブリッシャーからサブスクライバーへ流れます。テレメトリを公開するデバイスにオペレーターが応答する場合など、同じチャネルで 1 人のサブスクライバーが応答する必要があるときは、canReply: true を設定します。

graph LR
    P[パブリッシャー] -->|パブリッシャーのメッセージ| SFU[Cloudflare Realtime SFU]
    SFU -->|パブリッシャーのメッセージ| S1[canReply 付きのサブスクライバー]
    SFU -->|パブリッシャーのメッセージ| S2[他のサブスクライバー]
    S1 -->|返信| SFU
    SFU -->|返信| P

canReply は返信アクセスを次のように制御します。

• canReply は location: "remote" の DataChannels にのみ適用され、デフォルトは false です。
• 各パブリッシャー DataChannel で返信アクセスを持てるサブスクライバーは、最大 1 人です。別のサブスクライバーにアクセスを付与すると、以前のサブスクライバーは置き換わります。
• SFU は、アクセスを持つサブスクライバーからの返信だけを転送します。
• 返信を受け取るのはパブリッシャーです。他のサブスクライバーには届きません。

サブスクライバーセッションで、canReply: true を指定してリモート DataChannel を作成します。

{
	"dataChannels": [
		{
			"location": "remote",
			"sessionId": "<PUBLISHER_SESSION_ID>",
			"dataChannelName": "my-channel",
			"canReply": true
		}
	]
}

フローの例:

リモート DataChannel を再作成せずに返信アクセスを変更するには、PUT /apps/{appId}/sessions/{subscriberSessionId}/datachannels/update を呼び出します。

{
	"dataChannels": [
		{
			"location": "remote",
			"sessionId": "<PUBLISHER_SESSION_ID>",
			"dataChannelName": "my-channel",
			"canReply": true
		}
	]
}

同じ本文で "canReply": false を指定すると、アクセスを取り消します。次の表はよくあるパターンです。

// The subscriber already pulled "my-channel" without canReply.
// Allow replies later.
const response = await fetch(
	`${API_BASE}/sessions/${subscriberId}/datachannels/update`,
	{
		method: "PUT",
		headers,
		body: JSON.stringify({
			dataChannels: [
				{
					location: "remote",
					sessionId: publisherId,
					dataChannelName: "my-channel",
					canReply: true,
				},
			],
		}),
	},
);

if (!response.ok) {
	throw new Error(`Failed to update DataChannel: ${response.status}`);
}

// The same negotiated DataChannel can now send replies to the publisher.
dc.send(JSON.stringify({ type: "reply", body: "pong" }));

同じリモート DataChannel で canReply と waitForAck の両方を設定できます。サブスクライバーの最初のメッセージは確認応答ゲートを開き、転送されません。そのあと、そのサブスクライバーが返信アクセスを持っているあいだ、以降のサブスクライバーメッセージはパブリッシャーに転送されます。

トランスポート、公開、購読の一連のセットアップは、DataChannel echo の例 ↗ を確認してください。

この例は、ローカルテスト用にアプリトークンをブラウザーのコードに置いています。本番環境では、トークンはバックエンドに置いてください。