From 2af87a760d7a6448be01e5500c2c01010a2adcac Mon Sep 17 00:00:00 2001 From: "Timo K." Date: Fri, 2 Oct 2026 22:33:42 +0200 Subject: [PATCH] add support for sending data to the rtcSdk --- playwright/sdk/data.spec.ts | 41 ++++++++ sdk/SdkArchitecture.md | 29 ++++++ sdk/SdkMigration.md | 21 ++-- sdk/dev/index.html | 6 ++ sdk/dev/main.ts | 30 ++++++ sdk/src/api.ts | 22 ++++ sdk/src/session/DataChannel.test.ts | 149 ++++++++++++++++++++++++++++ sdk/src/session/DataChannel.ts | 118 ++++++++++++++++++++++ sdk/src/session/MatrixRTCClient.ts | 11 ++ 9 files changed, 420 insertions(+), 7 deletions(-) create mode 100644 playwright/sdk/data.spec.ts create mode 100644 sdk/src/session/DataChannel.test.ts create mode 100644 sdk/src/session/DataChannel.ts diff --git a/playwright/sdk/data.spec.ts b/playwright/sdk/data.spec.ts new file mode 100644 index 000000000..f29d3b140 --- /dev/null +++ b/playwright/sdk/data.spec.ts @@ -0,0 +1,41 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { expect, test } from "@playwright/test"; + +import { startPair, tileOf } from "./harness"; + +test.describe.configure({ timeout: 300_000 }); + +test("a message sent beside the media reaches the peer, and only the peer", async ({ + browser, +}) => { + const { pages, users } = await startPair(browser, "sdkdata"); + const [sender, receiver] = pages; + // Both are in the call before anything is sent, so the receiver knows the + // sender as a member and does not drop the message + for (const page of pages) + await expect(page.getByTestId("member")).toHaveCount(2, { + timeout: 60_000, + }); + await expect(tileOf(receiver, users[0])).toBeVisible(); + + await sender.getByPlaceholder("message").fill("hello from the SDK"); + await sender.getByRole("button", { name: "Send" }).click(); + + const message = receiver.getByTestId("message"); + await expect(message).toHaveText( + `${users[0].displayName}: hello from the SDK`, + { + timeout: 30_000, + }, + ); + await expect(message).toHaveAttribute("data-topic", "chat"); + await expect(message).toHaveAttribute("data-user-id", users[0].userId); + // Our own messages are not echoed back + await expect(sender.getByTestId("message")).toHaveCount(0); +}); diff --git a/sdk/SdkArchitecture.md b/sdk/SdkArchitecture.md index 5a4476796..aace36b8a 100644 --- a/sdk/SdkArchitecture.md +++ b/sdk/SdkArchitecture.md @@ -301,6 +301,28 @@ export interface MatrixRTCClient { * needs. */ connectedTransports$: Behavior; + + /** + * Sends a short text to every member on the local transport, over a + * reliable data channel beside the media. A packet holds about 15 KiB; + * anything larger belongs in a room event. Rejects while the local + * transport is not connected. Encrypted on the wire like the media, but + * not end to end with the media key. + */ + sendData(topic: string, text: string): Promise; + /** + * What remote members sent with `sendData`, on every transport the client + * is connected to. A message from an identity that is not a member is + * dropped, so a host only ever hears from attested members. + */ + data$: Observable; +} + +/** One `sendData` call as it arrives at a remote member. */ +export interface DataMessage { + member: RemoteRTCMember; + topic: string; + text: string; } ``` @@ -603,6 +625,13 @@ public API only. - **Developer panel.** `resolved$` on each connected transport covers what the panel shows today. Anything beyond that (LiveKit room state, connection quality) needs an opaque `debug$` per transport. +- **Data beside the media.** `sendData`/`data$` ride on LiveKit's reliable data + packets: topic-agnostic, matched to members by identity, received on every + connection but sent only on the local transport, so a member on another + transport does not hear us. They are transport-encrypted, not end to end: + LiveKit can encrypt data packets with the media key, but that is not wired + through the key provider yet. A host that needs either reach or end-to-end + encryption uses room events. - **Unencrypted rooms with per-participant keys.** Keys only reach devices the crypto tracks, which are the members of encrypted rooms. If the SDK is to work in an unencrypted room, tracking the members' devices becomes its job. diff --git a/sdk/SdkMigration.md b/sdk/SdkMigration.md index d2b2ea2ae..90c5a9c58 100644 --- a/sdk/SdkMigration.md +++ b/sdk/SdkMigration.md @@ -8,12 +8,13 @@ symbols, so it ages with the code; the architecture document should not. ## Status -| Slice | State | -| ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------- | -| 1. Scaffold: `sdk/` packaging, the interface, the harness, the smoke test | done, October 2026 | -| 2. First implementation of `createMatrixRTCClient` under `sdk/src`, the media adapter, the e2e suite | done, 1 October 2026 | -| 3. Copy the shared modules into `sdk/src/` with their tests, drop every `src/` import, turn on the import boundary | done, 2 October 2026 | -| 4. `createCallViewModel$` consumes the client; media view models take `media$`; Element Call imports the primitives and enums from the SDK and deletes its own copies; the old demo target is removed | done, 2 October 2026 | +| Slice | State | +| ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------- | +| 1. Scaffold: `sdk/` packaging, the interface, the harness, the smoke test | done, October 2026 | +| 2. First implementation of `createMatrixRTCClient` under `sdk/src`, the media adapter, the e2e suite | done, 1 October 2026 | +| 3. Copy the shared modules into `sdk/src/` with their tests, drop every `src/` import, turn on the import boundary | done, 2 October 2026 | +| 4. `createCallViewModel$` consumes the client; media view models take `media$`; Element Call imports the primitives and enums from the SDK and deletes its own copies | done, 2 October 2026 | +| 5. `sendData`/`data$` on the client; the legacy bundle `sdk-target-based-on-call-view-model/` is rebuilt on the SDK with its interface unchanged, since it is in production use | done, 2 October 2026 | Element Call now consumes the SDK. `InCallView` creates the client from the configuration, settings and URL parameters, and hands it to `createCallViewModel$`, @@ -336,4 +337,10 @@ document, "Vocabulary") settles the first three; the js-sdk rename is still open `vite-sdk.config.ts` (`pnpm build:sdk`), `vite-component.config.ts`, `vite-embedded.config.ts`, `vite.config.ts`. The old SDK target (`sdk-target-based-on-call-view-model/`) is the one whose dependency direction the -migration flips; it goes in slice 5. +migration flips. It is in production use, so slice 5 rebuilt it on the SDK instead of +removing it: `createMatrixRTCSdk` keeps its signature, `join`/`leave`/`stop`, +`connected$`, `data$` (`{ rtcBackendIdentity, data }` on the `matrixRTC` topic), +`sendData`, `sendRoomMessage`, `localMember$` and `remoteMembers$` keep their shapes. +The one break is inside the member objects: `connection` and `participant` are +always `null` now, because the SDK does not hand out LiveKit objects; `membership` +is unchanged. The bundle is removed once its consumers use the package directly. diff --git a/sdk/dev/index.html b/sdk/dev/index.html index b0de49e90..37153285e 100644 --- a/sdk/dev/index.html +++ b/sdk/dev/index.html @@ -63,6 +63,12 @@ Please see LICENSE in the repository root for full details.

+ + +
    diff --git a/sdk/dev/main.ts b/sdk/dev/main.ts index b608d301e..d28cc468b 100644 --- a/sdk/dev/main.ts +++ b/sdk/dev/main.ts @@ -29,6 +29,8 @@ import { createSession, joinRoom, type Login } from "./session"; const form = document.querySelector("form")!; const status = document.getElementById("status")!; const members = document.getElementById("members")!; +const dataForm = document.getElementById("data") as HTMLFormElement; +const messages = document.getElementById("messages")!; const buttons = { microphone: document.getElementById("microphone") as HTMLButtonElement, camera: document.getElementById("camera") as HTMLButtonElement, @@ -89,6 +91,7 @@ async function start( if (error !== null) status.textContent = `Error: ${error.message}`; }); showMembers(scope, rtcClient); + showMessages(scope, rtcClient); rtcClient.join(); toggle(buttons.microphone, async (enabled) => @@ -102,6 +105,8 @@ async function start( rtcClient.leave(); scope.end(); members.replaceChildren(); + messages.replaceChildren(); + dataForm.hidden = true; for (const button of Object.values(buttons)) button.hidden = true; status.textContent = "Left"; }; @@ -124,6 +129,31 @@ function toggle( }; } +/** Sends what the form holds on the "chat" topic, and lists what arrives. */ +function showMessages( + scope: ObservableScope, + rtcClient: MatrixRTCClient, +): void { + dataForm.hidden = false; + dataForm.onsubmit = (event): void => { + event.preventDefault(); + const text = new FormData(dataForm).get("text") as string; + rtcClient.sendData("chat", text).then( + () => dataForm.reset(), + (e: unknown) => { + status.textContent = `Error: ${e}`; + }, + ); + }; + rtcClient.data$.pipe(scope.bind()).subscribe(({ member, topic, text }) => { + const line = messages.appendChild(document.createElement("li")); + line.dataset.testid = "message"; + line.dataset.topic = topic; + line.dataset.userId = member.userId; + line.textContent = `${member.displayName$.value}: ${text}`; + }); +} + function showMembers(scope: ObservableScope, rtcClient: MatrixRTCClient): void { const tiles = new Map(); combineLatest([rtcClient.localMember$, rtcClient.remoteMembers$]) diff --git a/sdk/src/api.ts b/sdk/src/api.ts index a5e81e264..ba571c928 100644 --- a/sdk/src/api.ts +++ b/sdk/src/api.ts @@ -153,6 +153,28 @@ export interface MatrixRTCClient { /** Transports the session currently holds a live connection to. */ connectedTransports$: Behavior; + + /** + * Sends a short text to every member on the local transport, over a + * reliable data channel beside the media. A packet holds about 15 KiB; + * anything larger belongs in a room event. Rejects while the local + * transport is not connected. Encrypted on the wire like the media, but + * not end to end with the media key. + */ + sendData(topic: string, text: string): Promise; + /** + * What remote members sent with `sendData`, on every transport the client + * is connected to. A message from an identity that is not a member is + * dropped, so a host only ever hears from attested members. + */ + data$: Observable; +} + +/** One `sendData` call as it arrives at a remote member. */ +export interface DataMessage { + member: RemoteRTCMember; + topic: string; + text: string; } // --------------------------------------------------------------------------- diff --git a/sdk/src/session/DataChannel.test.ts b/sdk/src/session/DataChannel.test.ts new file mode 100644 index 000000000..eeb291f43 --- /dev/null +++ b/sdk/src/session/DataChannel.test.ts @@ -0,0 +1,149 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { EventEmitter } from "events"; +import { type RemoteParticipant, RoomEvent } from "livekit-client"; +import { logger } from "matrix-js-sdk/lib/logger"; +import { BehaviorSubject } from "rxjs"; +import { describe, expect, it, vi } from "vitest"; + +import { type DataMessage, type RemoteRTCMember } from "../api"; +import { Epoch } from "../reactive/ObservableScope"; +import { constant } from "../reactive/Behavior"; +import { type Connection, ConnectionState } from "./Connection"; +import { ConnectionManagerData } from "./ConnectionManager"; +import { createDataChannel$ } from "./DataChannel"; +import { exampleTransport, testScope } from "../utils/test"; + +const encode = (text: string): Uint8Array => new TextEncoder().encode(text); + +function fakeConnection( + url: string, + state = ConnectionState.LivekitConnected, +): { + connection: Connection; + room: EventEmitter; + publishData: ReturnType; +} { + const room = new EventEmitter(); + const publishData = vi.fn(async () => Promise.resolve()); + const connection = { + transport: { ...exampleTransport, livekit_service_url: url }, + state$: constant(state), + livekitRoom: Object.assign(room, { localParticipant: { publishData } }), + } as unknown as Connection; + return { connection, room, publishData }; +} + +function member(id: string): RemoteRTCMember { + return { id, userId: `@${id}:example.org` } as RemoteRTCMember; +} + +function setup( + connections: Connection[], + members: RemoteRTCMember[] = [member("alice")], +): { + received: DataMessage[]; + connections$: BehaviorSubject>; + sendData: (topic: string, text: string) => Promise; +} { + const data = new ConnectionManagerData(); + for (const connection of connections) data.add(connection, []); + const connections$ = new BehaviorSubject(new Epoch(data)); + const { data$, sendData } = createDataChannel$({ + scope: testScope(), + connectionManager: { connectionManagerData$: connections$ }, + remoteMembers$: constant(members), + connection$: constant(connections[0] ?? null), + logger, + }); + const received: DataMessage[] = []; + data$.subscribe((message) => received.push(message)); + return { received, connections$, sendData }; +} + +describe("data$", () => { + it("delivers a packet from a member with its topic and text", () => { + const { connection, room } = fakeConnection("https://sfu"); + const { received } = setup([connection]); + + room.emit( + RoomEvent.DataReceived, + encode("hello"), + { identity: "alice" } as RemoteParticipant, + undefined, + "chat", + ); + + expect(received).toEqual([ + { member: member("alice"), topic: "chat", text: "hello" }, + ]); + }); + + it("drops packets from the server and from identities that are not members", () => { + const { connection, room } = fakeConnection("https://sfu"); + const { received } = setup([connection]); + + room.emit(RoomEvent.DataReceived, encode("server"), undefined); + room.emit(RoomEvent.DataReceived, encode("stranger"), { + identity: "mallory", + } as RemoteParticipant); + + expect(received).toEqual([]); + }); + + it("listens on every connection, including ones added later", () => { + const first = fakeConnection("https://sfu-1"); + const second = fakeConnection("https://sfu-2"); + const { received, connections$ } = setup( + [first.connection], + [member("alice"), member("bob")], + ); + + const data = new ConnectionManagerData(); + data.add(first.connection, []); + data.add(second.connection, []); + connections$.next(new Epoch(data, 1)); + + first.room.emit(RoomEvent.DataReceived, encode("one"), { + identity: "alice", + } as RemoteParticipant); + second.room.emit(RoomEvent.DataReceived, encode("two"), { + identity: "bob", + } as RemoteParticipant); + + expect(received.map((m) => [m.member.id, m.text])).toEqual([ + ["alice", "one"], + ["bob", "two"], + ]); + }); +}); + +describe("sendData", () => { + it("publishes the text as a reliable packet on the local connection", async () => { + const { connection, publishData } = fakeConnection("https://sfu"); + const { sendData } = setup([connection]); + + await sendData("chat", "hello"); + + expect(publishData).toHaveBeenCalledWith(encode("hello"), { + reliable: true, + topic: "chat", + }); + }); + + it("rejects while the local transport is not connected", async () => { + const { connection, publishData } = fakeConnection( + "https://sfu", + ConnectionState.Initialized, + ); + const { sendData } = setup([connection]); + + await expect(sendData("chat", "hello")).rejects.toThrow("not connected"); + expect(publishData).not.toHaveBeenCalled(); + }); +}); diff --git a/sdk/src/session/DataChannel.ts b/sdk/src/session/DataChannel.ts new file mode 100644 index 000000000..4aba2bdb1 --- /dev/null +++ b/sdk/src/session/DataChannel.ts @@ -0,0 +1,118 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { + type RemoteParticipant, + type Room as LivekitRoom, + RoomEvent, +} from "livekit-client"; +import { type Logger } from "matrix-js-sdk/lib/logger"; +import { + distinctUntilChanged, + filter, + fromEvent, + map, + merge, + type Observable, + share, + switchMap, +} from "rxjs"; + +import { type DataMessage, type RemoteRTCMember } from "../api"; +import { type Behavior } from "../reactive/Behavior"; +import { type ObservableScope } from "../reactive/ObservableScope"; +import { type Connection, ConnectionState } from "./Connection"; +import { type IConnectionManager } from "./ConnectionManager"; + +interface Props { + scope: ObservableScope; + connectionManager: IConnectionManager; + /** Used to resolve the sender of each message, by its transport identity. */ + remoteMembers$: Behavior; + /** The local member's connection, which carries what we send. */ + connection$: Behavior; + logger: Logger; +} + +export interface DataChannel { + data$: Observable; + sendData: (topic: string, text: string) => Promise; +} + +/** + * A text channel beside the media: LiveKit's reliable data packets, received + * on every connection the client holds and sent on the local member's. The + * sender is matched to a member by its identity, so the packet's own claim of + * who sent it never reaches the host. + */ +export function createDataChannel$({ + scope, + connectionManager, + remoteMembers$, + connection$, + logger, +}: Props): DataChannel { + const decoder = new TextDecoder(); + const encoder = new TextEncoder(); + + const rooms$ = connectionManager.connectionManagerData$.pipe( + map(({ value }) => value.getConnections().map((c) => c.livekitRoom)), + distinctUntilChanged( + (a, b) => a.length === b.length && a.every((room, i) => room === b[i]), + ), + ); + + const data$ = rooms$.pipe( + switchMap((rooms) => merge(...rooms.map(receivedOn))), + map(({ payload, participant, topic }): DataMessage | null => { + // Only the server sends without a participant; a host has no use for it + if (participant === undefined) return null; + const member = remoteMembers$.value.find( + (m) => m.id === participant.identity, + ); + if (member === undefined) { + logger.warn(`Dropping data from ${participant.identity}: not a member`); + return null; + } + return { member, topic: topic ?? "", text: decoder.decode(payload) }; + }), + filter((message) => message !== null), + scope.bind(), + share(), + ); + + const sendData = async (topic: string, text: string): Promise => { + const connection = connection$.value; + if (connection?.state$.value !== ConnectionState.LivekitConnected) + throw new Error("Cannot send data: the local transport is not connected"); + await connection.livekitRoom.localParticipant.publishData( + encoder.encode(text), + { reliable: true, topic }, + ); + }; + + return { data$, sendData }; +} + +interface Received { + payload: Uint8Array; + participant: RemoteParticipant | undefined; + topic: string | undefined; +} + +function receivedOn(room: LivekitRoom): Observable { + return fromEvent( + room, + RoomEvent.DataReceived, + ( + payload: Uint8Array, + participant?: RemoteParticipant, + _kind?: unknown, + topic?: string, + ): Received => ({ payload, participant, topic }), + ); +} diff --git a/sdk/src/session/MatrixRTCClient.ts b/sdk/src/session/MatrixRTCClient.ts index b4e877684..d76cf409f 100644 --- a/sdk/src/session/MatrixRTCClient.ts +++ b/sdk/src/session/MatrixRTCClient.ts @@ -53,6 +53,7 @@ import { membershipKeys, } from "./Members"; import { createConnectionManager$ } from "./ConnectionManager"; +import { createDataChannel$ } from "./DataChannel"; import { type DesiredMedia, Publisher } from "./Publisher"; import { createKeyRotationSuppressed$, @@ -248,6 +249,14 @@ export function createMatrixRTCClient( ).pipe(map((member) => member ?? null)), ); + const { data$, sendData } = createDataChannel$({ + scope, + connectionManager, + remoteMembers$, + connection$: localMembership.connection$, + logger, + }); + const status$ = scope.behavior( combineLatest( [ @@ -280,6 +289,8 @@ export function createMatrixRTCClient( ), keyRotationSuppressed$: createKeyRotationSuppressed$(scope, jsSdkSession), connectedTransports$: transports.connected$, + sendData, + data$, }; }