mirror of
https://github.com/vector-im/element-call.git
synced 2026-11-10 01:02:27 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e2e504d307 | ||
|
|
de16cb4e37 |
@@ -11,18 +11,25 @@ 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 ({
|
||||
test("a message sent on a data track 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
|
||||
// sender as a member
|
||||
for (const page of pages)
|
||||
await expect(page.getByTestId("member")).toHaveCount(2, {
|
||||
timeout: 60_000,
|
||||
});
|
||||
await expect(tileOf(receiver, users[0])).toBeVisible();
|
||||
// The receiver listens on the sender's track, published with the join
|
||||
await expect(tileOf(receiver, users[0])).toHaveAttribute(
|
||||
"data-chat",
|
||||
"true",
|
||||
{
|
||||
timeout: 30_000,
|
||||
},
|
||||
);
|
||||
|
||||
await sender.getByPlaceholder("message").fill("hello from the SDK");
|
||||
await sender.getByRole("button", { name: "Send" }).click();
|
||||
@@ -34,7 +41,9 @@ test("a message sent beside the media reaches the peer, and only the peer", asyn
|
||||
timeout: 30_000,
|
||||
},
|
||||
);
|
||||
await expect(message).toHaveAttribute("data-topic", "chat");
|
||||
await expect(message).toHaveAttribute("data-name", "chat");
|
||||
// Decrypted with the sender's key, at the key index the stream carried
|
||||
await expect(message).toHaveAttribute("data-encrypted", "true");
|
||||
await expect(message).toHaveAttribute("data-user-id", users[0].userId);
|
||||
// Our own messages are not echoed back
|
||||
await expect(sender.getByTestId("message")).toHaveCount(0);
|
||||
|
||||
@@ -16,11 +16,25 @@ Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
|
||||
import { EMPTY, filter, map, type Observable, of, switchMap, take } from "rxjs";
|
||||
import {
|
||||
distinct,
|
||||
distinctUntilChanged,
|
||||
EMPTY,
|
||||
filter,
|
||||
map,
|
||||
mergeMap,
|
||||
type Observable,
|
||||
of,
|
||||
switchMap,
|
||||
take,
|
||||
takeUntil,
|
||||
} from "rxjs";
|
||||
import {
|
||||
type Behavior,
|
||||
createRTCSlot,
|
||||
type DataTrack,
|
||||
E2eeType,
|
||||
type LocalDataTrack,
|
||||
type LocalRTCMember,
|
||||
ObservableScope,
|
||||
type RemoteRTCMember,
|
||||
@@ -45,7 +59,7 @@ import { getUrlParams } from "../src/UrlParams";
|
||||
import { observeElementSize$ } from "../src/utils/elementSize";
|
||||
import { initializeWidget } from "../src/widget";
|
||||
|
||||
/** The data channel topic this bundle has always used, so that old and new builds talk. */
|
||||
/** The name of our data track: the topic this bundle has always used, so that old and new builds talk. */
|
||||
const TOPIC = "matrixRTC";
|
||||
|
||||
export interface MatrixRTCSdk {
|
||||
@@ -59,9 +73,9 @@ export interface MatrixRTCSdk {
|
||||
localMember$: Behavior<LocalRTCMember | null>;
|
||||
/** The other members of the slot, with the media the participation carries for them. */
|
||||
remoteMembers$: Behavior<RemoteRTCMember[]>;
|
||||
/** What other members sent with `sendData`. */
|
||||
/** What other members sent with `sendData`, on their data track. */
|
||||
data$: Observable<{ rtcBackendIdentity: string; data: string }>;
|
||||
/** Sends to every member on our transport. Rejects while not joined. */
|
||||
/** Sends on our data track, published with the first message. Rejects while not joined. */
|
||||
sendData: (data: unknown) => Promise<void>;
|
||||
sendRoomMessage: (message: string) => Promise<void>;
|
||||
}
|
||||
@@ -100,6 +114,9 @@ export async function createMatrixRTCSdk(
|
||||
id,
|
||||
});
|
||||
let vm: CallViewModel | undefined;
|
||||
let dataTrack: Promise<LocalDataTrack> | undefined;
|
||||
const encoder = new TextEncoder();
|
||||
const decoder = new TextDecoder();
|
||||
const failed =
|
||||
(what: string) =>
|
||||
(e: unknown): void =>
|
||||
@@ -115,6 +132,8 @@ export async function createMatrixRTCSdk(
|
||||
),
|
||||
);
|
||||
|
||||
const remoteMembers$ = over((p) => p.remoteMembers$, []);
|
||||
|
||||
return {
|
||||
join: (): void => {
|
||||
if (slot.participation$.value !== null) return;
|
||||
@@ -122,6 +141,7 @@ export async function createMatrixRTCSdk(
|
||||
widget.api
|
||||
.setAlwaysOnScreen(true)
|
||||
.catch(failed("Failed to make the widget sticky"));
|
||||
dataTrack = undefined;
|
||||
const participation = slot.join({
|
||||
...rtcParticipationOptions(urlParams),
|
||||
publish: initialPublishRequests(muteStates, mediaDevices, capture),
|
||||
@@ -169,20 +189,40 @@ export async function createMatrixRTCSdk(
|
||||
false,
|
||||
),
|
||||
localMember$: over((p) => p.localMember$, null),
|
||||
remoteMembers$: over((p) => p.remoteMembers$, []),
|
||||
// Attested members only: the SDK has matched the sender
|
||||
data$: slot.participation$.pipe(
|
||||
switchMap((p) => p?.data$ ?? EMPTY),
|
||||
filter(({ topic }) => topic === TOPIC),
|
||||
map(({ member, text }) => ({
|
||||
rtcBackendIdentity: member.rtcBackendIdentity,
|
||||
data: text,
|
||||
})),
|
||||
remoteMembers$,
|
||||
data$: remoteMembers$.pipe(
|
||||
// Each member once, for as long as it stays: a data track is only
|
||||
// subscribed while observed, so it is not dropped on every list change
|
||||
mergeMap((members) => members),
|
||||
distinct(),
|
||||
mergeMap((member) =>
|
||||
member.tracks$.pipe(
|
||||
map((tracks) =>
|
||||
tracks?.find(
|
||||
(t): t is DataTrack => t.kind === "data" && t.name === TOPIC,
|
||||
),
|
||||
),
|
||||
distinctUntilChanged(),
|
||||
switchMap((track) => track?.data$ ?? EMPTY),
|
||||
map((payload) => ({
|
||||
rtcBackendIdentity: member.rtcBackendIdentity,
|
||||
data: decoder.decode(payload),
|
||||
})),
|
||||
takeUntil(
|
||||
remoteMembers$.pipe(filter((members) => !members.includes(member))),
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
sendData: async (data): Promise<void> => {
|
||||
const participation = slot.participation$.value;
|
||||
if (participation === null) throw new Error("Not joined");
|
||||
await participation.sendData(TOPIC, JSON.stringify(data));
|
||||
dataTrack ??= participation.publish({ source: "data", name: TOPIC });
|
||||
const track = await dataTrack.catch((e: unknown) => {
|
||||
dataTrack = undefined;
|
||||
throw e;
|
||||
});
|
||||
await track.send(encoder.encode(JSON.stringify(data)));
|
||||
},
|
||||
sendRoomMessage: async (message): Promise<void> => {
|
||||
await client.sendTextMessage(room.roomId, message);
|
||||
|
||||
+93
-41
@@ -25,7 +25,7 @@ Call's `src/` to get there is listed in [`SdkMigration.md`](./SdkMigration.md).
|
||||
│ │
|
||||
│ public API, all Behaviors: │
|
||||
│ join() leave() state$ │
|
||||
│ localMember$ remoteMembers$ memberCount$ sendData() data$ │
|
||||
│ localMember$ remoteMembers$ memberCount$ publish(data) │
|
||||
│ keyRotationSuppressed$ connectedTransports$ │
|
||||
│ │
|
||||
│ per member RTCMember { id userId deviceId membership$ displayName$ │
|
||||
@@ -48,7 +48,7 @@ Call's `src/` to get there is listed in [`SdkMigration.md`](./SdkMigration.md).
|
||||
│ │ (LocalMembership) │ local.publish, │ Publisher: local tracks, │ │
|
||||
│ │ transport registry │ setPublishing▶│ upstream paused until │ │
|
||||
│ │ (TransportMetadata) │ ◀─ connections$ │ joined │ │
|
||||
│ │ data packet → member │ ◀─ data$ │ LivekitMemberMedia: │ │
|
||||
│ │ │ │ LivekitMemberMedia: │ │
|
||||
│ │ │ │ participant → tracks │ │
|
||||
│ └───────────────────────────┘ └───────────────────────────┘ │
|
||||
│ │
|
||||
@@ -146,12 +146,12 @@ the sections below still describe `MatrixRTCClient` until that lands.
|
||||
| Word | Meaning here | Elsewhere |
|
||||
| --------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------- |
|
||||
| slot | an `RTCSlot`: the namespace in a room that memberships join into, `application` plus `id`. It exists while the room does, with nobody in it too. Holds `members$`, its open or closed `status$`, and `join()` | the js-sdk `MatrixRTCSession` is one per slot; the MSC4143 slot event |
|
||||
| participation | an `RTCParticipation`: the span between our `join()` of a slot and our `leave()`. Holds the connection and its `state$`, `localMember$`, `remoteMembers$`, the connected transports and the data channel. Made by `slot.join()`, never reused | "session" is not used here, because in the js-sdk it means the slot |
|
||||
| participation | an `RTCParticipation`: the span between our `join()` of a slot and our `leave()`. Holds the connection and its `state$`, `localMember$`, `remoteMembers$`, and the connected transports. Made by `slot.join()`, never reused | "session" is not used here, because in the js-sdk it means the slot |
|
||||
| member | an `RTCMember`: one membership in a slot with what the SDK derives from it, `rtcBackendIdentity`, `userId`, `deviceId`, `memberId`, `displayName$`, `avatarUrl$`, `transport$` and `applicationData$`. No media. What the slot hands out in `members$` | the js-sdk `CallMembership` behind it, which never leaves the SDK; Element Call's old "matrixLivekitMember" |
|
||||
| local member, remote member | `LocalRTCMember` and `RemoteRTCMember`: an `RTCMember` plus its `MemberMedia`, which only a participation can carry. The local one's tracks carry the controls. What the participation hands out in `localMember$` and `remoteMembers$` | LiveKit `LocalParticipant` and `RemoteParticipant`, which never leave the SDK |
|
||||
| media | a `MemberMedia`: what a member sends once it has arrived on its transport, `tracks$`, null while waiting, and `encryptionError$` beside it. The media backend supplies exactly these two fields per member | Element Call's `MediaViewModel` is a view of it |
|
||||
| track | a `MediaTrack`: one publication of a member, audio or video, with `source`, `kind`, `id` and the behaviors on it. `LocalAudioMediaTrack` and `LocalVideoMediaTrack` add the controls over our own | LiveKit `TrackPublication` and `Track`, which never leave the SDK |
|
||||
| publish request | a `PublishRequest`: a source, where to capture it from and how to encode it. Given at the join in `publish`, or later to `publish` on the participation | |
|
||||
| track | a `Track`: one publication of a member. A `MediaTrack` is audio or video, with `source`, `kind`, `id` and the behaviors on it; a `DataTrack` is a named channel of bytes. The `Local…` variants add the controls over our own, and `send` for data | LiveKit `TrackPublication` and `Track`, which never leave the SDK |
|
||||
| publish request | a `PublishRequest`: a source, where to capture it from and how to encode it. Given at the join in `publish`, or later to `publish` on the participation. A `DataPublishRequest` names a data track, and only goes to `publish` | |
|
||||
| attach, detach | handing a `<video>` or `<audio>` element to a track and taking it back. The SDK owns the stream and the observers on the element | LiveKit `Track.attach` |
|
||||
| transport | an `RTCTransport`: where media is exchanged, a LiveKit service url today. Carries the raw js-sdk `Transport` from the membership and, in `resolved$`, what the backend fetched to connect to it | MSC4143 "focus" |
|
||||
| media backend | a `MediaBackend`: what carries the media, behind `sdk/src/media-backend/api.ts`. One per participation, chosen with `backend` in the join options; LiveKit under `sdk/src/media-backend/livekit/` is the one that exists | "SFU", a LiveKit room, a peer connection |
|
||||
@@ -340,28 +340,6 @@ export interface MatrixRTCClient {
|
||||
* the browser's choice, never calls it.
|
||||
*/
|
||||
setAudioOutputDeviceId(deviceId: string): Promise<void>;
|
||||
|
||||
/**
|
||||
* 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<void>;
|
||||
/**
|
||||
* 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<DataMessage>;
|
||||
}
|
||||
|
||||
/** One `sendData` call as it arrives at a remote member. */
|
||||
export interface DataMessage {
|
||||
member: RemoteRTCMember;
|
||||
topic: string;
|
||||
text: string;
|
||||
}
|
||||
```
|
||||
|
||||
@@ -452,11 +430,15 @@ export interface LocalRTCMember extends RTCMember {
|
||||
* user closing the picker. One publication per source: publishing a source
|
||||
* again unmutes it.
|
||||
*/
|
||||
publish(request: DataPublishRequest): Promise<LocalDataTrack>;
|
||||
publish(
|
||||
request: PublishRequest,
|
||||
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack>;
|
||||
/** Removes one of our tracks; a screen share takes its audio with it. */
|
||||
unpublish(id: string): Promise<void>;
|
||||
/**
|
||||
* Removes one of our tracks, as `publish` handed it out; a screen share
|
||||
* takes its audio with it. The track's controls and `send` reject from then on.
|
||||
*/
|
||||
unpublish(track: LocalTrack): Promise<void>;
|
||||
}
|
||||
```
|
||||
|
||||
@@ -557,8 +539,6 @@ export interface MediaBackend {
|
||||
membership$: Behavior<RTCMembership>,
|
||||
): MemberMediaFields;
|
||||
readonly connections$: Behavior<BackendConnection[]>; // transport, state, resolved; diagnostics
|
||||
sendData(topic: string, text: string): Promise<void>;
|
||||
readonly data$: Observable<DataPacket>;
|
||||
}
|
||||
```
|
||||
|
||||
@@ -568,8 +548,7 @@ LiveKit backend only, a last resort of another member's SFU that lets us
|
||||
publish (`findRemotePublishLivekitTransport`, wired in by `RTCParticipation` rather
|
||||
than by the backend interface, since it is an MSC4195 matter), the membership
|
||||
state machine in `LocalMembership.ts` and its combination with
|
||||
`local.connectionState$` into `state$`, the choice of delayed leave timings,
|
||||
mapping a data packet's sender to a member, and the transport registry, which keys on the raw
|
||||
`local.connectionState$` into `state$`, the choice of delayed leave timings, and the transport registry, which keys on the raw
|
||||
transport serialised with sorted keys and derives `connectedTransports$` and each
|
||||
`TransportMetadata.resolved$` from `connections$`.
|
||||
|
||||
@@ -613,6 +592,43 @@ the local transport, and keys its connections on the transport plus the server
|
||||
name where the transport has a `url`. A JWT service is reached directly, so
|
||||
members of several homeservers advertising the same one share a connection.
|
||||
|
||||
### Data tracks on LiveKit
|
||||
|
||||
A data track is a track like the media, in `tracks$`, so the backend carries it
|
||||
the way it carries the media: `local.publish` takes a `DataPublishRequest`, and
|
||||
`mediaFor` lists a remote member's data tracks after its media. Its sender is
|
||||
attested by construction: a track only exists on a member. The session never
|
||||
sees a key, a frame or a stream.
|
||||
|
||||
Every data track is one LiveKit data track, whose publication announces it: a
|
||||
receiver sees the track before its first message, and sees it go. LiveKit's
|
||||
data tracks are lossy only, so the LiveKit name says how the payloads travel
|
||||
(`dataTrackName.ts`):
|
||||
|
||||
| Request | LiveKit name | Payloads |
|
||||
| ----------------- | ----------------- | --------------------------------------------------------------- |
|
||||
| `reliable: false` | `lossy:<name>` | frames on the data track: unordered, no retransmits |
|
||||
| `reliable: true` | `reliable:<name>` | one byte stream per `send` on topic `reliable:<name>`, in order |
|
||||
|
||||
A track without a prefix, from a plain LiveKit publisher, is read as lossy. A
|
||||
room takes one byte stream handler per topic, so `ByteStreams.ts` shares it
|
||||
between every reliable track on that topic and filters by sender identity.
|
||||
|
||||
Encryption is LiveKit's. The rooms are built with `encryption` rather than
|
||||
`e2ee`, which encrypts the data channel as well as the media with the same
|
||||
E2EE manager and our key provider. The sender encrypts with its current key
|
||||
and puts the key index and IV on the wire: in each frame's E2EE extension for a
|
||||
lossy track, in each packet's `EncryptedPacket` for a stream. The receiver
|
||||
decrypts with the sender's key at that index. A frame that does not decrypt is
|
||||
dropped. A stream cannot be checked: LiveKit 2.23 sends an encrypted packet
|
||||
without its encryption type, so every received stream reads as unencrypted,
|
||||
and a cleartext stream injected on the way would be accepted as the sender's.
|
||||
|
||||
A local track is published on LiveKit once publishing is allowed, like the
|
||||
media, and again on every new connection to the local transport; `publish`
|
||||
resolves with that first publication. The object a host holds outlives the
|
||||
connections.
|
||||
|
||||
## Media: `tracks$` and `MediaTrack`
|
||||
|
||||
Every member has media. Today, in Element Call, that media _is_ a `livekit-client`
|
||||
@@ -707,14 +723,52 @@ export interface LocalVideoMediaTrack extends VideoMediaTrack, LocalMediaTrack {
|
||||
setProcessor(processor: VideoProcessor | undefined): Promise<void>;
|
||||
}
|
||||
|
||||
/** A named channel of bytes a member publishes beside its media. */
|
||||
export interface DataPublishRequest {
|
||||
source: "data";
|
||||
/** Unique per member; what receivers find the track by. */
|
||||
name: string;
|
||||
/** Whether every message arrives, in order. Default true. */
|
||||
reliable?: boolean;
|
||||
}
|
||||
|
||||
/** One data track of a member; it exists only on a member, so its sender is attested. */
|
||||
export interface DataTrack {
|
||||
source: "data";
|
||||
kind: "data";
|
||||
id: string;
|
||||
name: string;
|
||||
reliable: boolean;
|
||||
/** Whether payloads are end-to-end encrypted with the media key. */
|
||||
encrypted: boolean;
|
||||
/** What the member sends from now on, subscribed while observed. */
|
||||
data$: Observable<Uint8Array>;
|
||||
}
|
||||
|
||||
/** Our own data track; its `data$` replays what we sent. */
|
||||
export interface LocalDataTrack extends DataTrack {
|
||||
send(payload: Uint8Array): Promise<void>;
|
||||
}
|
||||
|
||||
export type Track = AudioMediaTrack | VideoMediaTrack | DataTrack;
|
||||
export type LocalTrack =
|
||||
LocalAudioMediaTrack | LocalVideoMediaTrack | LocalDataTrack;
|
||||
|
||||
export type EncryptionError = "MissingKey" | "InvalidKey";
|
||||
|
||||
/** The member's first track of a source; undefined while there is none, or no media yet. */
|
||||
export function trackBySource$<S extends MediaSource>(
|
||||
scope: ObservableScope,
|
||||
tracks$: Behavior<MediaTrack[] | null>,
|
||||
tracks$: Behavior<readonly (MediaTrack | DataTrack)[] | null>,
|
||||
source: S,
|
||||
): Behavior<TrackOfSource<S> | undefined>;
|
||||
|
||||
/** The member's data track of this name; undefined while there is none. */
|
||||
export function dataTrackByName$(
|
||||
scope: ObservableScope,
|
||||
tracks$: Behavior<readonly (MediaTrack | DataTrack)[] | null>,
|
||||
name: string,
|
||||
): Behavior<DataTrack | undefined>;
|
||||
```
|
||||
|
||||
### Rendering: `attach` and `detach`
|
||||
@@ -822,13 +876,11 @@ 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.
|
||||
- **Data before a receiver listens.** Nothing is replayed: a message sent
|
||||
before a receiver has subscribed to the track, which it does once it has
|
||||
seen the track published, does not reach it. Old builds of the legacy
|
||||
widget bundle, which sent bare packets, do not hear new builds or the other
|
||||
way round.
|
||||
- **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.
|
||||
|
||||
+67
-21
@@ -12,12 +12,22 @@ Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
import { combineLatest, type Observable, of, switchMap } from "rxjs";
|
||||
import {
|
||||
combineLatest,
|
||||
EMPTY,
|
||||
map,
|
||||
type Observable,
|
||||
of,
|
||||
switchMap,
|
||||
} from "rxjs";
|
||||
import {
|
||||
type AudioMediaTrack,
|
||||
createRTCSlot,
|
||||
dataTrackByName$,
|
||||
E2eeType,
|
||||
type LocalAudioMediaTrack,
|
||||
type LocalRTCMember,
|
||||
type LocalVideoMediaTrack,
|
||||
MatrixRTCMode,
|
||||
ObservableScope,
|
||||
type RemoteRTCMember,
|
||||
@@ -82,7 +92,7 @@ async function start(
|
||||
state.kind === "failed" ? `Error: ${state.error.message}` : state.kind;
|
||||
});
|
||||
showMembers(scope, rtcParticipation);
|
||||
showMessages(scope, rtcParticipation);
|
||||
showMessages(rtcParticipation);
|
||||
|
||||
toggle(buttons.microphone, async (enabled) =>
|
||||
setPublished(rtcParticipation, "microphone", enabled),
|
||||
@@ -120,10 +130,11 @@ async function setPublished(
|
||||
enabled: boolean,
|
||||
): Promise<boolean> {
|
||||
const track = rtcParticipation.localMember$.value?.tracks$.value?.find(
|
||||
(t) => t.source === source,
|
||||
(t): t is LocalAudioMediaTrack | LocalVideoMediaTrack =>
|
||||
t.source === source,
|
||||
);
|
||||
if (track && source === "screenShare") {
|
||||
if (!enabled) await rtcParticipation.unpublish(source);
|
||||
if (!enabled) await rtcParticipation.unpublish(track);
|
||||
return enabled;
|
||||
}
|
||||
if (track) return track.setEnabled(enabled);
|
||||
@@ -145,30 +156,63 @@ function toggle(
|
||||
};
|
||||
}
|
||||
|
||||
/** Sends what the form holds on the "chat" topic, and lists what arrives. */
|
||||
function showMessages(
|
||||
scope: ObservableScope,
|
||||
rtcParticipation: RTCParticipation,
|
||||
): void {
|
||||
/**
|
||||
* Publishes our "chat" data track with the join, so that the others hear the
|
||||
* first message, and sends what the form holds on it. What arrives is listed
|
||||
* per member, in `memberTile`.
|
||||
*/
|
||||
function showMessages(rtcParticipation: RTCParticipation): void {
|
||||
const encoder = new TextEncoder();
|
||||
const chat = rtcParticipation.publish({ source: "data", name: "chat" });
|
||||
chat.catch((e: unknown) => {
|
||||
status.textContent = `Error: ${e}`;
|
||||
});
|
||||
dataForm.hidden = false;
|
||||
dataForm.onsubmit = (event): void => {
|
||||
event.preventDefault();
|
||||
const text = new FormData(dataForm).get("text") as string;
|
||||
rtcParticipation.sendData("chat", text).then(
|
||||
() => dataForm.reset(),
|
||||
(e: unknown) => {
|
||||
status.textContent = `Error: ${e}`;
|
||||
},
|
||||
);
|
||||
chat
|
||||
.then(async (track) => track.send(encoder.encode(text)))
|
||||
.then(
|
||||
() => dataForm.reset(),
|
||||
(e: unknown) => {
|
||||
status.textContent = `Error: ${e}`;
|
||||
},
|
||||
);
|
||||
};
|
||||
rtcParticipation.data$
|
||||
.pipe(scope.bind())
|
||||
.subscribe(({ member, topic, text }) => {
|
||||
}
|
||||
|
||||
/**
|
||||
* Lists what a remote member sends on its "chat" data track, and marks its
|
||||
* tile while it has one.
|
||||
*/
|
||||
function listMessages(
|
||||
scope: ObservableScope,
|
||||
member: RemoteRTCMember,
|
||||
tile: HTMLElement,
|
||||
): void {
|
||||
const decoder = new TextDecoder();
|
||||
const chat$ = dataTrackByName$(scope, member.tracks$, "chat");
|
||||
chat$.pipe(scope.bind()).subscribe((track) => {
|
||||
tile.dataset.chat = String(track !== undefined);
|
||||
});
|
||||
chat$
|
||||
.pipe(
|
||||
switchMap(
|
||||
(track) =>
|
||||
track?.data$.pipe(
|
||||
map((payload) => ({ payload, encrypted: track.encrypted })),
|
||||
) ?? EMPTY,
|
||||
),
|
||||
scope.bind(),
|
||||
)
|
||||
.subscribe(({ payload, encrypted }) => {
|
||||
const line = messages.appendChild(document.createElement("li"));
|
||||
line.dataset.testid = "message";
|
||||
line.dataset.topic = topic;
|
||||
line.dataset.name = "chat";
|
||||
line.dataset.encrypted = String(encrypted);
|
||||
line.dataset.userId = member.userId;
|
||||
line.textContent = `${member.displayName$.value}: ${text}`;
|
||||
line.textContent = `${member.displayName$.value}: ${decoder.decode(payload)}`;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -218,12 +262,14 @@ function memberTile(
|
||||
tile.appendChild(document.createElement("video")),
|
||||
);
|
||||
// Our own microphone would only echo
|
||||
if (!member.local)
|
||||
if (!member.local) {
|
||||
render(
|
||||
scope,
|
||||
trackBySource$(scope, member.tracks$, "microphone"),
|
||||
tile.appendChild(document.createElement("audio")),
|
||||
);
|
||||
listMessages(scope, member, tile);
|
||||
}
|
||||
return tile;
|
||||
}
|
||||
|
||||
|
||||
+5
-1
@@ -55,7 +55,11 @@ export {
|
||||
export * from "./src/api";
|
||||
export * from "./src/media-api";
|
||||
export * from "./src/media-backend/api";
|
||||
export { trackBySource$, type TrackOfSource } from "./src/utils/tracks";
|
||||
export {
|
||||
dataTrackByName$,
|
||||
trackBySource$,
|
||||
type TrackOfSource,
|
||||
} from "./src/utils/tracks";
|
||||
export { createRTCSlot } from "./src/RTCSlot";
|
||||
export {
|
||||
createLivekitBackend,
|
||||
|
||||
@@ -14,11 +14,10 @@ import {
|
||||
type Transport,
|
||||
type UnstableLivekitTransport,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { BehaviorSubject, EMPTY, Subject } from "rxjs";
|
||||
import { BehaviorSubject, EMPTY } from "rxjs";
|
||||
import { describe, expect, it, type Mock, vi } from "vitest";
|
||||
|
||||
import {
|
||||
type DataMessage,
|
||||
type ParticipationState,
|
||||
type RTCMember,
|
||||
type RTCParticipationOptions,
|
||||
@@ -29,13 +28,16 @@ import { defaultSessionTimings, MatrixRTCMode } from "./config";
|
||||
import { E2eeType } from "./encryption";
|
||||
import { MatrixRTCError } from "./errors";
|
||||
import {
|
||||
type DataPacket,
|
||||
type MediaBackend,
|
||||
type MediaBackendContext,
|
||||
MediaConnectionState,
|
||||
type MediaKey,
|
||||
} from "./media-backend/api";
|
||||
import { type MemberMedia } from "./media-api";
|
||||
import {
|
||||
type LocalDataTrack,
|
||||
type LocalVideoMediaTrack,
|
||||
type MemberMedia,
|
||||
} from "./media-api";
|
||||
import { constant } from "./reactive/Behavior";
|
||||
import { ObservableScope } from "./reactive/ObservableScope";
|
||||
import { createRTCParticipation } from "./RTCParticipation";
|
||||
@@ -103,7 +105,6 @@ interface FakeBackend {
|
||||
context: MediaBackendContext;
|
||||
prepared: PromiseWithResolvers<{ canDelegateDelayedLeave: boolean }>;
|
||||
connectionState$: BehaviorSubject<MediaConnectionState | Error>;
|
||||
data$: Subject<DataPacket>;
|
||||
setPublishing: Mock<(publish: boolean) => void>;
|
||||
delegateDelayedLeave: Mock<(delayId: string) => Promise<void>>;
|
||||
mediaFor: Mock<(scope: ObservableScope, member: RTCMember) => MemberMedia>;
|
||||
@@ -151,7 +152,6 @@ function setup(
|
||||
connectionState$: new BehaviorSubject<MediaConnectionState | Error>(
|
||||
MediaConnectionState.Initialized,
|
||||
),
|
||||
data$: new Subject<DataPacket>(),
|
||||
setPublishing: vi.fn(),
|
||||
delegateDelayedLeave: vi.fn(async () => Promise.resolve()),
|
||||
mediaFor: vi.fn(() => ({
|
||||
@@ -179,8 +179,6 @@ function setup(
|
||||
mediaFor: fake.mediaFor,
|
||||
connections$: constant([]),
|
||||
setAudioOutputDeviceId: vi.fn(),
|
||||
sendData: vi.fn(),
|
||||
data$: fake.data$,
|
||||
};
|
||||
fake.context = context;
|
||||
fake.backend = backend;
|
||||
@@ -414,27 +412,30 @@ describe("createRTCParticipation", () => {
|
||||
scope.end();
|
||||
});
|
||||
|
||||
it("delivers data from a member with its member and drops the rest", () => {
|
||||
const { participation, members$, fake, scope } = setup();
|
||||
members$.next([
|
||||
member({
|
||||
userId: "@bob:example.org",
|
||||
memberId: "bob-member",
|
||||
backendIdentities: ["bob-hashed", "bob-member"],
|
||||
}),
|
||||
]);
|
||||
const messages: DataMessage[] = [];
|
||||
participation.data$.subscribe((message) => messages.push(message));
|
||||
it("hands publications to the backend, data tracks too", async () => {
|
||||
const { participation, fake, scope } = setup();
|
||||
const chat = { kind: "data", name: "chat" } as LocalDataTrack;
|
||||
const screenShare = {
|
||||
kind: "video",
|
||||
source: "screenShare",
|
||||
} as LocalVideoMediaTrack;
|
||||
|
||||
fake.data$.next({ senderId: "bob-hashed", topic: "chat", text: "hi" });
|
||||
fake.data$.next({ senderId: "stranger", topic: "chat", text: "boo" });
|
||||
|
||||
expect(messages).toHaveLength(1);
|
||||
expect(messages[0]).toMatchObject({
|
||||
topic: "chat",
|
||||
text: "hi",
|
||||
member: { memberId: "bob-member" },
|
||||
await participation.publish({ source: "data", name: "chat" });
|
||||
await participation.publish({ source: "camera" });
|
||||
expect(fake.backend.local.publish).toHaveBeenNthCalledWith(1, {
|
||||
source: "data",
|
||||
name: "chat",
|
||||
});
|
||||
expect(fake.backend.local.publish).toHaveBeenNthCalledWith(2, {
|
||||
source: "camera",
|
||||
});
|
||||
await participation.unpublish(chat);
|
||||
await participation.unpublish(screenShare);
|
||||
expect(fake.backend.local.unpublish).toHaveBeenNthCalledWith(1, chat);
|
||||
expect(fake.backend.local.unpublish).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
screenShare,
|
||||
);
|
||||
scope.end();
|
||||
});
|
||||
|
||||
|
||||
+23
-22
@@ -20,7 +20,6 @@ import { v4 as uuidv4 } from "uuid";
|
||||
import {
|
||||
combineLatest,
|
||||
distinctUntilChanged,
|
||||
filter,
|
||||
from,
|
||||
fromEvent,
|
||||
map,
|
||||
@@ -35,7 +34,6 @@ import { type ObservableScope } from "./reactive/ObservableScope";
|
||||
import { generateItems } from "./reactive/observable";
|
||||
import { mapScoped } from "./utils/mapScoped";
|
||||
import {
|
||||
type DataMessage,
|
||||
type LocalRTCMember,
|
||||
type ParticipationState,
|
||||
type RemoteRTCMember,
|
||||
@@ -45,6 +43,14 @@ import {
|
||||
type RTCSlot,
|
||||
type RTCSlotOptions,
|
||||
} from "./api";
|
||||
import {
|
||||
type DataPublishRequest,
|
||||
type LocalAudioMediaTrack,
|
||||
type LocalDataTrack,
|
||||
type LocalTrack,
|
||||
type LocalVideoMediaTrack,
|
||||
type PublishRequest,
|
||||
} from "./media-api";
|
||||
import {
|
||||
type MediaBackend,
|
||||
type MediaBackendContext,
|
||||
@@ -245,22 +251,6 @@ export function createRTCParticipation(
|
||||
).pipe(map((member) => member ?? null)),
|
||||
);
|
||||
|
||||
// A packet from an identity that is not a member is dropped, so a host only
|
||||
// ever hears from attested members
|
||||
const data$ = backend.data$.pipe(
|
||||
map(({ senderId, topic, text }): DataMessage | null => {
|
||||
const member = remoteMembers$.value.find((m) =>
|
||||
m.backendIdentities.includes(senderId),
|
||||
);
|
||||
if (member === undefined) {
|
||||
logger.warn(`Dropping data from ${senderId}: not a member`);
|
||||
return null;
|
||||
}
|
||||
return { member, topic, text };
|
||||
}),
|
||||
filter((message) => message !== null),
|
||||
);
|
||||
|
||||
// Nothing leaves this device while it may already have been dropped from
|
||||
// the session: the member would show as away while still being heard
|
||||
localMembership.state$
|
||||
@@ -285,6 +275,19 @@ export function createRTCParticipation(
|
||||
),
|
||||
);
|
||||
|
||||
function publish(request: DataPublishRequest): Promise<LocalDataTrack>;
|
||||
function publish(
|
||||
request: PublishRequest,
|
||||
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack>;
|
||||
async function publish(
|
||||
request: DataPublishRequest | PublishRequest,
|
||||
): Promise<LocalTrack> {
|
||||
// Narrowed for the overloads; both go to the backend alike
|
||||
return request.source === "data"
|
||||
? backend.local.publish(request)
|
||||
: backend.local.publish(request);
|
||||
}
|
||||
|
||||
const leave = (): void => {
|
||||
if (state$.value.kind === "left") return;
|
||||
localMembership.leave();
|
||||
@@ -299,14 +302,12 @@ export function createRTCParticipation(
|
||||
state$,
|
||||
localMember$,
|
||||
remoteMembers$,
|
||||
publish: async (request) => backend.local.publish(request),
|
||||
unpublish: async (source) => backend.local.unpublish(source),
|
||||
publish,
|
||||
unpublish: async (track) => backend.local.unpublish(track),
|
||||
keyRotationSuppressed$: createKeyRotationSuppressed$(scope, jsSdkSession),
|
||||
connectedTransports$: transports.connected$,
|
||||
setAudioOutputDeviceId: async (deviceId) =>
|
||||
backend.setAudioOutputDeviceId(deviceId),
|
||||
sendData: async (topic, text) => backend.sendData(topic, text),
|
||||
data$,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
+12
-30
@@ -15,15 +15,15 @@ import {
|
||||
type RTCNotificationType,
|
||||
type Transport,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { type Observable } from "rxjs";
|
||||
|
||||
import { type Behavior } from "./reactive/Behavior";
|
||||
import {
|
||||
type DataPublishRequest,
|
||||
type LocalAudioMediaTrack,
|
||||
type LocalDataTrack,
|
||||
type LocalMedia,
|
||||
type LocalMemberMedia,
|
||||
type LocalTrack,
|
||||
type LocalVideoMediaTrack,
|
||||
type MediaSource,
|
||||
type MemberMedia,
|
||||
type PublishRequest,
|
||||
} from "./media-api";
|
||||
@@ -176,15 +176,19 @@ export interface RTCParticipation {
|
||||
remoteMembers$: Behavior<RemoteRTCMember[]>;
|
||||
|
||||
/**
|
||||
* Publishes a source and resolves with its track, which also shows up in
|
||||
* `localMember$.tracks$`. Works from the moment `join()` returns: before the
|
||||
* transport is connected.
|
||||
* Publishes a source, or a data track, and resolves with its track, which
|
||||
* also shows up in `localMember$.tracks$`. Works from the moment `join()`
|
||||
* returns: before the transport is connected.
|
||||
*/
|
||||
publish(request: DataPublishRequest): Promise<LocalDataTrack>;
|
||||
publish(
|
||||
request: PublishRequest,
|
||||
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack>;
|
||||
/** Removes one of our publications; a screen share takes its audio with it. */
|
||||
unpublish(source: MediaSource): Promise<void>;
|
||||
/**
|
||||
* Removes one of our publications; a screen share takes its audio with it.
|
||||
* The track's object is finished: its controls and `send` reject from then on.
|
||||
*/
|
||||
unpublish(track: LocalTrack): Promise<void>;
|
||||
|
||||
/**
|
||||
* Whether the slot has grown large enough that MatrixRTC has stopped
|
||||
@@ -201,28 +205,6 @@ export interface RTCParticipation {
|
||||
* the browser's choice, never calls it.
|
||||
*/
|
||||
setAudioOutputDeviceId(deviceId: string): Promise<void>;
|
||||
|
||||
/**
|
||||
* 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<void>;
|
||||
/**
|
||||
* What remote members sent with `sendData`, on every transport the
|
||||
* participation 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<DataMessage>;
|
||||
}
|
||||
|
||||
/** One `sendData` call as it arrives at a remote member. */
|
||||
export interface DataMessage {
|
||||
member: RemoteRTCMember;
|
||||
topic: string;
|
||||
text: string;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
+65
-7
@@ -69,6 +69,21 @@ export type PublishRequest =
|
||||
capture?: VideoCaptureSettings;
|
||||
};
|
||||
|
||||
/** A named channel of bytes a member publishes beside its media. */
|
||||
export interface DataPublishRequest {
|
||||
source: "data";
|
||||
/**
|
||||
* What receivers find the track by, with `dataTrackByName$`. Unique per
|
||||
* member: publishing a name that is published already rejects.
|
||||
*/
|
||||
name: string;
|
||||
/**
|
||||
* Whether every message arrives, in order. Default true. Lossy delivery is
|
||||
* for high-rate state where only the newest value matters, such as cursors.
|
||||
*/
|
||||
reliable?: boolean;
|
||||
}
|
||||
|
||||
export type MediaSource =
|
||||
| "microphone"
|
||||
| "camera"
|
||||
@@ -82,7 +97,7 @@ export type MediaStreamStats =
|
||||
| RTCOutboundRtpStreamStats
|
||||
| undefined;
|
||||
|
||||
/** One published track of a member. */
|
||||
/** One published audio or video track of a member. */
|
||||
export interface MediaTrack {
|
||||
source: MediaSource;
|
||||
kind: "audio" | "video";
|
||||
@@ -163,6 +178,48 @@ export interface LocalVideoMediaTrack extends VideoMediaTrack, LocalMediaTrack {
|
||||
setProcessor(processor: VideoProcessor | undefined): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* One data track of a member. It exists only on a member of the slot, so
|
||||
* whatever arrives on it comes from an attested member.
|
||||
*/
|
||||
export interface DataTrack {
|
||||
source: "data";
|
||||
kind: "data";
|
||||
/** Stable for the life of the track. */
|
||||
id: string;
|
||||
/** As the publisher named it in its `DataPublishRequest`. */
|
||||
name: string;
|
||||
reliable: boolean;
|
||||
/** Whether payloads are end-to-end encrypted with the media key. */
|
||||
encrypted: boolean;
|
||||
/**
|
||||
* What the member sends from now on. The track is subscribed while this is
|
||||
* observed and not at all otherwise; nothing sent earlier is replayed.
|
||||
*/
|
||||
data$: Observable<Uint8Array>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Our own data track. Its `data$` replays what we sent: each payload once
|
||||
* `send` has handed it to the transport, so a host renders its own messages
|
||||
* from the same stream as a remote member's.
|
||||
*/
|
||||
export interface LocalDataTrack extends DataTrack {
|
||||
/**
|
||||
* Sends to every member subscribed to this track. Rejects while the local
|
||||
* transport is not connected, and where the payload is larger than the
|
||||
* backend carries.
|
||||
*/
|
||||
send(payload: Uint8Array): Promise<void>;
|
||||
}
|
||||
|
||||
/** Any track of a member: its media and its data. */
|
||||
export type Track = AudioMediaTrack | VideoMediaTrack | DataTrack;
|
||||
export type LocalTrack =
|
||||
| LocalAudioMediaTrack
|
||||
| LocalVideoMediaTrack
|
||||
| LocalDataTrack;
|
||||
|
||||
export type EncryptionError = "MissingKey" | "InvalidKey";
|
||||
|
||||
/**
|
||||
@@ -172,18 +229,19 @@ export type EncryptionError = "MissingKey" | "InvalidKey";
|
||||
*/
|
||||
export interface MemberMedia {
|
||||
/**
|
||||
* The member's tracks, in publication order, once it has shown up on its
|
||||
* transport; null until then ("waiting for media"). An entry stays the same
|
||||
* object for as long as the same publication is behind it. Which track is
|
||||
* which is in its `source`; `trackBySource$` picks one out.
|
||||
* The member's tracks, media and data, in publication order, once it has
|
||||
* shown up on its transport; null until then ("waiting for media"). An
|
||||
* entry stays the same object for as long as the same publication is
|
||||
* behind it. Which track is which is in its `source`: `trackBySource$`
|
||||
* picks a media track out, `dataTrackByName$` a data track.
|
||||
*/
|
||||
tracks$: Behavior<(AudioMediaTrack | VideoMediaTrack)[] | null>;
|
||||
tracks$: Behavior<Track[] | null>;
|
||||
/** Emits when the SFU reports a key problem for this member. */
|
||||
encryptionError$: Observable<EncryptionError>;
|
||||
}
|
||||
|
||||
export interface LocalMemberMedia extends MemberMedia {
|
||||
tracks$: Behavior<(LocalAudioMediaTrack | LocalVideoMediaTrack)[] | null>;
|
||||
tracks$: Behavior<LocalTrack[] | null>;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -19,11 +19,13 @@ import { type Observable } from "rxjs";
|
||||
|
||||
import { type ResolvedTransport, type RTCMember } from "../api";
|
||||
import {
|
||||
type DataPublishRequest,
|
||||
type LocalAudioMediaTrack,
|
||||
type LocalDataTrack,
|
||||
type LocalMedia,
|
||||
type LocalMemberMedia,
|
||||
type LocalTrack,
|
||||
type LocalVideoMediaTrack,
|
||||
type MediaSource,
|
||||
type MemberMedia,
|
||||
type PublishRequest,
|
||||
} from "../media-api";
|
||||
@@ -106,13 +108,6 @@ export interface BackendConnection {
|
||||
resolved: ResolvedTransport | undefined;
|
||||
}
|
||||
|
||||
export interface DataPacket {
|
||||
/** The backend identity of the sender. */
|
||||
senderId: string;
|
||||
topic: string;
|
||||
text: string;
|
||||
}
|
||||
|
||||
export interface MediaBackend {
|
||||
/**
|
||||
* The membership transport types this backend serves, ordered decreasingly
|
||||
@@ -158,14 +153,6 @@ export interface MediaBackend {
|
||||
* exists keeps the device it was created with.
|
||||
*/
|
||||
setAudioOutputDeviceId(deviceId: string): Promise<void>;
|
||||
|
||||
/**
|
||||
* A text channel beside the media: sent on the local transport, received on every one.
|
||||
* This could well be just another track kind. Not needed for element call. just for the sdk target.
|
||||
* Keeping it on the top level for visibility. Should be tackled in a future refactor.
|
||||
*/
|
||||
sendData(topic: string, text: string): Promise<void>;
|
||||
readonly data$: Observable<DataPacket>;
|
||||
}
|
||||
|
||||
/** `tracks$` is null until the connection carries a local participant. */
|
||||
@@ -187,14 +174,17 @@ export interface LocalMediaBackend extends LocalMemberMedia {
|
||||
readonly publishError$: Behavior<Error | null>;
|
||||
|
||||
/**
|
||||
* Publishes a source and resolves with its track once it is in `tracks$`.
|
||||
* Remembered across connections, so a request before the
|
||||
* Publishes a source, or a data track, and resolves with its track once it
|
||||
* is in `tracks$`. Remembered across connections, so a request before the
|
||||
* transport is up is applied once it is, and republished after a
|
||||
* reconnection. Rejects where the device could not be used.
|
||||
* reconnection. Rejects where the device could not be used. A data track
|
||||
* is end-to-end encrypted by the backend wherever the media is; the session
|
||||
* never sees how.
|
||||
*/
|
||||
publish(request: DataPublishRequest): Promise<LocalDataTrack>;
|
||||
publish(
|
||||
request: PublishRequest,
|
||||
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack>;
|
||||
/** Removes one of our publications; a screen share takes its audio with it. */
|
||||
unpublish(source: MediaSource): Promise<void>;
|
||||
unpublish(track: LocalTrack): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
/*
|
||||
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 ByteStreamHandler,
|
||||
type ByteStreamReader,
|
||||
type Room as LivekitRoom,
|
||||
} from "livekit-client";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { byteStreams$, type ReceivedBytes } from "./ByteStreams";
|
||||
import { flushPromises } from "../../utils/test";
|
||||
|
||||
function fakeRoom(): {
|
||||
room: LivekitRoom;
|
||||
handlers: Map<string, ByteStreamHandler>;
|
||||
} {
|
||||
const handlers = new Map<string, ByteStreamHandler>();
|
||||
const room = {
|
||||
registerByteStreamHandler: vi.fn(
|
||||
(topic: string, handler: ByteStreamHandler) => {
|
||||
if (handlers.has(topic)) throw new Error("already registered");
|
||||
handlers.set(topic, handler);
|
||||
},
|
||||
),
|
||||
unregisterByteStreamHandler: vi.fn((topic: string) =>
|
||||
handlers.delete(topic),
|
||||
),
|
||||
} as unknown as LivekitRoom;
|
||||
return { room, handlers };
|
||||
}
|
||||
|
||||
function reader(chunks: number[][]): {
|
||||
reader: ByteStreamReader;
|
||||
finish: () => void;
|
||||
} {
|
||||
const done = Promise.withResolvers<Uint8Array[]>();
|
||||
return {
|
||||
reader: {
|
||||
readAll: async () => done.promise,
|
||||
} as unknown as ByteStreamReader,
|
||||
finish: (): void => done.resolve(chunks.map((c) => new Uint8Array(c))),
|
||||
};
|
||||
}
|
||||
|
||||
describe("byteStreams$", () => {
|
||||
it("shares one handler per topic, released with the last observer", () => {
|
||||
const { room, handlers } = fakeRoom();
|
||||
const first = byteStreams$(room, "reliable:chat").subscribe();
|
||||
const second = byteStreams$(room, "reliable:chat").subscribe();
|
||||
expect(room.registerByteStreamHandler).toHaveBeenCalledTimes(1);
|
||||
first.unsubscribe();
|
||||
expect(handlers.has("reliable:chat")).toBe(true);
|
||||
second.unsubscribe();
|
||||
expect(handlers.has("reliable:chat")).toBe(false);
|
||||
});
|
||||
|
||||
it("joins the chunks and keeps the order the streams started in", async () => {
|
||||
const { room, handlers } = fakeRoom();
|
||||
const received: ReceivedBytes[] = [];
|
||||
const subscription = byteStreams$(room, "reliable:chat").subscribe((r) =>
|
||||
received.push(r),
|
||||
);
|
||||
const handler = handlers.get("reliable:chat")!;
|
||||
const one = reader([[1, 2], [3]]);
|
||||
const two = reader([[4]]);
|
||||
handler(one.reader, { identity: "alice" });
|
||||
handler(two.reader, { identity: "bob" });
|
||||
two.finish();
|
||||
await flushPromises();
|
||||
expect(received).toEqual([]);
|
||||
one.finish();
|
||||
await flushPromises();
|
||||
|
||||
expect(received).toEqual([
|
||||
{ identity: "alice", payload: new Uint8Array([1, 2, 3]) },
|
||||
{ identity: "bob", payload: new Uint8Array([4]) },
|
||||
]);
|
||||
subscription.unsubscribe();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,82 @@
|
||||
/*
|
||||
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 { logger as rootLogger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
type ByteStreamReader,
|
||||
type Room as LivekitRoom,
|
||||
} from "livekit-client";
|
||||
import { catchError, concatMap, EMPTY, from, Observable, share } from "rxjs";
|
||||
|
||||
/** One byte stream, read to its end. */
|
||||
export interface ReceivedBytes {
|
||||
/** The participant identity of the sender. */
|
||||
identity: string;
|
||||
payload: Uint8Array;
|
||||
}
|
||||
|
||||
const byTopicByRoom = new WeakMap<
|
||||
LivekitRoom,
|
||||
Map<string, Observable<ReceivedBytes>>
|
||||
>();
|
||||
|
||||
/**
|
||||
* The byte streams a room receives on a topic, in the order they started. A
|
||||
* room takes one handler per topic, so every observer of a topic shares it:
|
||||
* it is registered with the first subscription and released with the last.
|
||||
*/
|
||||
export function byteStreams$(
|
||||
room: LivekitRoom,
|
||||
topic: string,
|
||||
): Observable<ReceivedBytes> {
|
||||
let byTopic = byTopicByRoom.get(room);
|
||||
if (byTopic === undefined) {
|
||||
byTopic = new Map();
|
||||
byTopicByRoom.set(room, byTopic);
|
||||
}
|
||||
let streams$ = byTopic.get(topic);
|
||||
if (streams$ === undefined) {
|
||||
streams$ = new Observable<{ reader: ByteStreamReader; identity: string }>(
|
||||
(subscriber) => {
|
||||
room.registerByteStreamHandler(topic, (reader, { identity }) =>
|
||||
subscriber.next({ reader, identity }),
|
||||
);
|
||||
return (): void => room.unregisterByteStreamHandler(topic);
|
||||
},
|
||||
).pipe(
|
||||
concatMap(({ reader, identity }) =>
|
||||
from(readStream(reader, identity)).pipe(
|
||||
catchError((e: unknown) => {
|
||||
rootLogger
|
||||
.getChild("[ByteStreams]")
|
||||
.warn(`Dropping a stream on ${topic} from ${identity}`, e);
|
||||
return EMPTY;
|
||||
}),
|
||||
),
|
||||
),
|
||||
share(),
|
||||
);
|
||||
byTopic.set(topic, streams$);
|
||||
}
|
||||
return streams$;
|
||||
}
|
||||
|
||||
async function readStream(
|
||||
reader: ByteStreamReader,
|
||||
identity: string,
|
||||
): Promise<ReceivedBytes> {
|
||||
const chunks = await reader.readAll();
|
||||
const payload = new Uint8Array(
|
||||
chunks.reduce((length, chunk) => length + chunk.length, 0),
|
||||
);
|
||||
let offset = 0;
|
||||
for (const chunk of chunks) {
|
||||
payload.set(chunk, offset);
|
||||
offset += chunk.length;
|
||||
}
|
||||
return { identity, payload };
|
||||
}
|
||||
@@ -77,6 +77,7 @@ describe("LivekitConnectionFactory", () => {
|
||||
resolution: expect.objectContaining({ height: 720 }),
|
||||
});
|
||||
expect(options.audioOutput).toEqual({ deviceId: undefined });
|
||||
expect(options.encryption).toBeUndefined();
|
||||
expect(options.e2ee).toBeUndefined();
|
||||
});
|
||||
|
||||
|
||||
@@ -19,6 +19,7 @@ import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/En
|
||||
import { BehaviorSubject, combineLatest, map } from "rxjs";
|
||||
|
||||
import { type ResolvedTransport } from "../../api";
|
||||
import { watchRemoteDataTracks } from "./RemoteDataTracks";
|
||||
import { type MediaQuality } from "../../config";
|
||||
import { type PublishRequest } from "../../media-api";
|
||||
import { type Behavior } from "../../reactive/Behavior";
|
||||
@@ -128,12 +129,14 @@ export class LivekitConnectionFactory implements ConnectionFactory {
|
||||
client: this.client,
|
||||
scope,
|
||||
livekitRoomFactory: () =>
|
||||
new LivekitRoom(
|
||||
roomOptions(
|
||||
this.publish,
|
||||
this.audioOutputDeviceId$.value,
|
||||
this.keyProvider,
|
||||
this.mediaQuality,
|
||||
watchRemoteDataTracks(
|
||||
new LivekitRoom(
|
||||
roomOptions(
|
||||
this.publish,
|
||||
this.audioOutputDeviceId$.value,
|
||||
this.keyProvider,
|
||||
this.mediaQuality,
|
||||
),
|
||||
),
|
||||
),
|
||||
ownMembershipIdentity,
|
||||
@@ -169,8 +172,10 @@ function roomOptions(
|
||||
},
|
||||
audioOutput: { deviceId: audioOutputDeviceId },
|
||||
// Every room needs a worker of its own: one gets confused by streams from
|
||||
// several rooms
|
||||
e2ee: keyProvider && { keyProvider, worker: new E2EEWorker() },
|
||||
// several rooms. `encryption` rather than `e2ee`, so that LiveKit encrypts
|
||||
// the data channel too, which carries the reliable data tracks; media is
|
||||
// encrypted the same either way
|
||||
encryption: keyProvider && { keyProvider, worker: new E2EEWorker() },
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -1,143 +0,0 @@
|
||||
/*
|
||||
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 {
|
||||
ConnectionState as LivekitConnectionState,
|
||||
type RemoteParticipant,
|
||||
RoomEvent,
|
||||
} from "livekit-client";
|
||||
import { BehaviorSubject } from "rxjs";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { Epoch } from "../../reactive/ObservableScope";
|
||||
import { constant } from "../../reactive/Behavior";
|
||||
import { type DataPacket } from "../api";
|
||||
import { type Connection, type ConnectionState } from "./Connection";
|
||||
import { ConnectionMap } from "./ConnectionMap";
|
||||
import { createDataChannel$ } from "./DataChannel";
|
||||
import { testScope } from "../../utils/test";
|
||||
import { exampleTransport } from "./test";
|
||||
|
||||
const encode = (text: string): Uint8Array => new TextEncoder().encode(text);
|
||||
|
||||
const connected: ConnectionState = {
|
||||
state: "authenticated",
|
||||
livekitState: LivekitConnectionState.Connected,
|
||||
};
|
||||
|
||||
function fakeConnection(
|
||||
url: string,
|
||||
state: ConnectionState = connected,
|
||||
): {
|
||||
connection: Connection;
|
||||
room: EventEmitter;
|
||||
publishData: ReturnType<typeof vi.fn>;
|
||||
} {
|
||||
const room = new EventEmitter();
|
||||
const publishData = vi.fn(async () => Promise.resolve());
|
||||
const connection = {
|
||||
transport: { ...exampleTransport, livekit_service_url: url },
|
||||
serverName: "example.org",
|
||||
state$: constant(state),
|
||||
livekitRoom: Object.assign(room, { localParticipant: { publishData } }),
|
||||
} as unknown as Connection;
|
||||
return { connection, room, publishData };
|
||||
}
|
||||
|
||||
function setup(connections: Connection[]): {
|
||||
received: DataPacket[];
|
||||
connections$: BehaviorSubject<Epoch<ConnectionMap>>;
|
||||
sendData: (topic: string, text: string) => Promise<void>;
|
||||
} {
|
||||
const data = new ConnectionMap();
|
||||
for (const connection of connections) data.add(connection, []);
|
||||
const connections$ = new BehaviorSubject(new Epoch(data));
|
||||
const { data$, sendData } = createDataChannel$({
|
||||
scope: testScope(),
|
||||
connectionMap$: connections$,
|
||||
connection$: constant(connections[0] ?? null),
|
||||
});
|
||||
const received: DataPacket[] = [];
|
||||
data$.subscribe((message) => received.push(message));
|
||||
return { received, connections$, sendData };
|
||||
}
|
||||
|
||||
describe("data$", () => {
|
||||
it("delivers a packet with its sender, 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([
|
||||
{ senderId: "alice", topic: "chat", text: "hello" },
|
||||
]);
|
||||
});
|
||||
|
||||
it("drops packets from the server", () => {
|
||||
const { connection, room } = fakeConnection("https://sfu");
|
||||
const { received } = setup([connection]);
|
||||
|
||||
room.emit(RoomEvent.DataReceived, encode("server"), undefined);
|
||||
|
||||
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]);
|
||||
|
||||
const data = new ConnectionMap();
|
||||
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.senderId, 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", {
|
||||
state: "initialized",
|
||||
});
|
||||
const { sendData } = setup([connection]);
|
||||
|
||||
await expect(sendData("chat", "hello")).rejects.toThrow("not connected");
|
||||
expect(publishData).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -1,119 +0,0 @@
|
||||
/*
|
||||
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 {
|
||||
ConnectionState as LivekitConnectionState,
|
||||
type RemoteParticipant,
|
||||
type Room as LivekitRoom,
|
||||
RoomEvent,
|
||||
} from "livekit-client";
|
||||
import {
|
||||
distinctUntilChanged,
|
||||
filter,
|
||||
fromEvent,
|
||||
map,
|
||||
merge,
|
||||
type Observable,
|
||||
share,
|
||||
switchMap,
|
||||
} from "rxjs";
|
||||
|
||||
import { type Behavior } from "../../reactive/Behavior";
|
||||
import {
|
||||
type Epoch,
|
||||
type ObservableScope,
|
||||
} from "../../reactive/ObservableScope";
|
||||
import { type DataPacket } from "../api";
|
||||
import { type Connection } from "./Connection";
|
||||
import { type ConnectionMap } from "./ConnectionMap";
|
||||
|
||||
interface Props {
|
||||
scope: ObservableScope;
|
||||
connectionMap$: Behavior<Epoch<ConnectionMap>>;
|
||||
/** The local member's connection, which carries what we send. */
|
||||
connection$: Behavior<Connection | null>;
|
||||
}
|
||||
|
||||
export interface DataChannel {
|
||||
data$: Observable<DataPacket>;
|
||||
sendData: (topic: string, text: string) => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 reported by its identity; the packet's own claim of who sent it
|
||||
* never reaches the host.
|
||||
*/
|
||||
export function createDataChannel$({
|
||||
scope,
|
||||
connectionMap$,
|
||||
connection$,
|
||||
}: Props): DataChannel {
|
||||
const decoder = new TextDecoder();
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
const rooms$ = connectionMap$.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 }): DataPacket | null =>
|
||||
// Only the server sends without a participant; a host has no use for it
|
||||
participant === undefined
|
||||
? null
|
||||
: {
|
||||
senderId: participant.identity,
|
||||
topic: topic ?? "",
|
||||
text: decoder.decode(payload),
|
||||
},
|
||||
),
|
||||
filter((packet) => packet !== null),
|
||||
scope.bind(),
|
||||
share(),
|
||||
);
|
||||
|
||||
const sendData = async (topic: string, text: string): Promise<void> => {
|
||||
const connection = connection$.value;
|
||||
const state = connection?.state$.value;
|
||||
if (
|
||||
connection === null ||
|
||||
state?.state !== "authenticated" ||
|
||||
state.livekitState !== LivekitConnectionState.Connected
|
||||
)
|
||||
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<Received> {
|
||||
return fromEvent(
|
||||
room,
|
||||
RoomEvent.DataReceived,
|
||||
(
|
||||
payload: Uint8Array,
|
||||
participant?: RemoteParticipant,
|
||||
_kind?: unknown,
|
||||
topic?: string,
|
||||
): Received => ({ payload, participant, topic }),
|
||||
);
|
||||
}
|
||||
@@ -58,7 +58,6 @@ import {
|
||||
ResolvedConnection,
|
||||
} from "./ConnectionFactory";
|
||||
import { toMediaConnectionState } from "./connectionState";
|
||||
import { createDataChannel$ } from "./DataChannel";
|
||||
import { createKeyProvider } from "./KeyProvider";
|
||||
import {
|
||||
createLivekitMemberMedia,
|
||||
@@ -263,7 +262,12 @@ export function createLivekitBackend(
|
||||
`${member.rtcBackendIdentity}: LiveKit participant missing on ${transportString}`,
|
||||
);
|
||||
});
|
||||
return createLivekitMemberMedia(memberScope, source$, encryptionSystem);
|
||||
return createLivekitMemberMedia(
|
||||
memberScope,
|
||||
source$,
|
||||
encryptionSystem,
|
||||
logger,
|
||||
);
|
||||
};
|
||||
|
||||
const connections$ = scope.behavior<BackendConnection[]>(
|
||||
@@ -276,12 +280,6 @@ export function createLivekitBackend(
|
||||
),
|
||||
);
|
||||
|
||||
const { data$, sendData } = createDataChannel$({
|
||||
scope,
|
||||
connectionMap$,
|
||||
connection$: localMedia_.connection$,
|
||||
});
|
||||
|
||||
return {
|
||||
transportTypes: ["m.livekit", "livekit"],
|
||||
prepareLocalTransport,
|
||||
@@ -290,8 +288,6 @@ export function createLivekitBackend(
|
||||
mediaFor,
|
||||
connections$,
|
||||
setAudioOutputDeviceId: localMedia_.setAudioOutputDeviceId,
|
||||
sendData,
|
||||
data$,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
/*
|
||||
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 { logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
type ByteStreamHandler,
|
||||
type ByteStreamReader,
|
||||
type DataTrackFrame,
|
||||
type RemoteDataTrack,
|
||||
type Room as LivekitRoom,
|
||||
} from "livekit-client";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { createLivekitDataTrack } from "./LivekitDataTrack";
|
||||
import { flushPromises } from "../../utils/test";
|
||||
|
||||
function remoteTrack(
|
||||
name: string,
|
||||
usesE2ee = true,
|
||||
): {
|
||||
track: RemoteDataTrack;
|
||||
frames: ReadableStreamDefaultController<DataTrackFrame>[];
|
||||
cancel: ReturnType<typeof vi.fn>;
|
||||
} {
|
||||
const frames: ReadableStreamDefaultController<DataTrackFrame>[] = [];
|
||||
const cancel = vi.fn();
|
||||
const track = {
|
||||
info: { sid: "DT_1", name, usesE2ee },
|
||||
publisherIdentity: "alice",
|
||||
subscribe: vi.fn(
|
||||
() =>
|
||||
new ReadableStream<DataTrackFrame>({
|
||||
start: (controller): void => {
|
||||
frames.push(controller);
|
||||
},
|
||||
cancel,
|
||||
}),
|
||||
),
|
||||
} as unknown as RemoteDataTrack;
|
||||
return { track, frames, cancel };
|
||||
}
|
||||
|
||||
function roomWithStreams(): {
|
||||
room: LivekitRoom;
|
||||
send: (identity: string, bytes: number[]) => void;
|
||||
} {
|
||||
const handlers = new Map<string, ByteStreamHandler>();
|
||||
return {
|
||||
room: {
|
||||
registerByteStreamHandler: (topic: string, h: ByteStreamHandler) =>
|
||||
handlers.set(topic, h),
|
||||
unregisterByteStreamHandler: (topic: string) => handlers.delete(topic),
|
||||
} as unknown as LivekitRoom,
|
||||
send: (identity, bytes): void =>
|
||||
handlers.get("reliable:chat")!(
|
||||
{
|
||||
readAll: async () => Promise.resolve([new Uint8Array(bytes)]),
|
||||
} as unknown as ByteStreamReader,
|
||||
{ identity },
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
describe("a remote data track", () => {
|
||||
it("names itself without the LiveKit prefix", () => {
|
||||
const { track } = remoteTrack("reliable:chat", false);
|
||||
expect(
|
||||
createLivekitDataTrack(roomWithStreams().room, track, logger),
|
||||
).toMatchObject({
|
||||
source: "data",
|
||||
kind: "data",
|
||||
id: "DT_1",
|
||||
name: "chat",
|
||||
reliable: true,
|
||||
encrypted: false,
|
||||
});
|
||||
});
|
||||
|
||||
it("subscribes a lossy track while observed, once for all observers", async () => {
|
||||
const { track, frames, cancel } = remoteTrack("lossy:cursors");
|
||||
const dataTrack = createLivekitDataTrack(
|
||||
roomWithStreams().room,
|
||||
track,
|
||||
logger,
|
||||
);
|
||||
expect(track.subscribe).not.toHaveBeenCalled();
|
||||
|
||||
const received: Uint8Array[] = [];
|
||||
const first = dataTrack.data$.subscribe((p) => received.push(p));
|
||||
const second = dataTrack.data$.subscribe();
|
||||
expect(track.subscribe).toHaveBeenCalledTimes(1);
|
||||
frames[0].enqueue({ payload: new Uint8Array([7]) });
|
||||
await flushPromises();
|
||||
expect(received).toEqual([new Uint8Array([7])]);
|
||||
|
||||
first.unsubscribe();
|
||||
expect(cancel).not.toHaveBeenCalled();
|
||||
second.unsubscribe();
|
||||
expect(cancel).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("takes a reliable track's streams from its publisher only", async () => {
|
||||
const { room, send } = roomWithStreams();
|
||||
const dataTrack = createLivekitDataTrack(
|
||||
room,
|
||||
remoteTrack("reliable:chat").track,
|
||||
logger,
|
||||
);
|
||||
const received: Uint8Array[] = [];
|
||||
const subscription = dataTrack.data$.subscribe((p) => received.push(p));
|
||||
|
||||
send("bob", [1]);
|
||||
send("alice", [2]);
|
||||
await flushPromises();
|
||||
|
||||
// Bob's is not this track's
|
||||
expect(received).toEqual([new Uint8Array([2])]);
|
||||
subscription.unsubscribe();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,74 @@
|
||||
/*
|
||||
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 Logger } from "matrix-js-sdk/lib/logger";
|
||||
import { type RemoteDataTrack, type Room as LivekitRoom } from "livekit-client";
|
||||
import { filter, map, Observable, share } from "rxjs";
|
||||
|
||||
import { type DataTrack } from "../../media-api";
|
||||
import { byteStreams$ } from "./ByteStreams";
|
||||
import { fromLivekitDataTrackName } from "./dataTrackName";
|
||||
|
||||
/**
|
||||
* A remote member's data track. Decryption is LiveKit's: its E2EE manager
|
||||
* reads the key index and IV off each frame or stream and decrypts with the
|
||||
* sender's key, so nothing here sees a key. A frame of an encrypted track that
|
||||
* does not decrypt is dropped by LiveKit. A stream is not checked: LiveKit
|
||||
* 2.23 sends an encrypted packet without its encryption type, so a received
|
||||
* stream reads as unencrypted either way, and a cleartext one would pass.
|
||||
*/
|
||||
export function createLivekitDataTrack(
|
||||
room: LivekitRoom,
|
||||
track: RemoteDataTrack,
|
||||
logger: Logger,
|
||||
): DataTrack {
|
||||
const { name, reliable } = fromLivekitDataTrackName(track.info.name);
|
||||
const encrypted = track.info.usesE2ee;
|
||||
const data$ = reliable
|
||||
? byteStreams$(room, track.info.name).pipe(
|
||||
filter(({ identity }) => identity === track.publisherIdentity),
|
||||
map(({ payload }) => payload),
|
||||
)
|
||||
: framesOf(track, logger);
|
||||
return {
|
||||
source: "data",
|
||||
kind: "data",
|
||||
id: track.info.sid,
|
||||
name,
|
||||
reliable,
|
||||
encrypted,
|
||||
data$,
|
||||
};
|
||||
}
|
||||
|
||||
/** The track's frames, subscribed at the SFU while observed. */
|
||||
function framesOf(
|
||||
track: RemoteDataTrack,
|
||||
logger: Logger,
|
||||
): Observable<Uint8Array> {
|
||||
return new Observable<Uint8Array>((subscriber) => {
|
||||
const reader = track.subscribe().getReader();
|
||||
let cancelled = false;
|
||||
void (async (): Promise<void> => {
|
||||
try {
|
||||
for (;;) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
subscriber.next(value.payload);
|
||||
}
|
||||
} catch (e) {
|
||||
// A failed subscription ends the stream; a host's pipeline survives it
|
||||
if (!cancelled) logger.warn(`Data track ${track.info.name} ended`, e);
|
||||
}
|
||||
subscriber.complete();
|
||||
})();
|
||||
return (): void => {
|
||||
cancelled = true;
|
||||
reader.cancel().catch(() => {});
|
||||
};
|
||||
}).pipe(share());
|
||||
}
|
||||
@@ -6,9 +6,12 @@ Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from "events";
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
ParticipantEvent,
|
||||
type RemoteDataTrack,
|
||||
type RemoteParticipant,
|
||||
RoomEvent,
|
||||
type Room as LivekitRoom,
|
||||
Track,
|
||||
type TrackPublication,
|
||||
@@ -27,6 +30,7 @@ describe("createLivekitMemberMedia", () => {
|
||||
testScope(),
|
||||
constant({ participants: [participant], room }),
|
||||
{ kind: E2eeType.NONE },
|
||||
logger,
|
||||
);
|
||||
expect(media.tracks$.value).toEqual([]);
|
||||
|
||||
@@ -54,6 +58,7 @@ describe("createLivekitMemberMedia", () => {
|
||||
testScope(),
|
||||
constant({ participants: [participant], room }),
|
||||
{ kind: E2eeType.NONE },
|
||||
logger,
|
||||
);
|
||||
emitter.emit(
|
||||
ParticipantEvent.TrackPublished,
|
||||
@@ -85,6 +90,44 @@ function publish(
|
||||
return publication;
|
||||
}
|
||||
|
||||
describe("a member's data tracks", () => {
|
||||
it("follow the room's data track events, after the media", () => {
|
||||
const { participant, publications, room, emitter } = fakes();
|
||||
const media = createLivekitMemberMedia(
|
||||
testScope(),
|
||||
constant({ participants: [participant], room }),
|
||||
{ kind: E2eeType.NONE },
|
||||
logger,
|
||||
);
|
||||
const camera = publish(publications, "TR_1", Track.Source.Camera);
|
||||
emitter.emit(ParticipantEvent.TrackPublished, camera);
|
||||
|
||||
const chat = {
|
||||
info: { sid: "DT_1", name: "reliable:chat", usesE2ee: true },
|
||||
publisherIdentity: participant.identity,
|
||||
} as unknown as RemoteDataTrack;
|
||||
(room as unknown as EventEmitter).emit(RoomEvent.DataTrackPublished, chat);
|
||||
const [video, data] = media.tracks$.value!;
|
||||
expect(video.source).toBe("camera");
|
||||
expect(data).toMatchObject({
|
||||
kind: "data",
|
||||
name: "chat",
|
||||
reliable: true,
|
||||
encrypted: true,
|
||||
});
|
||||
|
||||
// An unrelated media event keeps the same object
|
||||
emitter.emit(ParticipantEvent.TrackMuted, camera);
|
||||
expect(media.tracks$.value![1]).toBe(data);
|
||||
|
||||
(room as unknown as EventEmitter).emit(
|
||||
RoomEvent.DataTrackUnpublished,
|
||||
"DT_1",
|
||||
);
|
||||
expect(media.tracks$.value).toEqual([video]);
|
||||
});
|
||||
});
|
||||
|
||||
function fakes(): {
|
||||
participant: RemoteParticipant;
|
||||
publications: Map<string, TrackPublication>;
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
observeParticipantMedia,
|
||||
roomEventSelector,
|
||||
} from "@livekit/components-core";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
type LocalParticipant,
|
||||
type Participant,
|
||||
@@ -31,11 +32,14 @@ import { generateItems } from "../../reactive/observable";
|
||||
import { E2eeType, type EncryptionSystem } from "../../encryption";
|
||||
import { type Behavior } from "../../reactive/Behavior";
|
||||
import {
|
||||
type DataTrack,
|
||||
type EncryptionError,
|
||||
type LocalMemberMedia,
|
||||
type MediaSource,
|
||||
type MemberMedia,
|
||||
} from "../../media-api";
|
||||
import { createLivekitDataTrack } from "./LivekitDataTrack";
|
||||
import { remoteDataTracks$ } from "./RemoteDataTracks";
|
||||
import {
|
||||
createLivekitMediaTrack,
|
||||
createLocalLivekitMediaTrack,
|
||||
@@ -70,13 +74,21 @@ export function createLivekitMemberMedia(
|
||||
scope: ObservableScope,
|
||||
source$: Behavior<ParticipantSource>,
|
||||
encryptionSystem: EncryptionSystem,
|
||||
logger: Logger,
|
||||
): MemberMedia {
|
||||
const media$ = memberTracks$(
|
||||
scope,
|
||||
source$,
|
||||
(trackScope, participant, publication, room) =>
|
||||
createLivekitMediaTrack(trackScope, participant, publication, room),
|
||||
);
|
||||
const data$ = memberDataTracks$(scope, source$, logger);
|
||||
return {
|
||||
tracks$: memberTracks$(
|
||||
scope,
|
||||
source$,
|
||||
(trackScope, participant, publication, room) =>
|
||||
createLivekitMediaTrack(trackScope, participant, publication, room),
|
||||
// The data tracks after the media, both null while there is no participant
|
||||
tracks$: scope.behavior(
|
||||
combineLatest([media$, data$], (media, data) =>
|
||||
media === null ? null : [...media, ...data],
|
||||
),
|
||||
),
|
||||
encryptionError$: encryptionErrors$(source$, encryptionSystem),
|
||||
};
|
||||
@@ -145,6 +157,40 @@ function memberTracks$<P extends Participant, T>(
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* One data track per LiveKit data track of the member's participants, each
|
||||
* living as long as it is published.
|
||||
*/
|
||||
function memberDataTracks$(
|
||||
scope: ObservableScope,
|
||||
source$: Behavior<ParticipantSource>,
|
||||
logger: Logger,
|
||||
): Behavior<DataTrack[]> {
|
||||
return scope.behavior(
|
||||
source$.pipe(
|
||||
switchMap((source) => {
|
||||
if (source === null) return of([]);
|
||||
const { participants, room } = source;
|
||||
const identities = new Set(participants.map((p) => p.identity));
|
||||
return remoteDataTracks$(room).pipe(
|
||||
map((tracks) =>
|
||||
tracks.filter((t) => identities.has(t.publisherIdentity)),
|
||||
),
|
||||
generateItems(
|
||||
`${[...identities].join(",")} dataTracks$`,
|
||||
function* (tracks) {
|
||||
for (const track of tracks)
|
||||
yield { keys: [track.info.sid], data: track };
|
||||
},
|
||||
(_trackScope, track$) =>
|
||||
createLivekitDataTrack(room, track$.value, logger),
|
||||
),
|
||||
);
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
function encryptionErrors$(
|
||||
source$: Behavior<ParticipantSource>,
|
||||
encryptionSystem: EncryptionSystem,
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
/*
|
||||
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 {
|
||||
ConnectionState as LivekitConnectionState,
|
||||
type LocalDataTrack as LivekitLocalDataTrack,
|
||||
type LocalParticipant,
|
||||
} from "livekit-client";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
BehaviorSubject,
|
||||
combineLatest,
|
||||
distinctUntilChanged,
|
||||
first,
|
||||
map,
|
||||
type Observable,
|
||||
of,
|
||||
startWith,
|
||||
Subject,
|
||||
switchMap,
|
||||
} from "rxjs";
|
||||
|
||||
import { type DataPublishRequest, type LocalDataTrack } from "../../media-api";
|
||||
import { type Behavior } from "../../reactive/Behavior";
|
||||
import { ObservableScope } from "../../reactive/ObservableScope";
|
||||
import { type Connection } from "./Connection";
|
||||
import { toLivekitDataTrackName } from "./dataTrackName";
|
||||
|
||||
interface Props {
|
||||
scope: ObservableScope;
|
||||
/** The connection the local member publishes on. */
|
||||
connection$: Behavior<Connection | null>;
|
||||
/** Whether anything may leave this device: joined and the homeserver reachable. */
|
||||
publishing$: Observable<boolean>;
|
||||
/** Whether the rooms carry a key provider, so LiveKit encrypts what we send. */
|
||||
encrypted: boolean;
|
||||
logger: Logger;
|
||||
}
|
||||
|
||||
export interface LocalDataTracks {
|
||||
tracks$: Behavior<LocalDataTrack[]>;
|
||||
/**
|
||||
* Resolves once the track is published on LiveKit, which waits for the
|
||||
* join. Rejects where a track of that name is published already, or where
|
||||
* LiveKit refuses it.
|
||||
*/
|
||||
publish(request: DataPublishRequest): Promise<LocalDataTrack>;
|
||||
unpublish(track: LocalDataTrack): void;
|
||||
}
|
||||
|
||||
interface Published {
|
||||
track: LivekitLocalDataTrack;
|
||||
participant: LocalParticipant;
|
||||
}
|
||||
|
||||
interface Entry {
|
||||
track: LocalDataTrack;
|
||||
finish: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* The local member's data tracks. Each is one LiveKit data track, published
|
||||
* on every connection to the local transport while publishing is allowed, so
|
||||
* a track requested before the join goes out once we are joined. The object a
|
||||
* host holds outlives the connections. Encryption is LiveKit's: its E2EE
|
||||
* manager encrypts with our current key and writes the key index and IV into
|
||||
* each frame, or each stream for a reliable track.
|
||||
*/
|
||||
export function createLocalDataTracks({
|
||||
scope,
|
||||
connection$,
|
||||
publishing$,
|
||||
encrypted,
|
||||
logger,
|
||||
}: Props): LocalDataTracks {
|
||||
const entries = new Map<string, Entry>();
|
||||
const tracks$ = new BehaviorSubject<LocalDataTrack[]>([]);
|
||||
const update = (): void =>
|
||||
tracks$.next([...entries.values()].map(({ track }) => track));
|
||||
scope.onEnd(() => {
|
||||
for (const entry of entries.values()) entry.finish();
|
||||
});
|
||||
|
||||
// Where to publish: the local connection once it has connected, while
|
||||
// publishing is allowed
|
||||
const target$ = scope.behavior(
|
||||
combineLatest([connection$, publishing$]).pipe(
|
||||
switchMap(([connection, publishing]) =>
|
||||
connection === null || !publishing
|
||||
? of(null)
|
||||
: connection.state$.pipe(
|
||||
first(
|
||||
(s) =>
|
||||
s.state === "authenticated" &&
|
||||
s.livekitState === LivekitConnectionState.Connected,
|
||||
),
|
||||
map(() => connection),
|
||||
startWith(null),
|
||||
),
|
||||
),
|
||||
distinctUntilChanged(),
|
||||
),
|
||||
);
|
||||
|
||||
const publish = async ({
|
||||
name,
|
||||
reliable = true,
|
||||
}: DataPublishRequest): Promise<LocalDataTrack> => {
|
||||
if (entries.has(name))
|
||||
throw new Error(`A data track named ${name} is published already`);
|
||||
const firstPublished = Promise.withResolvers<void>();
|
||||
const livekitName = toLivekitDataTrackName(name, reliable);
|
||||
const trackScope = new ObservableScope();
|
||||
const published$ = new BehaviorSubject<Published | null>(null);
|
||||
const sent$ = new Subject<Uint8Array>();
|
||||
let finished = false;
|
||||
|
||||
trackScope.reconcile(target$, async (connection) => {
|
||||
if (connection === null) return;
|
||||
const participant = connection.livekitRoom.localParticipant;
|
||||
let track: LivekitLocalDataTrack;
|
||||
try {
|
||||
track = await participant.publishDataTrack({ name: livekitName });
|
||||
} catch (e) {
|
||||
logger.error(`Failed to publish the data track ${name}`, e);
|
||||
firstPublished.reject(e);
|
||||
return;
|
||||
}
|
||||
published$.next({ track, participant });
|
||||
firstPublished.resolve();
|
||||
return async (): Promise<void> => {
|
||||
published$.next(null);
|
||||
await track.unpublish().catch((e: unknown) => {
|
||||
logger.warn(`Failed to unpublish the data track ${name}`, e);
|
||||
});
|
||||
};
|
||||
});
|
||||
|
||||
const track: LocalDataTrack = {
|
||||
source: "data",
|
||||
kind: "data",
|
||||
id: `data:${name}`,
|
||||
name,
|
||||
reliable,
|
||||
encrypted,
|
||||
data$: sent$.asObservable(),
|
||||
send: async (payload): Promise<void> => {
|
||||
if (finished) throw new Error(`The data track ${name} is unpublished`);
|
||||
const published = published$.value;
|
||||
if (published === null)
|
||||
throw new Error(
|
||||
`Cannot send on ${name}: not published on the transport yet`,
|
||||
);
|
||||
if (reliable)
|
||||
await published.participant.sendBytes(payload, {
|
||||
topic: livekitName,
|
||||
});
|
||||
else
|
||||
await published.track.tryPush({ payload: new Uint8Array(payload) });
|
||||
sent$.next(payload);
|
||||
},
|
||||
};
|
||||
|
||||
entries.set(name, {
|
||||
track,
|
||||
finish: (): void => {
|
||||
finished = true;
|
||||
trackScope.end();
|
||||
sent$.complete();
|
||||
},
|
||||
});
|
||||
update();
|
||||
try {
|
||||
await firstPublished.promise;
|
||||
} catch (e) {
|
||||
unpublish(track);
|
||||
throw e;
|
||||
}
|
||||
return track;
|
||||
};
|
||||
|
||||
const unpublish = (track: LocalDataTrack): void => {
|
||||
const entry = entries.get(track.name);
|
||||
if (entry?.track !== track) return;
|
||||
entries.delete(track.name);
|
||||
entry.finish();
|
||||
update();
|
||||
};
|
||||
|
||||
return { tracks$: scope.behavior(tracks$), publish, unpublish };
|
||||
}
|
||||
@@ -7,6 +7,7 @@ Please see LICENSE in the repository root for full details.
|
||||
|
||||
import {
|
||||
ConnectionState as LivekitConnectionState,
|
||||
type LocalDataTrack as LivekitLocalDataTrack,
|
||||
type LocalParticipant,
|
||||
type LocalTrack,
|
||||
type LocalTrackPublication,
|
||||
@@ -28,6 +29,10 @@ import { type Connection, type ConnectionState } from "./Connection";
|
||||
import { ConnectionMap } from "./ConnectionMap";
|
||||
import { createLivekitLocalMedia } from "./LocalMedia";
|
||||
import { exampleTransport } from "./test";
|
||||
import {
|
||||
type LocalAudioMediaTrack,
|
||||
type LocalDataTrack,
|
||||
} from "../../media-api";
|
||||
|
||||
const localTransport: TransportLocator = {
|
||||
transport: exampleTransport,
|
||||
@@ -270,9 +275,8 @@ describe("createLivekitLocalMedia", () => {
|
||||
media.setPublishing(true);
|
||||
await flushPromises();
|
||||
|
||||
await expect(media.tracks$.value![0].setEnabled(false)).resolves.toBe(
|
||||
false,
|
||||
);
|
||||
const microphone = media.tracks$.value![0] as LocalAudioMediaTrack;
|
||||
await expect(microphone.setEnabled(false)).resolves.toBe(false);
|
||||
expect(first.participant.setMicrophoneEnabled).toHaveBeenLastCalledWith(
|
||||
false,
|
||||
);
|
||||
@@ -293,10 +297,16 @@ describe("createLivekitLocalMedia", () => {
|
||||
await flushPromises();
|
||||
await media.publish({ source: "screenShare" });
|
||||
|
||||
await media.unpublish("screenShareAudio");
|
||||
await media.unpublish({
|
||||
kind: "audio",
|
||||
source: "screenShareAudio",
|
||||
} as LocalAudioMediaTrack);
|
||||
expect(participant.setScreenShareEnabled).toHaveBeenLastCalledWith(false);
|
||||
|
||||
await media.unpublish("microphone");
|
||||
await media.unpublish({
|
||||
kind: "audio",
|
||||
source: "microphone",
|
||||
} as LocalAudioMediaTrack);
|
||||
expect(participant.unpublishTrack).toHaveBeenCalledOnce();
|
||||
expect(media.tracks$.value).toEqual([]);
|
||||
});
|
||||
@@ -385,6 +395,116 @@ function fakeLocalTrack(source: Track.Source): LocalTrack {
|
||||
return t as unknown as LocalTrack;
|
||||
}
|
||||
|
||||
describe("local data tracks", () => {
|
||||
function joined(): ReturnType<typeof setup> & FakeConnection {
|
||||
const fixture = setup();
|
||||
const fake = fakeConnection();
|
||||
fixture.connectionMap$.next(mapOf(fake.connection));
|
||||
fixture.localTransport$.next(localTransport);
|
||||
return { ...fixture, ...fake };
|
||||
}
|
||||
|
||||
it("publishes on LiveKit only once publishing is allowed, and again on a new connection", async () => {
|
||||
const { media, participant, connectionMap$ } = joined();
|
||||
let published = false;
|
||||
const publishing = media
|
||||
.publish({ source: "data", name: "chat" })
|
||||
.then((track) => {
|
||||
published = true;
|
||||
return track;
|
||||
});
|
||||
await flushPromises();
|
||||
expect(participant.publishDataTrack).not.toHaveBeenCalled();
|
||||
expect(published).toBe(false);
|
||||
// Listed from the request on, like a media publication
|
||||
const listed = media.tracks$.value!.find((t) => t.kind === "data");
|
||||
expect(listed).toMatchObject({ name: "chat", reliable: true });
|
||||
|
||||
media.setPublishing(true);
|
||||
const chat = await publishing;
|
||||
expect(chat).toBe(listed);
|
||||
expect(participant.publishDataTrack).toHaveBeenCalledWith({
|
||||
name: "reliable:chat",
|
||||
});
|
||||
|
||||
const next = fakeConnection();
|
||||
connectionMap$.next(mapOf(next.connection));
|
||||
await flushPromises();
|
||||
expect(next.participant.publishDataTrack).toHaveBeenCalledWith({
|
||||
name: "reliable:chat",
|
||||
});
|
||||
// The host's object outlives the connection
|
||||
expect(media.tracks$.value).toContain(chat);
|
||||
});
|
||||
|
||||
it("sends reliable data as a byte stream and lossy data as frames, replaying what was sent", async () => {
|
||||
const { media, participant } = joined();
|
||||
media.setPublishing(true);
|
||||
const chat = await media.publish({ source: "data", name: "chat" });
|
||||
const cursors = await media.publish({
|
||||
source: "data",
|
||||
name: "cursors",
|
||||
reliable: false,
|
||||
});
|
||||
await flushPromises();
|
||||
const sent: Uint8Array[] = [];
|
||||
chat.data$.subscribe((p) => sent.push(p));
|
||||
|
||||
await chat.send(new Uint8Array([1]));
|
||||
expect(participant.sendBytes).toHaveBeenCalledWith(new Uint8Array([1]), {
|
||||
topic: "reliable:chat",
|
||||
});
|
||||
expect(sent).toEqual([new Uint8Array([1])]);
|
||||
|
||||
await cursors.send(new Uint8Array([2]));
|
||||
const livekitCursors = await vi.mocked(participant.publishDataTrack).mock
|
||||
.results[1].value;
|
||||
expect(livekitCursors.tryPush).toHaveBeenCalledWith({
|
||||
payload: new Uint8Array([2]),
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects a second track of the same name, or one LiveKit refuses", async () => {
|
||||
const { media, participant } = joined();
|
||||
media.setPublishing(true);
|
||||
await media.publish({ source: "data", name: "chat" });
|
||||
await expect(
|
||||
media.publish({ source: "data", name: "chat", reliable: false }),
|
||||
).rejects.toThrow("published already");
|
||||
|
||||
vi.mocked(participant.publishDataTrack).mockRejectedValueOnce(
|
||||
new Error("DuplicateName"),
|
||||
);
|
||||
await expect(
|
||||
media.publish({ source: "data", name: "cursors" }),
|
||||
).rejects.toThrow("DuplicateName");
|
||||
expect(
|
||||
media.tracks$.value!.filter((t) => t.kind === "data").map((t) => t.name),
|
||||
).toEqual(["chat"]);
|
||||
});
|
||||
|
||||
it("finishes a track on unpublish", async () => {
|
||||
const { media, participant } = joined();
|
||||
media.setPublishing(true);
|
||||
const chat: LocalDataTrack = await media.publish({
|
||||
source: "data",
|
||||
name: "chat",
|
||||
});
|
||||
await flushPromises();
|
||||
let completed = false;
|
||||
chat.data$.subscribe({ complete: () => (completed = true) });
|
||||
|
||||
await media.unpublish(chat);
|
||||
await flushPromises();
|
||||
const livekitChat = await vi.mocked(participant.publishDataTrack).mock
|
||||
.results[0].value;
|
||||
expect(livekitChat.unpublish).toHaveBeenCalled();
|
||||
expect(completed).toBe(true);
|
||||
expect(media.tracks$.value).not.toContain(chat);
|
||||
await expect(chat.send(new Uint8Array([1]))).rejects.toThrow("unpublished");
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* The slice of a local participant the publisher and the local tracks use:
|
||||
* enabling a source publishes a fake track at once, muting an existing one
|
||||
@@ -442,6 +562,14 @@ function fakeLocalParticipant(): LocalParticipant {
|
||||
|
||||
const participant = {
|
||||
...emitter,
|
||||
publishDataTrack: vi.fn(async ({ name }: { name: string }) =>
|
||||
Promise.resolve({
|
||||
name,
|
||||
tryPush: vi.fn(async () => Promise.resolve()),
|
||||
unpublish: vi.fn(async () => Promise.resolve()),
|
||||
} as unknown as LivekitLocalDataTrack),
|
||||
),
|
||||
sendBytes: vi.fn(async () => Promise.resolve({})),
|
||||
identity: "@alice:example.org:DEVICE",
|
||||
isLocal: true,
|
||||
isSpeaking: false,
|
||||
|
||||
@@ -26,13 +26,16 @@ import {
|
||||
} from "rxjs";
|
||||
|
||||
import {
|
||||
type DataPublishRequest,
|
||||
type LocalAudioMediaTrack,
|
||||
type LocalDataTrack,
|
||||
type LocalMedia,
|
||||
type LocalTrack,
|
||||
type LocalVideoMediaTrack,
|
||||
type MediaSource,
|
||||
type PublishRequest,
|
||||
} from "../../media-api";
|
||||
import { type EncryptionSystem } from "../../encryption";
|
||||
import { E2eeType, type EncryptionSystem } from "../../encryption";
|
||||
import { FailToStartLivekitConnection, toMatrixRTCError } from "../../errors";
|
||||
import { type Behavior } from "../../reactive/Behavior";
|
||||
import {
|
||||
@@ -50,6 +53,7 @@ import {
|
||||
} from "./LivekitMemberMedia";
|
||||
import { livekitSources } from "./LivekitMediaTrack";
|
||||
import { type DesiredMedia, Publisher } from "./Publisher";
|
||||
import { createLocalDataTracks } from "./LocalDataTracks";
|
||||
import { toMediaConnectionState } from "./connectionState";
|
||||
|
||||
interface Props {
|
||||
@@ -67,7 +71,7 @@ interface Props {
|
||||
}
|
||||
|
||||
export interface LivekitLocalMedia extends LocalMediaBackend {
|
||||
/** The connection the local member publishes on; the data channel sends on it. */
|
||||
/** The connection the local member publishes on. */
|
||||
connection$: Behavior<Connection | null>;
|
||||
setAudioOutputDeviceId(deviceId: string): Promise<void>;
|
||||
}
|
||||
@@ -171,14 +175,42 @@ export function createLivekitLocalMedia({
|
||||
return result;
|
||||
};
|
||||
|
||||
const { tracks$, encryptionError$ } = createLocalLivekitMemberMedia(
|
||||
const { tracks$: mediaTracks$, encryptionError$ } =
|
||||
createLocalLivekitMemberMedia(
|
||||
scope,
|
||||
participantAndRoom$(scope, participant$, connection$),
|
||||
encryptionSystem,
|
||||
setEnabled,
|
||||
);
|
||||
const dataTracks = createLocalDataTracks({
|
||||
scope,
|
||||
participantAndRoom$(scope, participant$, connection$),
|
||||
encryptionSystem,
|
||||
setEnabled,
|
||||
connection$,
|
||||
publishing$,
|
||||
encrypted: encryptionSystem.kind !== E2eeType.NONE,
|
||||
logger,
|
||||
});
|
||||
// The data tracks after the media, both null while there is no participant
|
||||
const tracks$ = scope.behavior<LocalTrack[] | null>(
|
||||
combineLatest([mediaTracks$, dataTracks.tracks$], (media, data) =>
|
||||
media === null ? null : [...media, ...data],
|
||||
),
|
||||
);
|
||||
|
||||
const publish = async (
|
||||
function publish(request: DataPublishRequest): Promise<LocalDataTrack>;
|
||||
function publish(
|
||||
request: PublishRequest,
|
||||
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack>;
|
||||
async function publish(
|
||||
request: DataPublishRequest | PublishRequest,
|
||||
): Promise<LocalTrack> {
|
||||
if (request.source === "data")
|
||||
return dataTracks.publish(request).catch((e: unknown) => {
|
||||
throw toMatrixRTCError(e);
|
||||
});
|
||||
return publishMedia(request);
|
||||
}
|
||||
|
||||
const publishMedia = async (
|
||||
request: PublishRequest,
|
||||
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack> => {
|
||||
desired.set(request.source, { request, enabled: true });
|
||||
@@ -195,8 +227,9 @@ export function createLivekitLocalMedia({
|
||||
}
|
||||
};
|
||||
|
||||
const unpublish = async (source: MediaSource): Promise<void> => {
|
||||
const published = publishedSource(source);
|
||||
const unpublish = async (track: LocalTrack): Promise<void> => {
|
||||
if (track.kind === "data") return dataTracks.unpublish(track);
|
||||
const published = publishedSource(track.source);
|
||||
if (published === undefined) return;
|
||||
desired.delete(published);
|
||||
await publisher$.value?.unpublish(published);
|
||||
@@ -259,11 +292,16 @@ function publication$(
|
||||
}
|
||||
|
||||
function trackWithId$(
|
||||
tracks$: Behavior<(LocalAudioMediaTrack | LocalVideoMediaTrack)[] | null>,
|
||||
tracks$: Behavior<LocalTrack[] | null>,
|
||||
id: string,
|
||||
): Observable<LocalAudioMediaTrack | LocalVideoMediaTrack> {
|
||||
return tracks$.pipe(
|
||||
map((tracks) => tracks?.find((track) => track.id === id)),
|
||||
map((tracks) =>
|
||||
tracks?.find(
|
||||
(track): track is LocalAudioMediaTrack | LocalVideoMediaTrack =>
|
||||
track.kind !== "data" && track.id === id,
|
||||
),
|
||||
),
|
||||
filter((track) => track !== undefined),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -72,9 +72,11 @@ export class Publisher {
|
||||
private readonly media?: LocalMedia,
|
||||
) {
|
||||
this.room = room;
|
||||
room.setE2EEEnabled(room.options.e2ee !== undefined)?.catch((e: Error) => {
|
||||
this.logger.error("Failed to enable E2EE on the room", e);
|
||||
});
|
||||
room
|
||||
.setE2EEEnabled(room.options.encryption !== undefined)
|
||||
?.catch((e: Error) => {
|
||||
this.logger.error("Failed to enable E2EE on the room", e);
|
||||
});
|
||||
this.onLocalTrackPublished = this.onLocalTrackPublished.bind(this);
|
||||
room.localParticipant.on(
|
||||
ParticipantEvent.LocalTrackPublished,
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
/*
|
||||
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 RemoteDataTrack,
|
||||
type Room as LivekitRoom,
|
||||
RoomEvent,
|
||||
} from "livekit-client";
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { remoteDataTracks$, watchRemoteDataTracks } from "./RemoteDataTracks";
|
||||
|
||||
const track = (sid: string): RemoteDataTrack =>
|
||||
({ info: { sid }, publisherIdentity: "alice" }) as RemoteDataTrack;
|
||||
|
||||
describe("remoteDataTracks$", () => {
|
||||
it("keeps what the room announced before anyone observed it", () => {
|
||||
const emitter = new EventEmitter();
|
||||
const room = watchRemoteDataTracks(emitter as unknown as LivekitRoom);
|
||||
// As in a join response: announced before any member asks
|
||||
emitter.emit(RoomEvent.DataTrackPublished, track("DT_1"));
|
||||
emitter.emit(RoomEvent.DataTrackPublished, track("DT_2"));
|
||||
emitter.emit(RoomEvent.DataTrackUnpublished, "DT_1");
|
||||
|
||||
const seen: string[][] = [];
|
||||
remoteDataTracks$(room).subscribe((tracks) =>
|
||||
seen.push(tracks.map((t) => t.info.sid)),
|
||||
);
|
||||
expect(seen).toEqual([["DT_2"]]);
|
||||
|
||||
emitter.emit(RoomEvent.Disconnected);
|
||||
expect(seen.at(-1)).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,60 @@
|
||||
/*
|
||||
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 RemoteDataTrack,
|
||||
type Room as LivekitRoom,
|
||||
RoomEvent,
|
||||
} from "livekit-client";
|
||||
import { BehaviorSubject, type Observable } from "rxjs";
|
||||
|
||||
const tracksByRoom = new WeakMap<
|
||||
LivekitRoom,
|
||||
BehaviorSubject<readonly RemoteDataTrack[]>
|
||||
>();
|
||||
|
||||
/**
|
||||
* Starts keeping the room's remote data tracks, from its own events. Called
|
||||
* as the room is created: LiveKit announces the tracks in the join response
|
||||
* before the participants that publish them exist, so a participant's
|
||||
* `dataTracks` misses them, and it adds a track to that map only after
|
||||
* announcing it.
|
||||
*/
|
||||
export function watchRemoteDataTracks(room: LivekitRoom): LivekitRoom {
|
||||
remoteDataTracksOf(room);
|
||||
return room;
|
||||
}
|
||||
|
||||
/** Every remote data track the room knows of, as LiveKit announced them. */
|
||||
export function remoteDataTracks$(
|
||||
room: LivekitRoom,
|
||||
): Observable<readonly RemoteDataTrack[]> {
|
||||
return remoteDataTracksOf(room);
|
||||
}
|
||||
|
||||
function remoteDataTracksOf(
|
||||
room: LivekitRoom,
|
||||
): BehaviorSubject<readonly RemoteDataTrack[]> {
|
||||
let tracks$ = tracksByRoom.get(room);
|
||||
if (tracks$ === undefined) {
|
||||
const subject$ = new BehaviorSubject<readonly RemoteDataTrack[]>([]);
|
||||
room.on(RoomEvent.DataTrackPublished, (track: RemoteDataTrack) =>
|
||||
subject$.next([
|
||||
...subject$.value.filter((t) => t.info.sid !== track.info.sid),
|
||||
track,
|
||||
]),
|
||||
);
|
||||
room.on(RoomEvent.DataTrackUnpublished, (sid: string) =>
|
||||
subject$.next(subject$.value.filter((t) => t.info.sid !== sid)),
|
||||
);
|
||||
// A new session announces its tracks again
|
||||
room.on(RoomEvent.Disconnected, () => subject$.next([]));
|
||||
tracksByRoom.set(room, subject$);
|
||||
tracks$ = subject$;
|
||||
}
|
||||
return tracks$;
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
/*
|
||||
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 { describe, expect, it } from "vitest";
|
||||
|
||||
import {
|
||||
fromLivekitDataTrackName,
|
||||
toLivekitDataTrackName,
|
||||
} from "./dataTrackName";
|
||||
|
||||
describe("data track names on LiveKit", () => {
|
||||
it.each([true, false])("round-trips reliable=%s", (reliable) => {
|
||||
expect(
|
||||
fromLivekitDataTrackName(toLivekitDataTrackName("chat:room", reliable)),
|
||||
).toEqual({ name: "chat:room", reliable });
|
||||
});
|
||||
|
||||
it("reads a name without a prefix as lossy", () => {
|
||||
expect(fromLivekitDataTrackName("telemetry")).toEqual({
|
||||
name: "telemetry",
|
||||
reliable: false,
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,36 @@
|
||||
/*
|
||||
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.
|
||||
*/
|
||||
|
||||
/**
|
||||
* How a data track is named on LiveKit. Every data track of a member is one
|
||||
* LiveKit data track, which announces it; the prefix says how its payloads
|
||||
* travel. A lossy track pushes them as frames on the data track itself. A
|
||||
* reliable one sends each as a byte stream on a topic of the same name,
|
||||
* because LiveKit's data tracks are lossy only.
|
||||
*/
|
||||
|
||||
const lossyPrefix = "lossy:";
|
||||
const reliablePrefix = "reliable:";
|
||||
|
||||
export function toLivekitDataTrackName(
|
||||
name: string,
|
||||
reliable: boolean,
|
||||
): string {
|
||||
return (reliable ? reliablePrefix : lossyPrefix) + name;
|
||||
}
|
||||
|
||||
/** A track without a prefix, from a plain LiveKit publisher, is lossy under its own name. */
|
||||
export function fromLivekitDataTrackName(livekitName: string): {
|
||||
name: string;
|
||||
reliable: boolean;
|
||||
} {
|
||||
if (livekitName.startsWith(reliablePrefix))
|
||||
return { name: livekitName.slice(reliablePrefix.length), reliable: true };
|
||||
if (livekitName.startsWith(lossyPrefix))
|
||||
return { name: livekitName.slice(lossyPrefix.length), reliable: false };
|
||||
return { name: livekitName, reliable: false };
|
||||
}
|
||||
@@ -8,9 +8,9 @@ Please see LICENSE in the repository root for full details.
|
||||
import { BehaviorSubject } from "rxjs";
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { type MediaTrack } from "../media-api";
|
||||
import { type DataTrack, type MediaTrack } from "../media-api";
|
||||
import { testScope } from "./test";
|
||||
import { trackBySource$ } from "./tracks";
|
||||
import { dataTrackByName$, trackBySource$ } from "./tracks";
|
||||
|
||||
const camera = { source: "camera", kind: "video", id: "1" } as MediaTrack;
|
||||
const microphone = {
|
||||
@@ -39,3 +39,40 @@ describe("trackBySource$", () => {
|
||||
expect(emissions).toEqual([camera]);
|
||||
});
|
||||
});
|
||||
|
||||
const chat = {
|
||||
source: "data",
|
||||
kind: "data",
|
||||
id: "3",
|
||||
name: "chat",
|
||||
} as DataTrack;
|
||||
const cursors = {
|
||||
source: "data",
|
||||
kind: "data",
|
||||
id: "4",
|
||||
name: "cursors",
|
||||
} as DataTrack;
|
||||
|
||||
describe("dataTrackByName$", () => {
|
||||
it("picks the data track of a name and follows the list", () => {
|
||||
const tracks$ = new BehaviorSubject<(MediaTrack | DataTrack)[] | null>(
|
||||
null,
|
||||
);
|
||||
const chat$ = dataTrackByName$(testScope(), tracks$, "chat");
|
||||
expect(chat$.value).toBeUndefined();
|
||||
tracks$.next([camera, cursors, chat]);
|
||||
expect(chat$.value).toBe(chat);
|
||||
tracks$.next([camera, cursors]);
|
||||
expect(chat$.value).toBeUndefined();
|
||||
});
|
||||
|
||||
it("does not emit when the list changes elsewhere", () => {
|
||||
const tracks$ = new BehaviorSubject<(MediaTrack | DataTrack)[]>([chat]);
|
||||
const chat$ = dataTrackByName$(testScope(), tracks$, "chat");
|
||||
const emissions: unknown[] = [];
|
||||
chat$.subscribe((t) => emissions.push(t));
|
||||
tracks$.next([chat, camera]);
|
||||
tracks$.next([microphone, chat, cursors]);
|
||||
expect(emissions).toEqual([chat]);
|
||||
});
|
||||
});
|
||||
|
||||
+20
-1
@@ -11,6 +11,7 @@ import { type Behavior } from "../reactive/Behavior";
|
||||
import { type ObservableScope } from "../reactive/ObservableScope";
|
||||
import {
|
||||
type AudioMediaTrack,
|
||||
type DataTrack,
|
||||
type MediaSource,
|
||||
type MediaTrack,
|
||||
type VideoMediaTrack,
|
||||
@@ -27,7 +28,7 @@ export type TrackOfSource<S extends MediaSource> = S extends
|
||||
/** The member's first track of a source; undefined while there is none, or no media yet. */
|
||||
export function trackBySource$<S extends MediaSource>(
|
||||
scope: ObservableScope,
|
||||
tracks$: Behavior<MediaTrack[] | null>,
|
||||
tracks$: Behavior<readonly (MediaTrack | DataTrack)[] | null>,
|
||||
source: S,
|
||||
): Behavior<TrackOfSource<S> | undefined> {
|
||||
return scope.behavior(
|
||||
@@ -42,3 +43,21 @@ export function trackBySource$<S extends MediaSource>(
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/** The member's data track of this name; undefined while there is none, or no media yet. */
|
||||
export function dataTrackByName$(
|
||||
scope: ObservableScope,
|
||||
tracks$: Behavior<readonly (MediaTrack | DataTrack)[] | null>,
|
||||
name: string,
|
||||
): Behavior<DataTrack | undefined> {
|
||||
return scope.behavior(
|
||||
tracks$.pipe(
|
||||
map((tracks) =>
|
||||
tracks?.find(
|
||||
(t): t is DataTrack => t.kind === "data" && t.name === name,
|
||||
),
|
||||
),
|
||||
distinctUntilChanged(),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -494,7 +494,12 @@ export function createCallViewModel$(
|
||||
>(
|
||||
rtcParticipation.localMember$.pipe(
|
||||
switchMap((member) => member?.tracks$ ?? of(null)),
|
||||
map((tracks) => tracks ?? []),
|
||||
map((tracks) =>
|
||||
(tracks ?? []).filter(
|
||||
(t): t is LocalAudioMediaTrack | LocalVideoMediaTrack =>
|
||||
t.kind !== "data",
|
||||
),
|
||||
),
|
||||
),
|
||||
);
|
||||
// Our own tracks, which carry the controls remote ones lack
|
||||
@@ -1454,7 +1459,7 @@ export function createCallViewModel$(
|
||||
source: "screenShare",
|
||||
capture: capture.screenShare,
|
||||
})
|
||||
: rtcParticipation.unpublish("screenShare")
|
||||
: rtcParticipation.unpublish(track)
|
||||
).catch((e: unknown) => {
|
||||
logger.error(
|
||||
`Screen share ${track === undefined ? "start" : "stop"} failed`,
|
||||
|
||||
+14
-9
@@ -71,6 +71,8 @@ import {
|
||||
type RTCParticipation,
|
||||
type RTCSlot,
|
||||
type ParticipationState,
|
||||
type DataPublishRequest,
|
||||
type LocalTrack,
|
||||
type PublishRequest,
|
||||
type RemoteRTCMember,
|
||||
type TransportMetadata,
|
||||
@@ -535,19 +537,22 @@ export function mockRTCParticipation(
|
||||
state$,
|
||||
localMember$,
|
||||
remoteMembers$,
|
||||
publish: vi.fn(async (request: PublishRequest) =>
|
||||
Promise.resolve(
|
||||
request.source === "microphone"
|
||||
? mockAudioTrack()
|
||||
: mockVideoTrack({ source: request.source }),
|
||||
),
|
||||
),
|
||||
publish: vi.fn(
|
||||
async (
|
||||
request: PublishRequest | DataPublishRequest,
|
||||
): Promise<LocalTrack> => {
|
||||
if (request.source === "data") throw new Error("Not mocked");
|
||||
return Promise.resolve(
|
||||
request.source === "microphone"
|
||||
? mockAudioTrack()
|
||||
: mockVideoTrack({ source: request.source }),
|
||||
);
|
||||
},
|
||||
) as RTCParticipation["publish"],
|
||||
unpublish: vi.fn(async () => Promise.resolve()),
|
||||
keyRotationSuppressed$,
|
||||
connectedTransports$,
|
||||
setAudioOutputDeviceId: vi.fn(async () => Promise.resolve()),
|
||||
sendData: vi.fn(async () => Promise.resolve()),
|
||||
data$: NEVER,
|
||||
};
|
||||
participation$.next(participation);
|
||||
return participation;
|
||||
|
||||
Reference in New Issue
Block a user