Matrix/element-call-Github
1
0
Fork 0
mirror of https://github.com/vector-im/element-call.git synced 2026-03-04 05:37:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

mvp

Browse Source
This commit is contained in:
Timo K
2025-12-01 12:43:17 +01:00
parent 2d8ffc0ccd
commit 284a52c23c
12 changed files with 296 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
godot/index.html
View File

@@ -20,7 +20,7 @@
await window.matrixRTCSdk.join();
console.info("matrixRTCSdk joined ");
// sdk.data$.subscribe((data) => {
// window.matrixRTCSdk.data$.subscribe((data) => {
// console.log(data);
// const div = document.getElementById("data");
// div.appendChild(document.createTextNode(data));
@@ -36,9 +36,9 @@
</head>
<body>
<button onclick="window.matrixRTCSdk.leave();">Leave</button>
<!--<button onclick="window.matrixRTCSdk.sendData({prop: 'Hello, world!'});">
<button onclick="window.matrixRTCSdk.sendData({prop: 'Hello, world!'});">
Send Text
</button>-->
</button>
<div id="data"></div>
<canvas id="canvas"></canvas>
</body>

198
godot/main.ts
View File

@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
*/
// import { type InitResult } from "../src/ClientContext";
import { map, type Observable, of, Subject, switchMap } from "rxjs";
import { map, type Observable, of, Subject, switchMap, tap } from "rxjs";
import { MatrixRTCSessionEvent } from "matrix-js-sdk/lib/matrixrtc";
import { type TextStreamInfo } from "livekit-client/dist/src/room/types";
import {
@@ -36,7 +36,7 @@ interface MatrixRTCSdk {
/** @throws on leave errors */
leave: () => void;
data$: Observable<{ sender: string; data: string }>;
sendData?: (data: Record<string, unknown>) => Promise<void>;
sendData?: (data: unknown) => Promise<void>;
}
export async function createMatrixRTCSdk(): Promise<MatrixRTCSdk> {
logger.info("Hello");
@@ -67,97 +67,115 @@ export async function createMatrixRTCSdk(): Promise<MatrixRTCSdk> {
// create data listener
const data$ = new Subject<{ sender: string; data: string }>();
// const lkTextStreamHandlerFunction = async (
// reader: TextStreamReader,
// participantInfo: { identity: string },
// livekitRoom: LivekitRoom,
// ): Promise<void> => {
// const info = reader.info;
// console.log(
// `Received text stream from ${participantInfo.identity}\n` +
// ` Topic: ${info.topic}\n` +
// ` Timestamp: ${info.timestamp}\n` +
// ` ID: ${info.id}\n` +
// ` Size: ${info.size}`, // Optional, only available if the stream was sent with `sendText`
// );
const lkTextStreamHandlerFunction = async (
reader: TextStreamReader,
participantInfo: { identity: string },
livekitRoom: LivekitRoom,
): Promise<void> => {
const info = reader.info;
logger.info(
`Received text stream from ${participantInfo.identity}\n` +
` Topic: ${info.topic}\n` +
` Timestamp: ${info.timestamp}\n` +
` ID: ${info.id}\n` +
` Size: ${info.size}`, // Optional, only available if the stream was sent with `sendText`
);
// const participants = callViewModel.livekitRoomItems$.value.find(
// (i) => i.livekitRoom === livekitRoom,
// )?.participants;
// if (participants && participants.includes(participantInfo.identity)) {
// const text = await reader.readAll();
// console.log(`Received text: ${text}`);
// data$.next({ sender: participantInfo.identity, data: text });
// } else {
// logger.warn(
// "Received text from unknown participant",
// participantInfo.identity,
// );
// }
// };
const participants = callViewModel.livekitRoomItems$.value.find(
(i) => i.livekitRoom === livekitRoom,
)?.participants;
if (participants && participants.includes(participantInfo.identity)) {
const text = await reader.readAll();
logger.info(`Received text: ${text}`);
data$.next({ sender: participantInfo.identity, data: text });
} else {
logger.warn(
"Received text from unknown participant",
participantInfo.identity,
);
}
};
// const livekitRoomItemsSub = callViewModel.livekitRoomItems$
// .pipe(currentAndPrev)
// .subscribe({
// next: ({ prev, current }) => {
// const prevRooms = prev.map((i) => i.livekitRoom);
// const currentRooms = current.map((i) => i.livekitRoom);
// const addedRooms = currentRooms.filter((r) => !prevRooms.includes(r));
// const removedRooms = prevRooms.filter((r) => !currentRooms.includes(r));
// addedRooms.forEach((r) =>
// r.registerTextStreamHandler(
// TEXT_LK_TOPIC,
// (reader, participantInfo) =>
// void lkTextStreamHandlerFunction(reader, participantInfo, r),
// ),
// );
// removedRooms.forEach((r) =>
// r.unregisterTextStreamHandler(TEXT_LK_TOPIC),
// );
// },
// complete: () => {
// logger.info("Livekit room items subscription completed");
// for (const item of callViewModel.livekitRoomItems$.value) {
// logger.info("unregistering room item from room", item.url);
// item.livekitRoom.unregisterTextStreamHandler(TEXT_LK_TOPIC);
// }
// },
// });
const livekitRoomItemsSub = callViewModel.livekitRoomItems$
.pipe(
tap((beforecurrentAndPrev) => {
logger.info(
`LiveKit room items updated: ${beforecurrentAndPrev.length}`,
beforecurrentAndPrev,
);
}),
currentAndPrev,
tap((aftercurrentAndPrev) => {
logger.info(
`LiveKit room items updated: ${aftercurrentAndPrev.current.length}, ${aftercurrentAndPrev.prev.length}`,
aftercurrentAndPrev,
);
}),
)
.subscribe({
next: ({ prev, current }) => {
const prevRooms = prev.map((i) => i.livekitRoom);
const currentRooms = current.map((i) => i.livekitRoom);
const addedRooms = currentRooms.filter((r) => !prevRooms.includes(r));
const removedRooms = prevRooms.filter((r) => !currentRooms.includes(r));
addedRooms.forEach((r) => {
logger.info(`Registering text stream handler for room `);
r.registerTextStreamHandler(
TEXT_LK_TOPIC,
(reader, participantInfo) =>
void lkTextStreamHandlerFunction(reader, participantInfo, r),
);
});
removedRooms.forEach((r) => {
logger.info(`Unregistering text stream handler for room `);
r.unregisterTextStreamHandler(TEXT_LK_TOPIC);
});
},
complete: () => {
logger.info("Livekit room items subscription completed");
for (const item of callViewModel.livekitRoomItems$.value) {
logger.info("unregistering room item from room", item.url);
item.livekitRoom.unregisterTextStreamHandler(TEXT_LK_TOPIC);
}
},
});
// create sendData function
// const sendFn: Behavior<(data: string) => Promise<TextStreamInfo>> =
// scope.behavior(
// callViewModel.localMatrixLivekitMember$.pipe(
// switchMap((m) => {
// if (!m)
// return of((data: string): never => {
// throw Error("local membership not yet ready.");
// });
// return m.participant$.pipe(
// map((p) => {
// if (p === null) {
// return (data: string): never => {
// throw Error("local participant not yet ready to send data.");
// };
// } else {
// return async (data: string): Promise<TextStreamInfo> =>
// p.sendText(data, { topic: TEXT_LK_TOPIC });
// }
// }),
// );
// }),
// ),
// );
const sendFn: Behavior<(data: string) => Promise<TextStreamInfo>> =
scope.behavior(
callViewModel.localmatrixLivekitMembers$.pipe(
switchMap((m) => {
if (!m)
return of((data: string): never => {
throw Error("local membership not yet ready.");
});
return m.participant$.pipe(
map((p) => {
if (p === null) {
return (data: string): never => {
throw Error("local participant not yet ready to send data.");
};
} else {
return async (data: string): Promise<TextStreamInfo> =>
p.sendText(data, { topic: TEXT_LK_TOPIC });
}
}),
);
}),
),
);
// const sendData = async (data: Record<string, unknown>): Promise<void> => {
// const dataString = JSON.stringify(data);
// try {
// const info = await sendFn.value(dataString);
// logger.info(`Sent text with stream ID: ${info.id}`);
// } catch (e) {
// console.error("failed sending: ", dataString, e);
// }
// };
const sendData = async (data: unknown): Promise<void> => {
const dataString = JSON.stringify(data);
logger.info("try sending: ", dataString);
try {
await Promise.resolve();
const info = await sendFn.value(dataString);
logger.info(`Sent text with stream ID: ${info.id}`);
} catch (e) {
logger.error("failed sending: ", dataString, e);
}
};
// after hangup gets called
const leaveSubs = callViewModel.leave$.subscribe(() => {
@@ -202,9 +220,9 @@ export async function createMatrixRTCSdk(): Promise<MatrixRTCSdk> {
leave: (): void => {
callViewModel.hangup();
leaveSubs.unsubscribe();
// livekitRoomItemsSub.unsubscribe();
livekitRoomItemsSub.unsubscribe();
},
data$,
// sendData,
sendData,
};
}

24
src/state/CallViewModel/CallViewModel.ts
View File

@@ -264,7 +264,7 @@ export interface CallViewModel {
livekitRoomItems$: Behavior<LivekitRoomItem[]>;
/** use the layout instead, this is just for the godot export. */
userMedia$: Behavior<UserMedia[]>;
localMatrixLivekitMember$: Behavior<LocalMatrixLivekitMember | null>;
localmatrixLivekitMembers$: Behavior<LocalMatrixLivekitMember | null>;
/** List of participants raising their hand */
handsRaised$: Behavior<Record<string, RaisedHandInfo>>;
/** List of reactions. Keys are: membership.membershipId (currently predefined as: `${membershipEvent.userId}:${membershipEvent.deviceId}`)*/
@@ -449,7 +449,7 @@ export function createCallViewModel$(
logger: logger,
});
const matrixLivekitMembers$ = createMatrixLivekitMembers$({
const { matrixLivekitMembers$ } = createMatrixLivekitMembers$({
scope: scope,
membershipsWithTransport$:
membershipsAndTransports.membershipsWithTransport$,
@@ -515,7 +515,7 @@ export function createCallViewModel$(
userId: userId,
};
const localMatrixLivekitMember$: Behavior<LocalMatrixLivekitMember | null> =
const localmatrixLivekitMembers$: Behavior<LocalMatrixLivekitMember | null> =
scope.behavior(
localRtcMembership$.pipe(
switchMap((membership) => {
@@ -607,8 +607,11 @@ export function createCallViewModel$(
const reconnecting$ = localMembership.reconnecting$;
const pretendToBeDisconnected$ = reconnecting$;
const audioParticipants$ = scope.behavior(
const livekitRoomItems$ = scope.behavior(
matrixLivekitMembers$.pipe(
tap((val) => {
logger.debug("matrixLivekitMembers$ updated", val.value);
}),
switchMap((membersWithEpoch) => {
const members = membersWithEpoch.value;
const a$ = combineLatest(
@@ -649,6 +652,12 @@ export function createCallViewModel$(
return acc;
}, []),
),
tap((val) => {
logger.debug(
"livekitRoomItems$ updated",
val.map((v) => v.url),
);
}),
),
[],
);
@@ -676,7 +685,7 @@ export function createCallViewModel$(
*/
const userMedia$ = scope.behavior<UserMedia[]>(
combineLatest([
localMatrixLivekitMember$,
localmatrixLivekitMembers$,
matrixLivekitMembers$,
duplicateTiles.value$,
]).pipe(
@@ -1489,8 +1498,7 @@ export function createCallViewModel$(
),
participantCount$: participantCount$,
livekitRoomItems$: audioParticipants$,
livekitRoomItems$,
handsRaised$: handsRaised$,
reactions$: reactions$,
joinSoundEffect$: joinSoundEffect$,
@@ -1510,7 +1518,7 @@ export function createCallViewModel$(
pip$: pip$,
layout$: layout$,
userMedia$,
localMatrixLivekitMember$,
localmatrixLivekitMembers$,
tileStoreGeneration$: tileStoreGeneration$,
showSpotlightIndicators$: showSpotlightIndicators$,
showSpeakingIndicators$: showSpeakingIndicators$,

33
src/state/CallViewModel/localMember/Publisher.ts
View File

@@ -56,15 +56,15 @@ export class Publisher {
devices: MediaDevices,
private readonly muteStates: MuteStates,
trackerProcessorState$: Behavior<ProcessorState>,
private logger?: Logger,
private logger: Logger,
) {
this.logger?.info("[PublishConnection] Create LiveKit room");
this.logger.info("[PublishConnection] Create LiveKit room");
const { controlledAudioDevices } = getUrlParams();
const room = connection.livekitRoom;
room.setE2EEEnabled(room.options.e2ee !== undefined)?.catch((e: Error) => {
this.logger?.error("Failed to set E2EE enabled on room", e);
this.logger.error("Failed to set E2EE enabled on room", e);
});
// Setup track processor syncing (blur)
@@ -74,7 +74,7 @@ export class Publisher {
this.workaroundRestartAudioInputTrackChrome(devices, scope);
this.scope.onEnd(() => {
this.logger?.info(
this.logger.info(
"[PublishConnection] Scope ended -> stop publishing all tracks",
);
void this.stopPublishing();
@@ -132,13 +132,14 @@ export class Publisher {
video,
})
.catch((error) => {
this.logger?.error("Failed to create tracks", error);
this.logger.error("Failed to create tracks", error);
})) ?? [];
}
return this.tracks;
}
public async startPublishing(): Promise<LocalTrack[]> {
this.logger.info("Start publishing");
const lkRoom = this.connection.livekitRoom;
const { promise, resolve, reject } = Promise.withResolvers<void>();
const sub = this.connection.state$.subscribe((s) => {
@@ -150,7 +151,7 @@ export class Publisher {
reject(new Error("Failed to connect to LiveKit server"));
break;
default:
this.logger?.info("waiting for connection: ", s.state);
this.logger.info("waiting for connection: ", s.state);
}
});
try {
@@ -160,12 +161,14 @@ export class Publisher {
} finally {
sub.unsubscribe();
}
this.logger.info("publish ", this.tracks.length, "tracks");
for (const track of this.tracks) {
// TODO: handle errors? Needs the signaling connection to be up, but it has some retries internally
// with a timeout.
await lkRoom.localParticipant.publishTrack(track).catch((error) => {
this.logger?.error("Failed to publish track", error);
this.logger.error("Failed to publish track", error);
});
this.logger.info("published track ", track.kind, track.id);
// TODO: check if the connection is still active? and break the loop if not?
}
@@ -229,7 +232,7 @@ export class Publisher {
.getTrackPublication(Track.Source.Microphone)
?.audioTrack?.restartTrack()
.catch((e) => {
this.logger?.error(`Failed to restart audio device track`, e);
this.logger.error(`Failed to restart audio device track`, e);
});
}
});
@@ -249,7 +252,7 @@ export class Publisher {
selected$.pipe(scope.bind()).subscribe((device) => {
if (lkRoom.state != LivekitConnectionState.Connected) return;
// if (this.connectionState$.value !== ConnectionState.Connected) return;
this.logger?.info(
this.logger.info(
"[LivekitRoom] syncDevice room.getActiveDevice(kind) !== d.id :",
lkRoom.getActiveDevice(kind),
" !== ",
@@ -262,7 +265,7 @@ export class Publisher {
lkRoom
.switchActiveDevice(kind, device.id)
.catch((e: Error) =>
this.logger?.error(
this.logger.error(
`Failed to sync ${kind} device with LiveKit`,
e,
),
@@ -287,10 +290,7 @@ export class Publisher {
try {
await lkRoom.localParticipant.setMicrophoneEnabled(desired);
} catch (e) {
this.logger?.error(
"Failed to update LiveKit audio input mute state",
e,
);
this.logger.error("Failed to update LiveKit audio input mute state", e);
}
return lkRoom.localParticipant.isMicrophoneEnabled;
});
@@ -298,10 +298,7 @@ export class Publisher {
try {
await lkRoom.localParticipant.setCameraEnabled(desired);
} catch (e) {
this.logger?.error(
"Failed to update LiveKit video input mute state",
e,
);
this.logger.error("Failed to update LiveKit video input mute state", e);
}
return lkRoom.localParticipant.isCameraEnabled;
});

28
src/state/CallViewModel/remoteMembers/Connection.test.ts
View File

@@ -382,17 +382,15 @@ describe("Publishing participants observations", () => {
const bobIsAPublisher = Promise.withResolvers<void>();
const danIsAPublisher = Promise.withResolvers<void>();
const observedPublishers: PublishingParticipant[][] = [];
const s = connection.remoteParticipantsWithTracks$.subscribe(
(publishers) => {
observedPublishers.push(publishers);
if (publishers.some((p) => p.identity === "@bob:example.org:DEV111")) {
bobIsAPublisher.resolve();
}
if (publishers.some((p) => p.identity === "@dan:example.org:DEV333")) {
danIsAPublisher.resolve();
}
},
);
const s = connection.remoteParticipants$.subscribe((publishers) => {
observedPublishers.push(publishers);
if (publishers.some((p) => p.identity === "@bob:example.org:DEV111")) {
bobIsAPublisher.resolve();
}
if (publishers.some((p) => p.identity === "@dan:example.org:DEV333")) {
danIsAPublisher.resolve();
}
});
onTestFinished(() => s.unsubscribe());
// The publishingParticipants$ observable is derived from the current members of the
// livekitRoom and the rtc membership in order to publish the members that are publishing
@@ -437,11 +435,9 @@ describe("Publishing participants observations", () => {
const connection = setupRemoteConnection();
let observedPublishers: PublishingParticipant[][] = [];
const s = connection.remoteParticipantsWithTracks$.subscribe(
(publishers) => {
observedPublishers.push(publishers);
},
);
const s = connection.remoteParticipants$.subscribe((publishers) => {
observedPublishers.push(publishers);
});
onTestFinished(() => s.unsubscribe());
let participants: RemoteParticipant[] = [

29
src/state/CallViewModel/remoteMembers/Connection.ts
View File

@@ -19,7 +19,7 @@ import {
RoomEvent,
} from "livekit-client";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject, map, type Observable } from "rxjs";
import { BehaviorSubject, type Observable } from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
@@ -146,6 +146,10 @@ export class Connection {
transport: this.transport,
livekitConnectionState$: connectionStateObserver(this.livekitRoom),
});
this.logger.info(
"Connected to LiveKit room",
this.transport.livekit_service_url,
);
} catch (error) {
this.logger.debug(`Failed to connect to LiveKit room: ${error}`);
this._state$.next({
@@ -189,9 +193,7 @@ export class Connection {
* This is derived from `participantsIncludingSubscribers$` and `remoteTransports$`.
* It filters the participants to only those that are associated with a membership that claims to publish on this connection.
*/
public readonly remoteParticipantsWithTracks$: Behavior<
PublishingParticipant[]
>;
public readonly remoteParticipants$: Behavior<PublishingParticipant[]>;
/**
* The media transport to connect to.
@@ -213,7 +215,7 @@ export class Connection {
public constructor(opts: ConnectionOpts, logger: Logger) {
this.logger = logger.getChild("[Connection]");
this.logger.info(
`[Connection] Creating new connection to ${opts.transport.livekit_service_url} ${opts.transport.livekit_alias}`,
`Creating new connection to ${opts.transport.livekit_service_url} ${opts.transport.livekit_alias}`,
);
const { transport, client, scope } = opts;
@@ -223,20 +225,21 @@ export class Connection {
// REMOTE participants with track!!!
// this.remoteParticipantsWithTracks$
this.remoteParticipantsWithTracks$ = scope.behavior(
this.remoteParticipants$ = scope.behavior(
// only tracks remote participants
connectedParticipantsObserver(this.livekitRoom, {
additionalRoomEvents: [
RoomEvent.TrackPublished,
RoomEvent.TrackUnpublished,
],
}).pipe(
map((participants) => {
return participants.filter(
(participant) => participant.getTrackPublications().length > 0,
);
}),
),
}),
// .pipe(
// map((participants) => {
// return participants.filter(
// (participant) => participant.getTrackPublications().length > 0,
// );
// }),
// )
[],
);

2
src/state/CallViewModel/remoteMembers/ConnectionFactory.ts
View File

@@ -13,7 +13,7 @@ import {
type BaseKeyProvider,
} from "livekit-client";
import { type Logger } from "matrix-js-sdk/lib/logger";
import E2EEWorker from "livekit-client/e2ee-worker?worker";
import E2EEWorker from "livekit-client/e2ee-worker?worker&inline";
import { type ObservableScope } from "../../ObservableScope.ts";
import { Connection } from "./Connection.ts";

52
src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts
View File

@@ -289,47 +289,47 @@ describe("connectionManagerData$ stream", () => {
a: expect.toSatisfy((e) => {
const data: ConnectionManagerData = e.value;
expect(data.getConnections().length).toBe(2);
expect(data.getParticipantForTransport(TRANSPORT_1).length).toBe(0);
expect(data.getParticipantForTransport(TRANSPORT_2).length).toBe(0);
expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(0);
expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(0);
return true;
}),
b: expect.toSatisfy((e) => {
const data: ConnectionManagerData = e.value;
expect(data.getConnections().length).toBe(2);
expect(data.getParticipantForTransport(TRANSPORT_1).length).toBe(1);
expect(data.getParticipantForTransport(TRANSPORT_2).length).toBe(0);
expect(data.getParticipantForTransport(TRANSPORT_1)[0].identity).toBe(
"user1A",
);
expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(1);
expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(0);
expect(
data.getParticipantsForTransport(TRANSPORT_1)[0].identity,
).toBe("user1A");
return true;
}),
c: expect.toSatisfy((e) => {
const data: ConnectionManagerData = e.value;
expect(data.getConnections().length).toBe(2);
expect(data.getParticipantForTransport(TRANSPORT_1).length).toBe(1);
expect(data.getParticipantForTransport(TRANSPORT_2).length).toBe(1);
expect(data.getParticipantForTransport(TRANSPORT_1)[0].identity).toBe(
"user1A",
);
expect(data.getParticipantForTransport(TRANSPORT_2)[0].identity).toBe(
"user2A",
);
expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(1);
expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(1);
expect(
data.getParticipantsForTransport(TRANSPORT_1)[0].identity,
).toBe("user1A");
expect(
data.getParticipantsForTransport(TRANSPORT_2)[0].identity,
).toBe("user2A");
return true;
}),
d: expect.toSatisfy((e) => {
const data: ConnectionManagerData = e.value;
expect(data.getConnections().length).toBe(2);
expect(data.getParticipantForTransport(TRANSPORT_1).length).toBe(2);
expect(data.getParticipantForTransport(TRANSPORT_2).length).toBe(1);
expect(data.getParticipantForTransport(TRANSPORT_1)[0].identity).toBe(
"user1A",
);
expect(data.getParticipantForTransport(TRANSPORT_1)[1].identity).toBe(
"user1B",
);
expect(data.getParticipantForTransport(TRANSPORT_2)[0].identity).toBe(
"user2A",
);
expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(2);
expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(1);
expect(
data.getParticipantsForTransport(TRANSPORT_1)[0].identity,
).toBe("user1A");
expect(
data.getParticipantsForTransport(TRANSPORT_1)[1].identity,
).toBe("user1B");
expect(
data.getParticipantsForTransport(TRANSPORT_2)[0].identity,
).toBe("user2A");
return true;
}),
});

39
src/state/CallViewModel/remoteMembers/ConnectionManager.ts
View File

@@ -24,7 +24,10 @@ import { type ConnectionFactory } from "./ConnectionFactory.ts";
export class ConnectionManagerData {
private readonly store: Map<
string,
[Connection, (LocalParticipant | RemoteParticipant)[]]
{
connection: Connection;
participants: (LocalParticipant | RemoteParticipant)[];
}
> = new Map();
public constructor() {}
@@ -36,9 +39,9 @@ export class ConnectionManagerData {
const key = this.getKey(connection.transport);
const existing = this.store.get(key);
if (!existing) {
this.store.set(key, [connection, participants]);
this.store.set(key, { connection, participants });
} else {
existing[1].push(...participants);
existing.participants.push(...participants);
}
}
@@ -47,25 +50,26 @@ export class ConnectionManagerData {
}
public getConnections(): Connection[] {
return Array.from(this.store.values()).map(([connection]) => connection);
return Array.from(this.store.values()).map(({ connection }) => connection);
}
public getConnectionForTransport(
transport: LivekitTransport,
): Connection | null {
return this.store.get(this.getKey(transport))?.[0] ?? null;
return this.store.get(this.getKey(transport))?.connection ?? null;
}
public getParticipantForTransport(
public getParticipantsForTransport(
transport: LivekitTransport,
): (LocalParticipant | RemoteParticipant)[] {
const key = transport.livekit_service_url + "|" + transport.livekit_alias;
const existing = this.store.get(key);
if (existing) {
return existing[1];
return existing.participants;
}
return [];
}
/**
* Get all connections where the given participant is publishing.
* In theory, there could be several connections where the same participant is publishing but with
@@ -76,8 +80,12 @@ export class ConnectionManagerData {
participantId: ParticipantId,
): Connection[] {
const connections: Connection[] = [];
for (const [connection, participants] of this.store.values()) {
if (participants.some((p) => p.identity === participantId)) {
for (const { connection, participants } of this.store.values()) {
if (
participants.some(
(participant) => participant?.identity === participantId,
)
) {
connections.push(connection);
}
}
@@ -183,23 +191,24 @@ export function createConnectionManager$({
const epoch = connections.epoch;
// Map the connections to list of {connection, participants}[]
const listOfConnectionsWithPublishingParticipants =
connections.value.map((connection) => {
return connection.remoteParticipantsWithTracks$.pipe(
const listOfConnectionsWithParticipants = connections.value.map(
(connection) => {
return connection.remoteParticipants$.pipe(
map((participants) => ({
connection,
participants,
})),
);
});
},
);
// probably not required
if (listOfConnectionsWithPublishingParticipants.length === 0) {
if (listOfConnectionsWithParticipants.length === 0) {
return of(new Epoch(new ConnectionManagerData(), epoch));
}
// combineLatest the several streams into a single stream with the ConnectionManagerData
return combineLatest(listOfConnectionsWithPublishingParticipants).pipe(
return combineLatest(listOfConnectionsWithParticipants).pipe(
map(
(lists) =>
new Epoch(

121
src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts
View File

@@ -91,7 +91,7 @@ test("should signal participant not yet connected to livekit", () => {
}),
);
const matrixLivekitMember$ = createMatrixLivekitMembers$({
const { matrixLivekitMembers$ } = createMatrixLivekitMembers$({
scope: testScope,
membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
connectionManager: {
@@ -99,21 +99,24 @@ test("should signal participant not yet connected to livekit", () => {
} as unknown as IConnectionManager,
});
expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe("a", {
a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
expect(data.length).toEqual(1);
expectObservable(data[0].membership$).toBe("a", {
a: bobMembership,
});
expectObservable(data[0].participant$).toBe("a", {
a: null,
});
expectObservable(data[0].connection$).toBe("a", {
a: null,
});
return true;
}),
});
expectObservable(matrixLivekitMembers$.pipe(map((e) => e.value))).toBe(
"a",
{
a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
expect(data.length).toEqual(1);
expectObservable(data[0].membership$).toBe("a", {
a: bobMembership,
});
expectObservable(data[0].participant$).toBe("a", {
a: null,
});
expectObservable(data[0].connection$).toBe("a", {
a: null,
});
return true;
}),
},
);
});
});
@@ -171,7 +174,7 @@ test("should signal participant on a connection that is publishing", () => {
}),
);
const matrixLivekitMember$ = createMatrixLivekitMembers$({
const { matrixLivekitMembers$ } = createMatrixLivekitMembers$({
scope: testScope,
membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
connectionManager: {
@@ -179,25 +182,28 @@ test("should signal participant on a connection that is publishing", () => {
} as unknown as IConnectionManager,
});
expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe("a", {
a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
expect(data.length).toEqual(1);
expectObservable(data[0].membership$).toBe("a", {
a: bobMembership,
});
expectObservable(data[0].participant$).toBe("a", {
a: expect.toSatisfy((participant) => {
expect(participant).toBeDefined();
expect(participant!.identity).toEqual(bobParticipantId);
return true;
}),
});
expectObservable(data[0].connection$).toBe("a", {
a: connection,
});
return true;
}),
});
expectObservable(matrixLivekitMembers$.pipe(map((e) => e.value))).toBe(
"a",
{
a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
expect(data.length).toEqual(1);
expectObservable(data[0].membership$).toBe("a", {
a: bobMembership,
});
expectObservable(data[0].participant$).toBe("a", {
a: expect.toSatisfy((participant) => {
expect(participant).toBeDefined();
expect(participant!.identity).toEqual(bobParticipantId);
return true;
}),
});
expectObservable(data[0].connection$).toBe("a", {
a: connection,
});
return true;
}),
},
);
});
});
@@ -222,7 +228,7 @@ test("should signal participant on a connection that is not publishing", () => {
}),
);
const matrixLivekitMember$ = createMatrixLivekitMembers$({
const { matrixLivekitMembers$ } = createMatrixLivekitMembers$({
scope: testScope,
membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
connectionManager: {
@@ -230,21 +236,24 @@ test("should signal participant on a connection that is not publishing", () => {
} as unknown as IConnectionManager,
});
expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe("a", {
a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
expect(data.length).toEqual(1);
expectObservable(data[0].membership$).toBe("a", {
a: bobMembership,
});
expectObservable(data[0].participant$).toBe("a", {
a: null,
});
expectObservable(data[0].connection$).toBe("a", {
a: connection,
});
return true;
}),
});
expectObservable(matrixLivekitMembers$.pipe(map((e) => e.value))).toBe(
"a",
{
a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
expect(data.length).toEqual(1);
expectObservable(data[0].membership$).toBe("a", {
a: bobMembership,
});
expectObservable(data[0].participant$).toBe("a", {
a: null,
});
expectObservable(data[0].connection$).toBe("a", {
a: connection,
});
return true;
}),
},
);
});
});
@@ -283,7 +292,7 @@ describe("Publication edge case", () => {
}),
);
const matrixLivekitMember$ = createMatrixLivekitMembers$({
const { matrixLivekitMembers$ } = createMatrixLivekitMembers$({
scope: testScope,
membershipsWithTransport$: testScope.behavior(
membershipsWithTransport$,
@@ -293,7 +302,7 @@ describe("Publication edge case", () => {
} as unknown as IConnectionManager,
});
expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe(
expectObservable(matrixLivekitMembers$.pipe(map((e) => e.value))).toBe(
"a",
{
a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
@@ -349,7 +358,7 @@ describe("Publication edge case", () => {
}),
);
const matrixLivekitMember$ = createMatrixLivekitMembers$({
const { matrixLivekitMembers$ } = createMatrixLivekitMembers$({
scope: testScope,
membershipsWithTransport$: testScope.behavior(
membershipsWithTransport$,
@@ -359,7 +368,7 @@ describe("Publication edge case", () => {
} as unknown as IConnectionManager,
});
expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe(
expectObservable(matrixLivekitMembers$.pipe(map((e) => e.value))).toBe(
"a",
{
a: expect.toSatisfy((data: MatrixLivekitMember[]) => {

11
src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts
View File

@@ -61,12 +61,12 @@ export function createMatrixLivekitMembers$({
scope,
membershipsWithTransport$,
connectionManager,
}: Props): Behavior<Epoch<MatrixLivekitMember[]>> {
}: Props): { matrixLivekitMembers$: Behavior<Epoch<MatrixLivekitMember[]>> } {
/**
* Stream of all the call members and their associated livekit data (if available).
*/
return scope.behavior(
const matrixLivekitMembers$ = scope.behavior(
combineLatest([
membershipsWithTransport$,
connectionManager.connectionManagerData$,
@@ -91,7 +91,7 @@ export function createMatrixLivekitMembers$({
const participantId = /*membership.membershipID*/ `${membership.userId}:${membership.deviceId}`;
const participants = transport
? managerData.getParticipantForTransport(transport)
? managerData.getParticipantsForTransport(transport)
: [];
const participant =
participants.find((p) => p.identity == participantId) ?? null;
@@ -121,6 +121,11 @@ export function createMatrixLivekitMembers$({
),
),
);
return {
matrixLivekitMembers$,
// TODO add only publishing participants... maybe. disucss at least
// scope.behavior(matrixLivekitMembers$.pipe(map((items) => items.value.map((i)=>{ i.}))))
};
}
// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)

4
src/state/CallViewModel/remoteMembers/integration.test.ts
View File

@@ -124,14 +124,14 @@ test("bob, carl, then bob joining no tracks yet", () => {
logger: logger,
});
const matrixLivekitItems$ = createMatrixLivekitMembers$({
const { matrixLivekitMembers$ } = createMatrixLivekitMembers$({
scope: testScope,
membershipsWithTransport$:
membershipsAndTransports.membershipsWithTransport$,
connectionManager,
});
expectObservable(matrixLivekitItems$).toBe(vMarble, {
expectObservable(matrixLivekitMembers$).toBe(vMarble, {
a: expect.toSatisfy((e: Epoch<MatrixLivekitMember[]>) => {
const items = e.value;
expect(items.length).toBe(1);