Skip to content

Commit

Permalink
MatrixRTC: refactor MatrixRTCSession MemberManager API (#4610)
Browse files Browse the repository at this point in the history 
* update join and leave internal api.

* rename onMembershipUpdate and triggerCallMembershipEventUpdate to onMembershipsUpdate
This makes it more clear that we do not talk about our own membership but all memberships in the session

* cleanup MembershipManager
 - add comments and interface how to test this class.
 - sort methods by public/private
 - make triggerCallMembershipEventUpdate private

* docstrings for getFocusInUse and getActiveFocus

* simplify tests and make them only use MembershipManagerInterface methods.
This allows to exchange the membershipManager with a different implementation.

* convert interface to abstract class.

* review (implement interface, make interface internal, dont change public api.)

* Make the interface an actual interface.
The actual constructor of the class now contains the `Pick` to define what it needs from the client.

* move update condition into MembershipManager

* renaming public api

* Update src/matrixrtc/MatrixRTCSession.ts

Co-authored-by: Hugh Nimmo-Smith <[email protected]>

* Update src/matrixrtc/MatrixRTCSession.ts

Co-authored-by: Hugh Nimmo-Smith <[email protected]>

---------

Co-authored-by: Hugh Nimmo-Smith <[email protected]>
  • Loading branch information
@toger5 @hughns
toger5 and hughns authored Jan 13, 2025
1 parent bed4e95 commit ffb228b
Show file tree
Hide file tree
Showing 4 changed files with 174 additions and 100 deletions.
56 changes: 25 additions & 31 deletions spec/unit/matrixrtc/MatrixRTCSession.spec.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,10 @@ limitations under the License.
import { encodeBase64, EventType, MatrixClient, MatrixError, MatrixEvent, Room } from "../../../src";
import { KnownMembership } from "../../../src/@types/membership";
import { DEFAULT_EXPIRE_DURATION, SessionMembershipData } from "../../../src/matrixrtc/CallMembership";
import { MembershipManager } from "../../../src/matrixrtc/MembershipManager";
import { MatrixRTCSession, MatrixRTCSessionEvent } from "../../../src/matrixrtc/MatrixRTCSession";
import { EncryptionKeysEventContent } from "../../../src/matrixrtc/types";
import { randomString } from "../../../src/randomstring";
import { flushPromises } from "../../test-utils/flushPromises";
import { makeMockRoom, makeMockRoomState, membershipTemplate } from "./mocks";

const mockFocus = { type: "mock" };
Expand Down Expand Up @@ -236,16 +236,15 @@ describe("MatrixRTCSession", () => {
});

async function testSession(membershipData: SessionMembershipData): Promise<void> {
const makeNewMembershipSpy = jest.spyOn(MembershipManager.prototype as any, "makeNewMembership");
sess = MatrixRTCSession.roomSessionForRoom(client, makeMockRoom(membershipData));

sess.joinRoomSession([mockFocus], mockFocus, joinSessionConfig);
await Promise.race([sentStateEvent, new Promise((resolve) => setTimeout(resolve, 500))]);

expect(makeNewMembershipSpy).toHaveBeenCalledTimes(1);
expect(sendStateEventMock).toHaveBeenCalledTimes(1);

await Promise.race([sentDelayedState, new Promise((resolve) => setTimeout(resolve, 500))]);
expect(client._unstable_sendDelayedStateEvent).toHaveBeenCalledTimes(1);
expect(sendDelayedStateMock).toHaveBeenCalledTimes(1);
}

it("sends events", async () => {
Expand Down Expand Up @@ -323,9 +322,11 @@ describe("MatrixRTCSession", () => {
let sendStateEventMock: jest.Mock;
let sendDelayedStateMock: jest.Mock;
let sendEventMock: jest.Mock;
let updateDelayedEventMock: jest.Mock;

let sentStateEvent: Promise<void>;
let sentDelayedState: Promise<void>;
let updatedDelayedEvent: Promise<void>;

beforeEach(() => {
sentStateEvent = new Promise((resolve) => {
Expand All @@ -339,12 +340,15 @@ describe("MatrixRTCSession", () => {
};
});
});
updatedDelayedEvent = new Promise((r) => {
updateDelayedEventMock = jest.fn(r);
});
sendEventMock = jest.fn();
client.sendStateEvent = sendStateEventMock;
client._unstable_sendDelayedStateEvent = sendDelayedStateMock;
client.sendEvent = sendEventMock;

client._unstable_updateDelayedEvent = jest.fn();
client._unstable_updateDelayedEvent = updateDelayedEventMock;

mockRoom = makeMockRoom([]);
sess = MatrixRTCSession.roomSessionForRoom(client, mockRoom);
Expand Down Expand Up @@ -482,19 +486,7 @@ describe("MatrixRTCSession", () => {
membershipServerSideExpiryTimeout: 9000,
});

// needed to advance the mock timers properly
// depends on myMembershipManager being created
const scheduledDelayDisconnection = new Promise<void>((resolve) => {
const membershipManager = (sess as any).membershipManager;
const originalFn: () => void = membershipManager.scheduleDelayDisconnection;
membershipManager.scheduleDelayDisconnection = jest.fn(() => {
originalFn.call(membershipManager);
resolve();
});
});

await sendDelayedStateExceedAttempt.then(); // needed to resolve after the send attempt catches

await sendDelayedStateAttempt;
const callProps = (d: number) => {
return [mockRoom!.roomId, { delay: d }, "org.matrix.msc3401.call.member", {}, userStateKey];
Expand Down Expand Up @@ -525,11 +517,13 @@ describe("MatrixRTCSession", () => {
await sentDelayedState;

// should have prepared the heartbeat to keep delaying the leave event while still connected
await scheduledDelayDisconnection;
// should have tried updating the delayed leave to test that it wasn't replaced by own state
await updatedDelayedEvent;
expect(client._unstable_updateDelayedEvent).toHaveBeenCalledTimes(1);
// should update delayed disconnect

// ensures that we reach the code that schedules the timeout for the next delay update before we advance the timers.
await flushPromises();
jest.advanceTimersByTime(5000);
// should update delayed disconnect
expect(client._unstable_updateDelayedEvent).toHaveBeenCalledTimes(2);

jest.useRealTimers();
Expand Down Expand Up @@ -561,7 +555,7 @@ describe("MatrixRTCSession", () => {

const onMembershipsChanged = jest.fn();
sess.on(MatrixRTCSessionEvent.MembershipsChanged, onMembershipsChanged);
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();

expect(onMembershipsChanged).not.toHaveBeenCalled();
});
Expand All @@ -574,7 +568,7 @@ describe("MatrixRTCSession", () => {
sess.on(MatrixRTCSessionEvent.MembershipsChanged, onMembershipsChanged);

mockRoom.getLiveTimeline().getState = jest.fn().mockReturnValue(makeMockRoomState([], mockRoom.roomId));
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();

expect(onMembershipsChanged).toHaveBeenCalled();
});
Expand Down Expand Up @@ -763,7 +757,7 @@ describe("MatrixRTCSession", () => {
mockRoom.getLiveTimeline().getState = jest
.fn()
.mockReturnValue(makeMockRoomState([membershipTemplate], mockRoom.roomId));
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();

// member2 re-joins which should trigger an immediate re-send
const keysSentPromise2 = new Promise<EncryptionKeysEventContent>((resolve) => {
Expand All @@ -772,7 +766,7 @@ describe("MatrixRTCSession", () => {
mockRoom.getLiveTimeline().getState = jest
.fn()
.mockReturnValue(makeMockRoomState([membershipTemplate, member2], mockRoom.roomId));
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();
// but, that immediate resend is throttled so we need to wait a bit
jest.advanceTimersByTime(1000);
const { keys } = await keysSentPromise2;
Expand Down Expand Up @@ -825,7 +819,7 @@ describe("MatrixRTCSession", () => {
mockRoom.getLiveTimeline().getState = jest
.fn()
.mockReturnValue(makeMockRoomState([membershipTemplate, member2], mockRoom.roomId));
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();

await keysSentPromise2;

Expand Down Expand Up @@ -879,7 +873,7 @@ describe("MatrixRTCSession", () => {
sendEventMock.mockClear();

// these should be a no-op:
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();
expect(sendEventMock).toHaveBeenCalledTimes(0);
expect(sess!.statistics.counters.roomEventEncryptionKeysSent).toEqual(1);
} finally {
Expand Down Expand Up @@ -933,7 +927,7 @@ describe("MatrixRTCSession", () => {
sendEventMock.mockClear();

// this should be a no-op:
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();
expect(sendEventMock).toHaveBeenCalledTimes(0);

// advance time to avoid key throttling
Expand All @@ -947,7 +941,7 @@ describe("MatrixRTCSession", () => {
});

// this should re-send the key
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();

await keysSentPromise2;

Expand Down Expand Up @@ -1010,7 +1004,7 @@ describe("MatrixRTCSession", () => {
mockRoom.getLiveTimeline().getState = jest
.fn()
.mockReturnValue(makeMockRoomState([membershipTemplate], mockRoom.roomId));
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();

jest.advanceTimersByTime(10000);

Expand Down Expand Up @@ -1055,7 +1049,7 @@ describe("MatrixRTCSession", () => {
);
}

sess!.onMembershipUpdate();
sess!.onRTCSessionMemberUpdate();

// advance time to avoid key throttling
jest.advanceTimersByTime(10000);
Expand Down Expand Up @@ -1096,7 +1090,7 @@ describe("MatrixRTCSession", () => {
mockRoom.getLiveTimeline().getState = jest
.fn()
.mockReturnValue(makeMockRoomState([membershipTemplate, member2], mockRoom.roomId));
sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();

await new Promise((resolve) => {
realSetTimeout(resolve);
Expand Down
71 changes: 50 additions & 21 deletions src/matrixrtc/MatrixRTCSession.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import { decodeBase64, encodeUnpaddedBase64 } from "../base64.ts";
import { KnownMembership } from "../@types/membership.ts";
import { MatrixError, safeGetRetryAfterMs } from "../http-api/errors.ts";
import { MatrixEvent } from "../models/event.ts";
import { MembershipManager } from "./MembershipManager.ts";
import { LegacyMembershipManager, IMembershipManager } from "./MembershipManager.ts";

const logger = rootLogger.getChild("MatrixRTCSession");

Expand Down Expand Up @@ -132,7 +132,7 @@ export type JoinSessionConfig = MembershipConfig & EncryptionConfig;
* This class doesn't deal with media at all, just membership & properties of a session.
*/
export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, MatrixRTCSessionEventHandlerMap> {
private membershipManager?: MembershipManager;
private membershipManager?: IMembershipManager;

// The session Id of the call, this is the call_id of the call Member event.
private _callId: string | undefined;
Expand Down Expand Up @@ -283,7 +283,8 @@ export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, M
super();
this._callId = memberships[0]?.callId;
const roomState = this.room.getLiveTimeline().getState(EventTimeline.FORWARDS);
roomState?.on(RoomStateEvent.Members, this.onMembershipUpdate);
// TODO: double check if this is actually needed. Should be covered by refreshRoom in MatrixRTCSessionManager
roomState?.on(RoomStateEvent.Members, this.onRoomMemberUpdate);
this.setExpiryTimer();
}

Expand All @@ -299,14 +300,13 @@ export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, M
* Performs cleanup & removes timers for client shutdown
*/
public async stop(): Promise<void> {
await this.membershipManager?.leaveRoomSession(1000);
await this.membershipManager?.leave(1000);
if (this.expiryTimeout) {
clearTimeout(this.expiryTimeout);
this.expiryTimeout = undefined;
}
this.membershipManager?.stop();
const roomState = this.room.getLiveTimeline().getState(EventTimeline.FORWARDS);
roomState?.off(RoomStateEvent.Members, this.onMembershipUpdate);
roomState?.off(RoomStateEvent.Members, this.onRoomMemberUpdate);
}

/**
Expand All @@ -324,24 +324,21 @@ export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, M
* @param joinConfig - Additional configuration for the joined session.
*/
public joinRoomSession(fociPreferred: Focus[], fociActive?: Focus, joinConfig?: JoinSessionConfig): void {
this.joinConfig = joinConfig;
if (this.isJoined()) {
logger.info(`Already joined to session in room ${this.room.roomId}: ignoring join call`);
return;
} else {
this.membershipManager = new MembershipManager(joinConfig, this.room, this.client, () =>
this.membershipManager = new LegacyMembershipManager(joinConfig, this.room, this.client, () =>
this.getOldestMembership(),
);
}
this.joinConfig = joinConfig;
this.membershipManager!.join(fociPreferred, fociActive);
this.manageMediaKeys = joinConfig?.manageMediaKeys ?? this.manageMediaKeys;
// TODO: it feels wrong to be doing `setJoined()` and then `joinRoomSession()` non-atomically
// A new api between MembershipManager and the session will need to be defined.
this.membershipManager.setJoined(fociPreferred, fociActive);
if (joinConfig?.manageMediaKeys) {
this.makeNewSenderKey();
this.requestSendCurrentKey();
}
this.membershipManager.joinRoomSession();
this.emit(MatrixRTCSessionEvent.JoinStateChanged, true);
}

Expand Down Expand Up @@ -383,12 +380,17 @@ export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, M

logger.info(`Leaving call session in room ${this.room.roomId}`);
this.joinConfig = undefined;
this.membershipManager!.setLeft();
this.manageMediaKeys = false;
const leavePromise = this.membershipManager!.leave(timeout);
this.emit(MatrixRTCSessionEvent.JoinStateChanged, false);
return await this.membershipManager!.leaveRoomSession(timeout);
return await leavePromise;
}

/**
* Get the active focus from the current CallMemberState event
* @returns The focus that is currently in use to connect to this session. This is undefined
* if the client is not connected to this session.
*/
public getActiveFocus(): Focus | undefined {
return this.membershipManager?.getActiveFocus();
}
Expand Down Expand Up @@ -650,14 +652,21 @@ export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, M
}

if (soonestExpiry != undefined) {
this.expiryTimeout = setTimeout(this.onMembershipUpdate, soonestExpiry);
this.expiryTimeout = setTimeout(this.onRTCSessionMemberUpdate, soonestExpiry);
}
}

public getOldestMembership(): CallMembership | undefined {
return this.memberships[0];
}

/**
* This method is used when the user is not yet connected to the Session but wants to know what focus
* the users in the session are using to make a decision how it wants/should connect.
*
* See also `getActiveFocus`
* @returns The focus which should be used when joining this session.
*/
public getFocusInUse(): Focus | undefined {
const oldestMembership = this.getOldestMembership();
if (oldestMembership?.getFocusSelection() === "oldest_membership") {
Expand Down Expand Up @@ -746,11 +755,35 @@ export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, M
m.sender === this.client.getUserId() && m.deviceId === this.client.getDeviceId();

/**
* @deprecated use onRoomMemberUpdate or onRTCSessionMemberUpdate instead. this should be called when any membership in the call is updated
* the old name might have implied to only need to call this when your own membership changes.
*/
public onMembershipUpdate = (): void => {
this.recalculateSessionMembers();
};

/**
* Call this when the Matrix room members have changed.
*/
public onRoomMemberUpdate = (): void => {
this.recalculateSessionMembers();
};

/**
* Call this when something changed that may impacts the current MatrixRTC members in this session.
*/
public onRTCSessionMemberUpdate = (): void => {
this.recalculateSessionMembers();
};

/**
* Call this when anything that could impact rtc memberships has changed: Room Members or RTC members.
*
* Examines the latest call memberships and handles any encryption key sending or rotation that is needed.
*
* This function should be called when the room members or call memberships might have changed.
*/
public onMembershipUpdate = (): void => {
private recalculateSessionMembers = (): void => {
const oldMemberships = this.memberships;
this.memberships = MatrixRTCSession.callMembershipsForRoom(this.room);

Expand All @@ -764,11 +797,7 @@ export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, M
logger.info(`Memberships for call in room ${this.room.roomId} have changed: emitting`);
this.emit(MatrixRTCSessionEvent.MembershipsChanged, oldMemberships, this.memberships);

if (this.isJoined() && !this.memberships.some(this.isMyMembership)) {
logger.warn("Missing own membership: force re-join");
// TODO: Should this be awaited? And is there anything to tell the focus?
this.membershipManager?.triggerCallMembershipEventUpdate();
}
this.membershipManager?.onRTCSessionMemberUpdate(this.memberships);
}

if (this.manageMediaKeys && this.isJoined()) {
Expand Down
2 changes: 1 addition & 1 deletion src/matrixrtc/MatrixRTCSessionManager.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,7 @@ export class MatrixRTCSessionManager extends TypedEventEmitter<MatrixRTCSessionM

const wasActiveAndKnown = sess.memberships.length > 0 && !isNewSession;

sess.onMembershipUpdate();
sess.onRTCSessionMemberUpdate();

const nowActive = sess.memberships.length > 0;

Expand Down
Loading

0 comments on commit ffb228b

Please sign in to comment.