use createSomething$ mathods instead of classes
Rename several classes/behaviors to factory-style creators and adapt call wiring and tests accordingly: - Replace ConnectionManager class with createConnectionManager$ which returns transports$, connectionManagerData$, connections$ - Convert MatrixLivekitMerger to createMatrixLivekitMembers$ (matrixLivekitMerger$) - Rename sessionBehaviors$, localMembership$, localTransport$ to createSessionMembershipsAndTransports$, createLocalMembership$, createLocalTransport$ - Adjust participant types and hook up connectOptions$; expose join via localMembership.requestConnect - Update tests to use the new factory APIs
This commit is contained in:
@@ -110,15 +110,12 @@ import {
|
|||||||
} from "./layout-types.ts";
|
} from "./layout-types.ts";
|
||||||
import { type ElementCallError } from "../utils/errors.ts";
|
import { type ElementCallError } from "../utils/errors.ts";
|
||||||
import { type ObservableScope } from "./ObservableScope.ts";
|
import { type ObservableScope } from "./ObservableScope.ts";
|
||||||
import { ConnectionManager } from "./remoteMembers/ConnectionManager.ts";
|
import { createMatrixLivekitMembers$ } from "./remoteMembers/matrixLivekitMerger.ts";
|
||||||
import { MatrixLivekitMerger } from "./remoteMembers/matrixLivekitMerger.ts";
|
import { createLocalMembership$ } from "./localMember/LocalMembership.ts";
|
||||||
import {
|
import { createLocalTransport$ } from "./localMember/LocalTransport.ts";
|
||||||
localMembership$,
|
import { createSessionMembershipsAndTransports$ } from "./SessionBehaviors.ts";
|
||||||
type LocalMemberState,
|
|
||||||
} from "./localMember/LocalMembership.ts";
|
|
||||||
import { localTransport$ as computeLocalTransport$ } from "./localMember/LocalTransport.ts";
|
|
||||||
import { sessionBehaviors$ } from "./SessionBehaviors.ts";
|
|
||||||
import { ECConnectionFactory } from "./remoteMembers/ConnectionFactory.ts";
|
import { ECConnectionFactory } from "./remoteMembers/ConnectionFactory.ts";
|
||||||
|
import { createConnectionManager$ } from "./remoteMembers/ConnectionManager.ts";
|
||||||
|
|
||||||
//TODO
|
//TODO
|
||||||
// Larger rename
|
// Larger rename
|
||||||
@@ -192,13 +189,13 @@ export class CallViewModel {
|
|||||||
}
|
}
|
||||||
: undefined;
|
: undefined;
|
||||||
|
|
||||||
private sessionBehaviors = sessionBehaviors$({
|
private sessionBehaviors = createSessionMembershipsAndTransports$({
|
||||||
scope: this.scope,
|
scope: this.scope,
|
||||||
matrixRTCSession: this.matrixRTCSession,
|
matrixRTCSession: this.matrixRTCSession,
|
||||||
});
|
});
|
||||||
private memberships$ = this.sessionBehaviors.memberships$;
|
private memberships$ = this.sessionBehaviors.memberships$;
|
||||||
|
|
||||||
private localTransport$ = computeLocalTransport$({
|
private localTransport$ = createLocalTransport$({
|
||||||
scope: this.scope,
|
scope: this.scope,
|
||||||
memberships$: this.memberships$,
|
memberships$: this.memberships$,
|
||||||
client: this.matrixRoom.client,
|
client: this.matrixRoom.client,
|
||||||
@@ -229,25 +226,34 @@ export class CallViewModel {
|
|||||||
),
|
),
|
||||||
);
|
);
|
||||||
|
|
||||||
private connectionManager = new ConnectionManager(
|
private connectionManager = createConnectionManager$({
|
||||||
this.scope,
|
scope: this.scope,
|
||||||
this.connectionFactory,
|
connectionFactory: this.connectionFactory,
|
||||||
this.allTransports$,
|
inputTransports$: this.allTransports$,
|
||||||
);
|
});
|
||||||
|
|
||||||
// ------------------------------------------------------------------------
|
// ------------------------------------------------------------------------
|
||||||
|
|
||||||
private matrixLivekitMerger = new MatrixLivekitMerger(
|
private matrixLivekitMembers$ = createMatrixLivekitMembers$({
|
||||||
this.scope,
|
scope: this.scope,
|
||||||
this.sessionBehaviors.membershipsWithTransport$,
|
membershipsWithTransport$: this.sessionBehaviors.membershipsWithTransport$,
|
||||||
this.connectionManager,
|
connectionManager: this.connectionManager,
|
||||||
this.matrixRoom,
|
matrixRoom: this.matrixRoom,
|
||||||
this.userId,
|
userId: this.userId,
|
||||||
this.deviceId,
|
deviceId: this.deviceId,
|
||||||
);
|
});
|
||||||
private matrixLivekitMembers$ = this.matrixLivekitMerger.matrixLivekitMember$;
|
|
||||||
|
|
||||||
private localMembership = localMembership$({
|
private connectOptions$ = this.scope.behavior(
|
||||||
|
matrixRTCMode.value$.pipe(
|
||||||
|
map((mode) => ({
|
||||||
|
encryptMedia: this.e2eeLivekitOptions !== undefined,
|
||||||
|
// TODO. This might need to get called again on each cahnge of matrixRTCMode...
|
||||||
|
matrixRTCMode: mode,
|
||||||
|
})),
|
||||||
|
),
|
||||||
|
);
|
||||||
|
|
||||||
|
private localMembership = createLocalMembership$({
|
||||||
scope: this.scope,
|
scope: this.scope,
|
||||||
muteStates: this.muteStates,
|
muteStates: this.muteStates,
|
||||||
mediaDevices: this.mediaDevices,
|
mediaDevices: this.mediaDevices,
|
||||||
@@ -258,6 +264,7 @@ export class CallViewModel {
|
|||||||
e2eeLivekitOptions: this.e2eeLivekitOptions,
|
e2eeLivekitOptions: this.e2eeLivekitOptions,
|
||||||
trackProcessorState$: this.trackProcessorState$,
|
trackProcessorState$: this.trackProcessorState$,
|
||||||
widget,
|
widget,
|
||||||
|
options: this.connectOptions$,
|
||||||
});
|
});
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -269,13 +276,7 @@ export class CallViewModel {
|
|||||||
return this.localMembership.configError$;
|
return this.localMembership.configError$;
|
||||||
}
|
}
|
||||||
|
|
||||||
public join(): LocalMemberState {
|
public join = this.localMembership.requestConnect;
|
||||||
return this.localMembership.requestConnect({
|
|
||||||
encryptMedia: this.e2eeLivekitOptions !== undefined,
|
|
||||||
// TODO. This might need to get called again on each cahnge of matrixRTCMode...
|
|
||||||
matrixRTCMode: matrixRTCMode.getValue(),
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
// CODESMELL?
|
// CODESMELL?
|
||||||
// This is functionally the same Observable as leave$, except here it's
|
// This is functionally the same Observable as leave$, except here it's
|
||||||
|
|||||||
@@ -22,24 +22,16 @@ interface Props {
|
|||||||
matrixRTCSession: MatrixRTCSession;
|
matrixRTCSession: MatrixRTCSession;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
export const createSessionMembershipsAndTransports$ = ({
|
||||||
* Wraps behaviors that we extract from an matrixRTCSession.
|
scope,
|
||||||
*/
|
matrixRTCSession,
|
||||||
interface RxRtcSession {
|
}: Props): {
|
||||||
/**
|
|
||||||
* some prop
|
|
||||||
*/
|
|
||||||
memberships$: Behavior<CallMembership[]>;
|
memberships$: Behavior<CallMembership[]>;
|
||||||
membershipsWithTransport$: Behavior<
|
membershipsWithTransport$: Behavior<
|
||||||
{ membership: CallMembership; transport?: LivekitTransport }[]
|
{ membership: CallMembership; transport?: LivekitTransport }[]
|
||||||
>;
|
>;
|
||||||
transports$: Behavior<LivekitTransport[]>;
|
transports$: Behavior<LivekitTransport[]>;
|
||||||
}
|
} => {
|
||||||
|
|
||||||
export const sessionBehaviors$ = ({
|
|
||||||
scope,
|
|
||||||
matrixRTCSession,
|
|
||||||
}: Props): RxRtcSession => {
|
|
||||||
const memberships$ = scope.behavior(
|
const memberships$ = scope.behavior(
|
||||||
fromEvent(
|
fromEvent(
|
||||||
matrixRTCSession,
|
matrixRTCSession,
|
||||||
|
|||||||
@@ -28,7 +28,7 @@ import {
|
|||||||
import { logger } from "matrix-js-sdk/lib/logger";
|
import { logger } from "matrix-js-sdk/lib/logger";
|
||||||
|
|
||||||
import { type Behavior } from "../Behavior";
|
import { type Behavior } from "../Behavior";
|
||||||
import { type ConnectionManager } from "../remoteMembers/ConnectionManager";
|
import { type createConnectionManager$ } from "../remoteMembers/ConnectionManager";
|
||||||
import { ObservableScope } from "../ObservableScope";
|
import { ObservableScope } from "../ObservableScope";
|
||||||
import { Publisher } from "./Publisher";
|
import { Publisher } from "./Publisher";
|
||||||
import { type MuteStates } from "../MuteStates";
|
import { type MuteStates } from "../MuteStates";
|
||||||
@@ -90,7 +90,7 @@ interface Props {
|
|||||||
scope: ObservableScope;
|
scope: ObservableScope;
|
||||||
mediaDevices: MediaDevices;
|
mediaDevices: MediaDevices;
|
||||||
muteStates: MuteStates;
|
muteStates: MuteStates;
|
||||||
connectionManager: ConnectionManager;
|
connectionManager: ReturnType<typeof createConnectionManager$>;
|
||||||
matrixRTCSession: MatrixRTCSession;
|
matrixRTCSession: MatrixRTCSession;
|
||||||
matrixRoom: MatrixRoom;
|
matrixRoom: MatrixRoom;
|
||||||
localTransport$: Behavior<LivekitTransport | undefined>;
|
localTransport$: Behavior<LivekitTransport | undefined>;
|
||||||
@@ -111,7 +111,7 @@ interface Props {
|
|||||||
* - transport$: the transport object the ownMembership$ ended up using.
|
* - transport$: the transport object the ownMembership$ ended up using.
|
||||||
*
|
*
|
||||||
*/
|
*/
|
||||||
export const localMembership$ = ({
|
export const createLocalMembership$ = ({
|
||||||
scope,
|
scope,
|
||||||
options,
|
options,
|
||||||
muteStates,
|
muteStates,
|
||||||
@@ -151,13 +151,14 @@ export const localMembership$ = ({
|
|||||||
const tracks$ = new BehaviorSubject<LocalTrack[]>([]);
|
const tracks$ = new BehaviorSubject<LocalTrack[]>([]);
|
||||||
|
|
||||||
const connection$ = scope.behavior(
|
const connection$ = scope.behavior(
|
||||||
combineLatest([connectionManager.connections$, localTransport$]).pipe(
|
combineLatest(
|
||||||
map(([connections, transport]) => {
|
[connectionManager.connections$, localTransport$],
|
||||||
|
(connections, transport) => {
|
||||||
if (transport === undefined) return undefined;
|
if (transport === undefined) return undefined;
|
||||||
return connections.find((connection) =>
|
return connections.find((connection) =>
|
||||||
areLivekitTransportsEqual(connection.transport, transport),
|
areLivekitTransportsEqual(connection.transport, transport),
|
||||||
);
|
);
|
||||||
}),
|
},
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -50,7 +50,7 @@ interface Props {
|
|||||||
* @prop useOldestMember Whether to use the same transport as the oldest member.
|
* @prop useOldestMember Whether to use the same transport as the oldest member.
|
||||||
* This will only update once the first oldest member appears. Will not recompute if the oldest member leaves.
|
* This will only update once the first oldest member appears. Will not recompute if the oldest member leaves.
|
||||||
*/
|
*/
|
||||||
export const localTransport$ = ({
|
export const createLocalTransport$ = ({
|
||||||
scope,
|
scope,
|
||||||
memberships$,
|
memberships$,
|
||||||
client,
|
client,
|
||||||
|
|||||||
@@ -13,7 +13,8 @@ import {
|
|||||||
ConnectionError,
|
ConnectionError,
|
||||||
type ConnectionState as LivekitConenctionState,
|
type ConnectionState as LivekitConenctionState,
|
||||||
type Room as LivekitRoom,
|
type Room as LivekitRoom,
|
||||||
type Participant,
|
type LocalParticipant,
|
||||||
|
type RemoteParticipant,
|
||||||
RoomEvent,
|
RoomEvent,
|
||||||
} from "livekit-client";
|
} from "livekit-client";
|
||||||
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
|
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
|
||||||
@@ -32,7 +33,7 @@ import {
|
|||||||
SFURoomCreationRestrictedError,
|
SFURoomCreationRestrictedError,
|
||||||
} from "../../utils/errors.ts";
|
} from "../../utils/errors.ts";
|
||||||
|
|
||||||
export type PublishingParticipant = Participant;
|
export type PublishingParticipant = LocalParticipant | RemoteParticipant;
|
||||||
|
|
||||||
export interface ConnectionOpts {
|
export interface ConnectionOpts {
|
||||||
/** The media transport to connect to. */
|
/** The media transport to connect to. */
|
||||||
|
|||||||
@@ -6,13 +6,12 @@ Please see LICENSE in the repository root for full details.
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
|
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
|
||||||
import { logger } from "matrix-js-sdk/lib/logger";
|
|
||||||
import { BehaviorSubject } from "rxjs";
|
import { BehaviorSubject } from "rxjs";
|
||||||
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
|
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
|
||||||
import { type Participant as LivekitParticipant } from "livekit-client";
|
import { type Participant as LivekitParticipant } from "livekit-client";
|
||||||
|
|
||||||
import { ObservableScope } from "../ObservableScope.ts";
|
import { ObservableScope } from "../ObservableScope.ts";
|
||||||
import { ConnectionManager } from "./ConnectionManager.ts";
|
import { createConnectionManager$ } from "./ConnectionManager.ts";
|
||||||
import { type ConnectionFactory } from "./ConnectionFactory.ts";
|
import { type ConnectionFactory } from "./ConnectionFactory.ts";
|
||||||
import { type Connection } from "./Connection.ts";
|
import { type Connection } from "./Connection.ts";
|
||||||
import { areLivekitTransportsEqual } from "./matrixLivekitMerger.ts";
|
import { areLivekitTransportsEqual } from "./matrixLivekitMerger.ts";
|
||||||
@@ -37,15 +36,15 @@ const TRANSPORT_2: LivekitTransport = {
|
|||||||
// livekit_service_url: "https://lk-other.sample.com",
|
// livekit_service_url: "https://lk-other.sample.com",
|
||||||
// livekit_alias: "!alias:sample.com",
|
// livekit_alias: "!alias:sample.com",
|
||||||
// };
|
// };
|
||||||
|
|
||||||
let testScope: ObservableScope;
|
|
||||||
let fakeConnectionFactory: ConnectionFactory;
|
let fakeConnectionFactory: ConnectionFactory;
|
||||||
|
let testScope: ObservableScope;
|
||||||
let testTransportStream$: BehaviorSubject<LivekitTransport[]>;
|
let testTransportStream$: BehaviorSubject<LivekitTransport[]>;
|
||||||
|
let connectionManagerInputs: {
|
||||||
// The connection manager under test
|
scope: ObservableScope;
|
||||||
let manager: ConnectionManager;
|
connectionFactory: ConnectionFactory;
|
||||||
|
inputTransports$: BehaviorSubject<LivekitTransport[]>;
|
||||||
|
};
|
||||||
|
let manager: ReturnType<typeof createConnectionManager$>;
|
||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
testScope = new ObservableScope();
|
testScope = new ObservableScope();
|
||||||
|
|
||||||
@@ -68,9 +67,12 @@ beforeEach(() => {
|
|||||||
);
|
);
|
||||||
|
|
||||||
testTransportStream$ = new BehaviorSubject<LivekitTransport[]>([]);
|
testTransportStream$ = new BehaviorSubject<LivekitTransport[]>([]);
|
||||||
|
connectionManagerInputs = {
|
||||||
manager = new ConnectionManager(testScope, fakeConnectionFactory, logger);
|
scope: testScope,
|
||||||
manager.registerTransports(testTransportStream$);
|
connectionFactory: fakeConnectionFactory,
|
||||||
|
inputTransports$: testTransportStream$,
|
||||||
|
};
|
||||||
|
manager = createConnectionManager$(connectionManagerInputs);
|
||||||
});
|
});
|
||||||
|
|
||||||
afterEach(() => {
|
afterEach(() => {
|
||||||
@@ -84,7 +86,7 @@ describe("connections$ stream", () => {
|
|||||||
if (connections.length > 0) managedConnections.resolve(connections);
|
if (connections.length > 0) managedConnections.resolve(connections);
|
||||||
});
|
});
|
||||||
|
|
||||||
testTransportStream$.next([TRANSPORT_1, TRANSPORT_2]);
|
connectionManagerInputs.inputTransports$.next([TRANSPORT_1, TRANSPORT_2]);
|
||||||
|
|
||||||
const connections = await managedConnections.promise;
|
const connections = await managedConnections.promise;
|
||||||
|
|
||||||
@@ -211,11 +213,13 @@ describe("connectionManagerData$ stream", () => {
|
|||||||
|
|
||||||
test("Should report connections with the publishing participants", () => {
|
test("Should report connections with the publishing participants", () => {
|
||||||
withTestScheduler(({ expectObservable, schedule, behavior }) => {
|
withTestScheduler(({ expectObservable, schedule, behavior }) => {
|
||||||
manager.registerTransports(
|
manager = createConnectionManager$({
|
||||||
behavior("a", {
|
...connectionManagerInputs,
|
||||||
|
inputTransports$: behavior("a", {
|
||||||
a: [TRANSPORT_1, TRANSPORT_2],
|
a: [TRANSPORT_1, TRANSPORT_2],
|
||||||
}),
|
}),
|
||||||
);
|
});
|
||||||
|
|
||||||
const conn1Participants$ = fakePublishingParticipantsStreams.get(
|
const conn1Participants$ = fakePublishingParticipantsStreams.get(
|
||||||
keyForTransport(TRANSPORT_1),
|
keyForTransport(TRANSPORT_1),
|
||||||
)!;
|
)!;
|
||||||
|
|||||||
@@ -14,8 +14,8 @@ import {
|
|||||||
type ParticipantId,
|
type ParticipantId,
|
||||||
} from "matrix-js-sdk/lib/matrixrtc";
|
} from "matrix-js-sdk/lib/matrixrtc";
|
||||||
import { BehaviorSubject, combineLatest, map, switchMap } from "rxjs";
|
import { BehaviorSubject, combineLatest, map, switchMap } from "rxjs";
|
||||||
import { logger, type Logger } from "matrix-js-sdk/lib/logger";
|
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
|
||||||
import { type Participant as LivekitParticipant } from "livekit-client";
|
import { type LocalParticipant, type RemoteParticipant } from "livekit-client";
|
||||||
|
|
||||||
import { type Behavior } from "../Behavior";
|
import { type Behavior } from "../Behavior";
|
||||||
import { type Connection } from "./Connection";
|
import { type Connection } from "./Connection";
|
||||||
@@ -25,12 +25,17 @@ import { areLivekitTransportsEqual } from "./matrixLivekitMerger";
|
|||||||
import { type ConnectionFactory } from "./ConnectionFactory.ts";
|
import { type ConnectionFactory } from "./ConnectionFactory.ts";
|
||||||
|
|
||||||
export class ConnectionManagerData {
|
export class ConnectionManagerData {
|
||||||
private readonly store: Map<string, [Connection, LivekitParticipant[]]> =
|
private readonly store: Map<
|
||||||
new Map();
|
string,
|
||||||
|
[Connection, (LocalParticipant | RemoteParticipant)[]]
|
||||||
|
> = new Map();
|
||||||
|
|
||||||
public constructor() {}
|
public constructor() {}
|
||||||
|
|
||||||
public add(connection: Connection, participants: LivekitParticipant[]): void {
|
public add(
|
||||||
|
connection: Connection,
|
||||||
|
participants: (LocalParticipant | RemoteParticipant)[],
|
||||||
|
): void {
|
||||||
const key = this.getKey(connection.transport);
|
const key = this.getKey(connection.transport);
|
||||||
const existing = this.store.get(key);
|
const existing = this.store.get(key);
|
||||||
if (!existing) {
|
if (!existing) {
|
||||||
@@ -56,7 +61,7 @@ export class ConnectionManagerData {
|
|||||||
|
|
||||||
public getParticipantForTransport(
|
public getParticipantForTransport(
|
||||||
transport: LivekitTransport,
|
transport: LivekitTransport,
|
||||||
): LivekitParticipant[] {
|
): (LocalParticipant | RemoteParticipant)[] {
|
||||||
const key = transport.livekit_service_url + "|" + transport.livekit_alias;
|
const key = transport.livekit_service_url + "|" + transport.livekit_alias;
|
||||||
const existing = this.store.get(key);
|
const existing = this.store.get(key);
|
||||||
if (existing) {
|
if (existing) {
|
||||||
@@ -82,35 +87,41 @@ export class ConnectionManagerData {
|
|||||||
return connections;
|
return connections;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
interface Props {
|
||||||
|
scope: ObservableScope;
|
||||||
|
connectionFactory: ConnectionFactory;
|
||||||
|
inputTransports$: Behavior<LivekitTransport[]>;
|
||||||
|
}
|
||||||
// TODO - write test for scopes (do we really need to bind scope)
|
// TODO - write test for scopes (do we really need to bind scope)
|
||||||
export class ConnectionManager {
|
|
||||||
private readonly logger: Logger;
|
|
||||||
|
|
||||||
private running$ = new BehaviorSubject(true);
|
/**
|
||||||
/**
|
* Crete a `ConnectionManager`
|
||||||
* Crete a `ConnectionManager`
|
* @param scope the observable scope used by this object.
|
||||||
* @param scope the observable scope used by this object.
|
* @param connectionFactory used to create new connections.
|
||||||
* @param connectionFactory used to create new connections.
|
* @param _transportsSubscriptions$ A list of Behaviors each containing a LIST of LivekitTransport.
|
||||||
* @param _transportsSubscriptions$ A list of Behaviors each containing a LIST of LivekitTransport.
|
* Each of these behaviors can be interpreted as subscribed list of transports.
|
||||||
* Each of these behaviors can be interpreted as subscribed list of transports.
|
*
|
||||||
*
|
* Using `registerTransports` independent external modules can control what connections
|
||||||
* Using `registerTransports` independent external modules can control what connections
|
* are created by the ConnectionManager.
|
||||||
* are created by the ConnectionManager.
|
*
|
||||||
*
|
* The connection manager will remove all duplicate transports in each subscibed list.
|
||||||
* The connection manager will remove all duplicate transports in each subscibed list.
|
*
|
||||||
*
|
* See `unregisterAllTransports` and `unregisterTransport` for details on how to unsubscribe.
|
||||||
* See `unregisterAllTransports` and `unregisterTransport` for details on how to unsubscribe.
|
*/
|
||||||
*/
|
export function createConnectionManager$({
|
||||||
public constructor(
|
scope,
|
||||||
private readonly scope: ObservableScope,
|
connectionFactory,
|
||||||
private readonly connectionFactory: ConnectionFactory,
|
inputTransports$,
|
||||||
private readonly inputTransports$: Behavior<LivekitTransport[]>,
|
}: Props): {
|
||||||
) {
|
transports$: Behavior<LivekitTransport[]>;
|
||||||
// TODO logger: only construct one logger from the client and make it compatible via a EC specific sing
|
connectionManagerData$: Behavior<ConnectionManagerData>;
|
||||||
this.logger = logger.getChild("ConnectionManager");
|
connections$: Behavior<Connection[]>;
|
||||||
scope.onEnd(() => this.running$.next(false));
|
} {
|
||||||
}
|
const logger = rootLogger.getChild("ConnectionManager");
|
||||||
|
|
||||||
|
const running$ = new BehaviorSubject(true);
|
||||||
|
scope.onEnd(() => running$.next(false));
|
||||||
|
// TODO logger: only construct one logger from the client and make it compatible via a EC specific sing
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* All transports currently managed by the ConnectionManager.
|
* All transports currently managed by the ConnectionManager.
|
||||||
@@ -120,8 +131,8 @@ export class ConnectionManager {
|
|||||||
* It is build based on the list of subscribed transports (`transportsSubscriptions$`).
|
* It is build based on the list of subscribed transports (`transportsSubscriptions$`).
|
||||||
* externally this is modified via `registerTransports()`.
|
* externally this is modified via `registerTransports()`.
|
||||||
*/
|
*/
|
||||||
private readonly transports$ = this.scope.behavior(
|
const transports$ = scope.behavior(
|
||||||
combineLatest([this.running$, this.inputTransports$]).pipe(
|
combineLatest([running$, inputTransports$]).pipe(
|
||||||
map(([running, transports]) => (running ? transports : [])),
|
map(([running, transports]) => (running ? transports : [])),
|
||||||
map(removeDuplicateTransports),
|
map(removeDuplicateTransports),
|
||||||
),
|
),
|
||||||
@@ -130,19 +141,19 @@ export class ConnectionManager {
|
|||||||
/**
|
/**
|
||||||
* Connections for each transport in use by one or more session members.
|
* Connections for each transport in use by one or more session members.
|
||||||
*/
|
*/
|
||||||
public readonly connections$ = this.scope.behavior(
|
const connections$ = scope.behavior(
|
||||||
generateKeyed$<LivekitTransport[], Connection, Connection[]>(
|
generateKeyed$<LivekitTransport[], Connection, Connection[]>(
|
||||||
this.transports$,
|
transports$,
|
||||||
(transports, createOrGet) => {
|
(transports, createOrGet) => {
|
||||||
const createConnection =
|
const createConnection =
|
||||||
(
|
(
|
||||||
transport: LivekitTransport,
|
transport: LivekitTransport,
|
||||||
): ((scope: ObservableScope) => Connection) =>
|
): ((scope: ObservableScope) => Connection) =>
|
||||||
(scope) => {
|
(scope) => {
|
||||||
const connection = this.connectionFactory.createConnection(
|
const connection = connectionFactory.createConnection(
|
||||||
transport,
|
transport,
|
||||||
scope,
|
scope,
|
||||||
this.logger,
|
logger,
|
||||||
);
|
);
|
||||||
// Start the connection immediately
|
// Start the connection immediately
|
||||||
// Use connection state to track connection progress
|
// Use connection state to track connection progress
|
||||||
@@ -160,9 +171,9 @@ export class ConnectionManager {
|
|||||||
),
|
),
|
||||||
);
|
);
|
||||||
|
|
||||||
public connectionManagerData$: Behavior<ConnectionManagerData> =
|
const connectionManagerData$: Behavior<ConnectionManagerData> =
|
||||||
this.scope.behavior(
|
scope.behavior(
|
||||||
this.connections$.pipe(
|
connections$.pipe(
|
||||||
switchMap((connections) => {
|
switchMap((connections) => {
|
||||||
// Map the connections to list of {connection, participants}[]
|
// Map the connections to list of {connection, participants}[]
|
||||||
const listOfConnectionsWithPublishingParticipants = connections.map(
|
const listOfConnectionsWithPublishingParticipants = connections.map(
|
||||||
@@ -191,6 +202,7 @@ export class ConnectionManager {
|
|||||||
// start empty
|
// start empty
|
||||||
new ConnectionManagerData(),
|
new ConnectionManagerData(),
|
||||||
);
|
);
|
||||||
|
return { transports$, connectionManagerData$, connections$ };
|
||||||
}
|
}
|
||||||
|
|
||||||
function removeDuplicateTransports(
|
function removeDuplicateTransports(
|
||||||
|
|||||||
@@ -24,7 +24,7 @@ import { getParticipantId } from "matrix-js-sdk/lib/matrixrtc/utils";
|
|||||||
|
|
||||||
import {
|
import {
|
||||||
type MatrixLivekitMember,
|
type MatrixLivekitMember,
|
||||||
MatrixLivekitMerger,
|
matrixLivekitMerger$,
|
||||||
} from "./matrixLivekitMerger";
|
} from "./matrixLivekitMerger";
|
||||||
import { ObservableScope } from "../ObservableScope";
|
import { ObservableScope } from "../ObservableScope";
|
||||||
import {
|
import {
|
||||||
@@ -44,7 +44,7 @@ const userId = "@local:example.com";
|
|||||||
const deviceId = "DEVICE000";
|
const deviceId = "DEVICE000";
|
||||||
|
|
||||||
// The merger beeing tested
|
// The merger beeing tested
|
||||||
let matrixLivekitMerger: MatrixLivekitMerger;
|
let matrixLivekitMerger: matrixLivekitMerger$;
|
||||||
|
|
||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
testScope = new ObservableScope();
|
testScope = new ObservableScope();
|
||||||
@@ -62,7 +62,7 @@ beforeEach(() => {
|
|||||||
removeEventListener: vi.fn(),
|
removeEventListener: vi.fn(),
|
||||||
} as unknown as MatrixRoom);
|
} as unknown as MatrixRoom);
|
||||||
|
|
||||||
matrixLivekitMerger = new MatrixLivekitMerger(
|
matrixLivekitMerger = new matrixLivekitMerger$(
|
||||||
testScope,
|
testScope,
|
||||||
fakeMemberships$,
|
fakeMemberships$,
|
||||||
mockConnectionManager,
|
mockConnectionManager,
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ import { ECConnectionFactory } from "./ConnectionFactory.ts";
|
|||||||
import { type OpenIDClientParts } from "../../livekit/openIDSFU.ts";
|
import { type OpenIDClientParts } from "../../livekit/openIDSFU.ts";
|
||||||
import { mockMediaDevices, withTestScheduler } from "../../utils/test";
|
import { mockMediaDevices, withTestScheduler } from "../../utils/test";
|
||||||
import { type ProcessorState } from "../../livekit/TrackProcessorContext.tsx";
|
import { type ProcessorState } from "../../livekit/TrackProcessorContext.tsx";
|
||||||
import { MatrixLivekitMerger } from "./matrixLivekitMerger.ts";
|
import { matrixLivekitMerger$ } from "./matrixLivekitMerger.ts";
|
||||||
import type { CallMembership, Transport } from "matrix-js-sdk/lib/matrixrtc";
|
import type { CallMembership, Transport } from "matrix-js-sdk/lib/matrixrtc";
|
||||||
import { TRANSPORT_1 } from "./ConnectionManager.test.ts";
|
import { TRANSPORT_1 } from "./ConnectionManager.test.ts";
|
||||||
|
|
||||||
@@ -39,9 +39,9 @@ let connectionManager: ConnectionManager;
|
|||||||
|
|
||||||
function createLkMerger(
|
function createLkMerger(
|
||||||
memberships$: Observable<CallMembership[]>,
|
memberships$: Observable<CallMembership[]>,
|
||||||
): MatrixLivekitMerger {
|
): matrixLivekitMerger$ {
|
||||||
const mockRoomEmitter = new EventEmitter();
|
const mockRoomEmitter = new EventEmitter();
|
||||||
return new MatrixLivekitMerger(
|
return new matrixLivekitMerger$(
|
||||||
testScope,
|
testScope,
|
||||||
memberships$,
|
memberships$,
|
||||||
connectionManager,
|
connectionManager,
|
||||||
|
|||||||
@@ -16,12 +16,12 @@ import {
|
|||||||
import { combineLatest, map, startWith, type Observable } from "rxjs";
|
import { combineLatest, map, startWith, type Observable } from "rxjs";
|
||||||
// eslint-disable-next-line rxjs/no-internal
|
// eslint-disable-next-line rxjs/no-internal
|
||||||
import { type NodeStyleEventEmitter } from "rxjs/internal/observable/fromEvent";
|
import { type NodeStyleEventEmitter } from "rxjs/internal/observable/fromEvent";
|
||||||
|
import { type Room as MatrixRoom, type RoomMember } from "matrix-js-sdk";
|
||||||
|
|
||||||
import type { Room as MatrixRoom, RoomMember } from "matrix-js-sdk";
|
|
||||||
// import type { Logger } from "matrix-js-sdk/lib/logger";
|
// import type { Logger } from "matrix-js-sdk/lib/logger";
|
||||||
import { type Behavior } from "../Behavior";
|
import { type Behavior } from "../Behavior";
|
||||||
import { type ObservableScope } from "../ObservableScope";
|
import { type ObservableScope } from "../ObservableScope";
|
||||||
import { type ConnectionManager } from "./ConnectionManager";
|
import { type createConnectionManager$ } from "./ConnectionManager";
|
||||||
import { getRoomMemberFromRtcMember, memberDisplaynames$ } from "./displayname";
|
import { getRoomMemberFromRtcMember, memberDisplaynames$ } from "./displayname";
|
||||||
import { type Connection } from "./Connection";
|
import { type Connection } from "./Connection";
|
||||||
|
|
||||||
@@ -45,11 +45,25 @@ export interface MatrixLivekitMember {
|
|||||||
participantId: string;
|
participantId: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
interface Props {
|
||||||
|
scope: ObservableScope;
|
||||||
|
membershipsWithTransport$: Behavior<
|
||||||
|
{ membership: CallMembership; transport?: LivekitTransport }[]
|
||||||
|
>;
|
||||||
|
connectionManager: ReturnType<typeof createConnectionManager$>;
|
||||||
|
// TODO this is too much information for that class,
|
||||||
|
// apparently needed to get a room member to later get the Avatar
|
||||||
|
// => Extract an AvatarService instead?
|
||||||
|
// Better with just `getMember`
|
||||||
|
matrixRoom: Pick<MatrixRoom, "getMember"> & NodeStyleEventEmitter;
|
||||||
|
userId: string;
|
||||||
|
deviceId: string;
|
||||||
|
}
|
||||||
// Alternative structure idea:
|
// Alternative structure idea:
|
||||||
// const livekitMatrixMember$ = (callMemberships$,connectionManager,scope): Observable<MatrixLivekitMember[]> => {
|
// const livekitMatrixMember$ = (callMemberships$,connectionManager,scope): Observable<MatrixLivekitMember[]> => {
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Combines MatrixRtc and Livekit worlds.
|
* Combines MatrixRTC and Livekit worlds.
|
||||||
*
|
*
|
||||||
* It has a small public interface:
|
* It has a small public interface:
|
||||||
* - in (via constructor):
|
* - in (via constructor):
|
||||||
@@ -58,54 +72,30 @@ export interface MatrixLivekitMember {
|
|||||||
* - out (via public Observable):
|
* - out (via public Observable):
|
||||||
* - `remoteMatrixLivekitMember` an observable of MatrixLivekitMember[] to track the remote members and associated livekit data.
|
* - `remoteMatrixLivekitMember` an observable of MatrixLivekitMember[] to track the remote members and associated livekit data.
|
||||||
*/
|
*/
|
||||||
export class MatrixLivekitMerger {
|
export function createMatrixLivekitMembers$({
|
||||||
|
scope,
|
||||||
|
membershipsWithTransport$,
|
||||||
|
connectionManager,
|
||||||
|
matrixRoom,
|
||||||
|
userId,
|
||||||
|
deviceId,
|
||||||
|
}: Props): Behavior<MatrixLivekitMember[]> {
|
||||||
/**
|
/**
|
||||||
* Stream of all the call members and their associated livekit data (if available).
|
* Stream of all the call members and their associated livekit data (if available).
|
||||||
*/
|
*/
|
||||||
public matrixLivekitMember$: Behavior<MatrixLivekitMember[]>;
|
|
||||||
|
|
||||||
// private readonly logger: Logger;
|
function createMatrixLivekitMember$(): Observable<MatrixLivekitMember[]> {
|
||||||
|
|
||||||
public constructor(
|
|
||||||
private scope: ObservableScope,
|
|
||||||
private membershipsWithTransport$: Behavior<
|
|
||||||
{ membership: CallMembership; transport?: LivekitTransport }[]
|
|
||||||
>,
|
|
||||||
private connectionManager: ConnectionManager,
|
|
||||||
// TODO this is too much information for that class,
|
|
||||||
// apparently needed to get a room member to later get the Avatar
|
|
||||||
// => Extract an AvatarService instead?
|
|
||||||
// Better with just `getMember`
|
|
||||||
private matrixRoom: Pick<MatrixRoom, "getMember"> & NodeStyleEventEmitter,
|
|
||||||
private userId: string,
|
|
||||||
private deviceId: string,
|
|
||||||
// parentLogger: Logger,
|
|
||||||
) {
|
|
||||||
// this.logger = parentLogger.getChild("MatrixLivekitMerger");
|
|
||||||
|
|
||||||
this.matrixLivekitMember$ = this.scope.behavior(
|
|
||||||
this.start$().pipe(startWith([])),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
// =======================================
|
|
||||||
/// PRIVATES
|
|
||||||
// =======================================
|
|
||||||
private start$(): Observable<MatrixLivekitMember[]> {
|
|
||||||
const displaynameMap$ = memberDisplaynames$(
|
const displaynameMap$ = memberDisplaynames$(
|
||||||
this.scope,
|
scope,
|
||||||
this.matrixRoom,
|
matrixRoom,
|
||||||
this.membershipsWithTransport$.pipe(
|
membershipsWithTransport$.pipe(map((v) => v.map((v) => v.membership))),
|
||||||
map((v) => v.map((v) => v.membership)),
|
userId,
|
||||||
),
|
deviceId,
|
||||||
this.userId,
|
|
||||||
this.deviceId,
|
|
||||||
);
|
);
|
||||||
const membershipsWithTransport$ = this.membershipsWithTransport$;
|
|
||||||
|
|
||||||
return combineLatest([
|
return combineLatest([
|
||||||
membershipsWithTransport$,
|
membershipsWithTransport$,
|
||||||
this.connectionManager.connectionManagerData$,
|
connectionManager.connectionManagerData$,
|
||||||
]).pipe(
|
]).pipe(
|
||||||
map(([memberships, managerData]) => {
|
map(([memberships, managerData]) => {
|
||||||
const items: MatrixLivekitMember[] = memberships.map(
|
const items: MatrixLivekitMember[] = memberships.map(
|
||||||
@@ -121,12 +111,12 @@ export class MatrixLivekitMerger {
|
|||||||
);
|
);
|
||||||
const member = getRoomMemberFromRtcMember(
|
const member = getRoomMemberFromRtcMember(
|
||||||
membership,
|
membership,
|
||||||
this.matrixRoom,
|
matrixRoom,
|
||||||
)?.member;
|
)?.member;
|
||||||
const connection = transport
|
const connection = transport
|
||||||
? managerData.getConnectionForTransport(transport)
|
? managerData.getConnectionForTransport(transport)
|
||||||
: undefined;
|
: undefined;
|
||||||
const displayName$ = this.scope.behavior(
|
const displayName$ = scope.behavior(
|
||||||
displaynameMap$.pipe(
|
displaynameMap$.pipe(
|
||||||
map(
|
map(
|
||||||
(displayNameMap) =>
|
(displayNameMap) =>
|
||||||
@@ -139,7 +129,8 @@ export class MatrixLivekitMerger {
|
|||||||
membership,
|
membership,
|
||||||
connection,
|
connection,
|
||||||
// This makes sense to add the the js-sdk callMembership (we only need the avatar so probably the call memberhsip just should aquire the avatar)
|
// This makes sense to add the the js-sdk callMembership (we only need the avatar so probably the call memberhsip just should aquire the avatar)
|
||||||
member,
|
// TODO Ugh this is hidign that it might be undefined!! best we remove the member entirely.
|
||||||
|
member: member as RoomMember,
|
||||||
displayName$,
|
displayName$,
|
||||||
mxcAvatarUrl: member?.getMxcAvatarUrl(),
|
mxcAvatarUrl: member?.getMxcAvatarUrl(),
|
||||||
participantId,
|
participantId,
|
||||||
@@ -150,6 +141,8 @@ export class MatrixLivekitMerger {
|
|||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
return scope.behavior(createMatrixLivekitMember$().pipe(startWith([])));
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
|
// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
|
||||||
|
|||||||
Reference in New Issue
Block a user