mirror of
https://github.com/vector-im/hydrogen-web.git
synced 2025-02-02 23:51:38 +01:00
Merge branch 'bwindels/calls' into thirdroom/dev
This commit is contained in:
commit
537a910420
@ -31,7 +31,7 @@
|
||||
},
|
||||
"homepage": "https://github.com/vector-im/hydrogen-web/#readme",
|
||||
"devDependencies": {
|
||||
"@matrixdotorg/structured-logviewer": "^0.0.1",
|
||||
"@matrixdotorg/structured-logviewer": "^0.0.3",
|
||||
"@typescript-eslint/eslint-plugin": "^4.29.2",
|
||||
"@typescript-eslint/parser": "^4.29.2",
|
||||
"acorn": "^8.6.0",
|
||||
|
@ -30,10 +30,11 @@ export class Logger implements ILogger {
|
||||
this._platform = platform;
|
||||
}
|
||||
|
||||
log(labelOrValues: LabelOrValues, logLevel: LogLevel = LogLevel.Info): void {
|
||||
log(labelOrValues: LabelOrValues, logLevel: LogLevel = LogLevel.Info): ILogItem {
|
||||
const item = new LogItem(labelOrValues, logLevel, this);
|
||||
item.end = item.start;
|
||||
this._persistItem(item, undefined, false);
|
||||
return item;
|
||||
}
|
||||
|
||||
/** Prefer `run()` or `log()` above this method; only use it if you have a long-running operation
|
||||
|
@ -21,7 +21,9 @@ function noop (): void {}
|
||||
export class NullLogger implements ILogger {
|
||||
public readonly item: ILogItem = new NullLogItem(this);
|
||||
|
||||
log(): void {}
|
||||
log(): ILogItem {
|
||||
return this.item;
|
||||
}
|
||||
|
||||
addReporter() {}
|
||||
|
||||
|
@ -68,7 +68,7 @@ export interface ILogItemCreator {
|
||||
*/
|
||||
|
||||
export interface ILogger {
|
||||
log(labelOrValues: LabelOrValues, logLevel?: LogLevel): void;
|
||||
log(labelOrValues: LabelOrValues, logLevel?: LogLevel): ILogItem;
|
||||
child(labelOrValues: LabelOrValues, logLevel?: LogLevel, filterCreator?: FilterCreator): ILogItem;
|
||||
wrapOrRun<T>(item: ILogItem | undefined, labelOrValues: LabelOrValues, callback: LogCallback<T>, logLevel?: LogLevel, filterCreator?: FilterCreator): T;
|
||||
runDetached<T>(labelOrValues: LabelOrValues, callback: LogCallback<T>, logLevel?: LogLevel, filterCreator?: FilterCreator): ILogItem;
|
||||
|
@ -78,6 +78,7 @@ export class Session {
|
||||
this._user = new User(sessionInfo.userId);
|
||||
this._callHandler = new CallHandler({
|
||||
clock: this._platform.clock,
|
||||
random: this._platform.random,
|
||||
hsApi: this._hsApi,
|
||||
encryptDeviceMessage: async (roomId, userId, deviceId, message, log) => {
|
||||
if (!this._deviceTracker || !this._olmEncryption) {
|
||||
|
@ -55,12 +55,10 @@ export class CallHandler implements RoomStateHandler {
|
||||
private roomMemberToCallIds: Map<string, Set<string>> = new Map();
|
||||
private groupCallOptions: GroupCallOptions;
|
||||
private sessionId = makeId("s");
|
||||
private turnServerSource: TurnServerSource;
|
||||
|
||||
constructor(private readonly options: Options) {
|
||||
this.turnServerSource = new TurnServerSource(this.options.hsApi, this.options.clock);
|
||||
this.groupCallOptions = Object.assign({}, this.options, {
|
||||
turnServerSource: this.turnServerSource,
|
||||
turnServerSource: new TurnServerSource(this.options.hsApi, this.options.clock),
|
||||
emitUpdate: (groupCall, params) => this._calls.update(groupCall.id, params),
|
||||
createTimeout: this.options.clock.createTimeout,
|
||||
sessionId: this.sessionId
|
||||
@ -248,9 +246,10 @@ export class CallHandler implements RoomStateHandler {
|
||||
}
|
||||
|
||||
dispose() {
|
||||
this.turnServerSource.dispose();
|
||||
const joinedCalls = Array.from(this._calls.values()).filter(c => c.hasJoined);
|
||||
Promise.all(joinedCalls.map(c => c.leave())).then(() => {}, () => {});
|
||||
this.groupCallOptions.turnServerSource.dispose();
|
||||
for(const call of this._calls.values()) {
|
||||
call.dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -771,7 +771,8 @@ export class PeerCall implements IDisposable {
|
||||
const {flushCandidatesLog} = this;
|
||||
// MSC2746 recommends these values (can be quite long when calling because the
|
||||
// callee will need a while to answer the call)
|
||||
await this.delay(this.direction === CallDirection.Inbound ? 500 : 2000);
|
||||
try { await this.delay(this.direction === CallDirection.Inbound ? 500 : 2000); }
|
||||
catch (err) { return; }
|
||||
this.sendCandidateQueue(flushCandidatesLog);
|
||||
this.flushCandidatesLog = undefined;
|
||||
}
|
||||
@ -1098,9 +1099,12 @@ export class PeerCall implements IDisposable {
|
||||
private async delay(timeoutMs: number): Promise<void> {
|
||||
// Allow a short time for initial candidates to be gathered
|
||||
const timeout = this.disposables.track(this.options.createTimeout(timeoutMs));
|
||||
try {
|
||||
await timeout.elapsed();
|
||||
} finally {
|
||||
this.disposables.untrack(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
private sendSignallingMessage(message: SignallingMessage<MCallBase>, log: ILogItem) {
|
||||
return log.wrap({l: "send", id: message.type}, async log => {
|
||||
|
@ -24,7 +24,9 @@ export const SDPStreamMetadataKey = "org.matrix.msc3077.sdp_stream_metadata";
|
||||
|
||||
export interface CallDeviceMembership {
|
||||
device_id: string,
|
||||
session_id: string
|
||||
session_id: string,
|
||||
["m.expires_ts"]?: number,
|
||||
feeds?: Array<{purpose: string}>
|
||||
}
|
||||
|
||||
export interface CallMembership {
|
||||
|
@ -59,3 +59,4 @@ export class MuteSettings {
|
||||
}
|
||||
|
||||
export const CALL_LOG_TYPE = "call";
|
||||
export const CALL_MEMBER_VALIDITY_PERIOD_MS = 3600 * 1000; // 1h
|
||||
|
@ -15,9 +15,9 @@ limitations under the License.
|
||||
*/
|
||||
|
||||
import {ObservableMap} from "../../../observable/map/ObservableMap";
|
||||
import {Member} from "./Member";
|
||||
import {Member, isMemberExpired, memberExpiresAt} from "./Member";
|
||||
import {LocalMedia} from "../LocalMedia";
|
||||
import {MuteSettings, CALL_LOG_TYPE} from "../common";
|
||||
import {MuteSettings, CALL_LOG_TYPE, CALL_MEMBER_VALIDITY_PERIOD_MS} from "../common";
|
||||
import {MemberChange, RoomMember} from "../../room/members/RoomMember";
|
||||
import {EventEmitter} from "../../../utils/EventEmitter";
|
||||
import {EventType, CallIntent} from "../callEventTypes";
|
||||
@ -26,7 +26,7 @@ import type {Options as MemberOptions} from "./Member";
|
||||
import type {TurnServerSource} from "../TurnServerSource";
|
||||
import type {BaseObservableMap} from "../../../observable/map/BaseObservableMap";
|
||||
import type {Track} from "../../../platform/types/MediaDevices";
|
||||
import type {SignallingMessage, MGroupCallBase, CallMembership} from "../callEventTypes";
|
||||
import type {SignallingMessage, MGroupCallBase, CallMembership, CallMemberContent, CallDeviceMembership} from "../callEventTypes";
|
||||
import type {Room} from "../../room/Room";
|
||||
import type {StateEvent} from "../../storage/types";
|
||||
import type {Platform} from "../../../platform/web/Platform";
|
||||
@ -34,6 +34,7 @@ import type {EncryptedMessage} from "../../e2ee/olm/Encryption";
|
||||
import type {ILogItem, ILogger} from "../../../logging/types";
|
||||
import type {Storage} from "../../storage/idb/Storage";
|
||||
import type {BaseObservableValue} from "../../../observable/value/BaseObservableValue";
|
||||
import type {Clock, Timeout} from "../../../platform/web/dom/Clock";
|
||||
|
||||
export enum GroupCallState {
|
||||
Fledgling = "fledgling",
|
||||
@ -59,22 +60,26 @@ export type Options = Omit<MemberOptions, "emitUpdate" | "confId" | "encryptDevi
|
||||
emitUpdate: (call: GroupCall, params?: any) => void;
|
||||
encryptDeviceMessage: (roomId: string, userId: string, deviceId: string, message: SignallingMessage<MGroupCallBase>, log: ILogItem) => Promise<EncryptedMessage | undefined>,
|
||||
storage: Storage,
|
||||
random: () => number,
|
||||
logger: ILogger,
|
||||
turnServerSource: TurnServerSource
|
||||
};
|
||||
|
||||
class JoinedData {
|
||||
public renewMembershipTimeout?: Timeout;
|
||||
|
||||
constructor(
|
||||
public readonly logItem: ILogItem,
|
||||
public readonly membersLogItem: ILogItem,
|
||||
public localMedia: LocalMedia,
|
||||
public localMuteSettings: MuteSettings,
|
||||
public turnServer: BaseObservableValue<RTCIceServer>
|
||||
public readonly turnServer: BaseObservableValue<RTCIceServer>
|
||||
) {}
|
||||
|
||||
dispose() {
|
||||
this.localMedia.dispose();
|
||||
this.logItem.finish();
|
||||
this.renewMembershipTimeout?.dispose();
|
||||
}
|
||||
}
|
||||
|
||||
@ -83,6 +88,7 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
private _memberOptions: MemberOptions;
|
||||
private _state: GroupCallState;
|
||||
private bufferedDeviceMessages = new Map<string, Set<SignallingMessage<MGroupCallBase>>>();
|
||||
/** Set between calling join and leave. */
|
||||
private joinedData?: JoinedData;
|
||||
|
||||
private _deviceIndex?: number;
|
||||
@ -99,7 +105,22 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
this._state = newCall ? GroupCallState.Fledgling : GroupCallState.Created;
|
||||
this._memberOptions = Object.assign({}, options, {
|
||||
confId: this.id,
|
||||
emitUpdate: member => this._members.update(getMemberKey(member.userId, member.deviceId), member),
|
||||
emitUpdate: member => {
|
||||
const memberKey = getMemberKey(member.userId, member.deviceId);
|
||||
// only remove expired members to whom we're not already connected
|
||||
if (member.isExpired && !member.isConnected) {
|
||||
const logItem = this.options.logger.log({
|
||||
l: "removing expired member from call",
|
||||
memberKey,
|
||||
callId: this.id
|
||||
})
|
||||
member.logItem?.refDetached(logItem);
|
||||
member.dispose();
|
||||
this._members.remove(memberKey);
|
||||
} else {
|
||||
this._members.update(memberKey, member);
|
||||
}
|
||||
},
|
||||
encryptDeviceMessage: (userId: string, deviceId: string, message: SignallingMessage<MGroupCallBase>, log) => {
|
||||
return this.options.encryptDeviceMessage(this.roomId, userId, deviceId, message, log);
|
||||
}
|
||||
@ -167,7 +188,7 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
this._state = GroupCallState.Joining;
|
||||
this.emitChange();
|
||||
await log.wrap("update member state", async log => {
|
||||
const memberContent = await this._createJoinPayload();
|
||||
const memberContent = await this._createMemberPayload(true);
|
||||
log.set("payload", memberContent);
|
||||
// send m.call.member state event
|
||||
const request = this.options.hsApi.sendState(this.roomId, EventType.GroupCallMember, this.options.ownUserId, memberContent, {log});
|
||||
@ -231,7 +252,9 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
}
|
||||
await joinedData.logItem.wrap("leave", async log => {
|
||||
try {
|
||||
const memberContent = await this._leaveCallMemberContent();
|
||||
joinedData.renewMembershipTimeout?.dispose();
|
||||
joinedData.renewMembershipTimeout = undefined;
|
||||
const memberContent = await this._createMemberPayload(false);
|
||||
// send m.call.member state event
|
||||
if (memberContent) {
|
||||
const request = this.options.hsApi.sendState(this.roomId, EventType.GroupCallMember, this.options.ownUserId, memberContent, {log});
|
||||
@ -307,24 +330,43 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
/** @internal */
|
||||
updateMembership(userId: string, roomMember: RoomMember, callMembership: CallMembership, eventTimestamp: number, syncLog: ILogItem) {
|
||||
syncLog.wrap({l: "update call membership", t: CALL_LOG_TYPE, id: this.id, userId}, log => {
|
||||
const now = this.options.clock.now();
|
||||
const devices = callMembership["m.devices"];
|
||||
const previousDeviceIds = this.getDeviceIdsForUserId(userId);
|
||||
for (let deviceIndex = 0; deviceIndex < devices.length; deviceIndex++) {
|
||||
const device = devices[deviceIndex];
|
||||
const deviceId = device.device_id;
|
||||
const memberKey = getMemberKey(userId, deviceId);
|
||||
log.wrap({l: "update device membership", id: memberKey, sessionId: device.session_id}, log => {
|
||||
if (userId === this.options.ownUserId && deviceId === this.options.ownDeviceId) {
|
||||
|
||||
this._deviceIndex = deviceIndex;
|
||||
this._eventTimestamp = eventTimestamp;
|
||||
|
||||
log.wrap("update own membership", log => {
|
||||
if (this.hasJoined) {
|
||||
if (this.joinedData) {
|
||||
this.joinedData.logItem.refDetached(log);
|
||||
}
|
||||
this._setupRenewMembershipTimeout(device, log);
|
||||
}
|
||||
if (this._state === GroupCallState.Joining) {
|
||||
log.set("update_own", true);
|
||||
log.set("joined", true);
|
||||
this._state = GroupCallState.Joined;
|
||||
this.emitChange();
|
||||
}
|
||||
});
|
||||
} else {
|
||||
log.wrap({l: "update device membership", id: memberKey, sessionId: device.session_id}, log => {
|
||||
if (isMemberExpired(device, now)) {
|
||||
log.set("expired", true);
|
||||
const member = this._members.get(memberKey);
|
||||
if (member) {
|
||||
member.dispose();
|
||||
this._members.remove(memberKey);
|
||||
log.set("removed", true);
|
||||
}
|
||||
return;
|
||||
}
|
||||
let member = this._members.get(memberKey);
|
||||
const sessionIdChanged = member && member.sessionId !== device.session_id;
|
||||
if (member && !sessionIdChanged) {
|
||||
@ -337,6 +379,7 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
if (disconnectLogItem) {
|
||||
log.refDetached(disconnectLogItem);
|
||||
}
|
||||
member.dispose();
|
||||
this._members.remove(memberKey);
|
||||
member = undefined;
|
||||
}
|
||||
@ -344,6 +387,7 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
member = new Member(
|
||||
roomMember,
|
||||
device, deviceIndex, eventTimestamp, this._memberOptions,
|
||||
log
|
||||
);
|
||||
this._members.add(memberKey, member);
|
||||
if (this.joinedData) {
|
||||
@ -353,9 +397,9 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
// flush pending messages, either after having created the member,
|
||||
// or updated the session id with updateCallInfo
|
||||
this.flushPendingIncomingDeviceMessages(member, log);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const newDeviceIds = new Set<string>(devices.map(call => call.device_id));
|
||||
// remove user as member of any calls not present anymore
|
||||
@ -416,13 +460,14 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
}
|
||||
|
||||
private removeOwnDevice(log: ILogItem) {
|
||||
log.set("leave_own", true);
|
||||
log.wrap("remove own membership", log => {
|
||||
this.disconnect(log);
|
||||
});
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
disconnect(log: ILogItem) {
|
||||
if (this._state === GroupCallState.Joined) {
|
||||
if (this.hasJoined) {
|
||||
for (const [,member] of this._members) {
|
||||
const disconnectLogItem = member.disconnect(true);
|
||||
if (disconnectLogItem) {
|
||||
@ -443,11 +488,12 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
const member = this._members.get(memberKey);
|
||||
if (member) {
|
||||
log.set("leave", true);
|
||||
this._members.remove(memberKey);
|
||||
const disconnectLogItem = member.disconnect(false);
|
||||
if (disconnectLogItem) {
|
||||
log.refDetached(disconnectLogItem);
|
||||
}
|
||||
member.dispose();
|
||||
this._members.remove(memberKey);
|
||||
}
|
||||
this.emitChange();
|
||||
});
|
||||
@ -482,14 +528,14 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
}
|
||||
}
|
||||
|
||||
private async _createJoinPayload() {
|
||||
private async _createMemberPayload(includeOwn: boolean): Promise<CallMemberContent> {
|
||||
const {storage} = this.options;
|
||||
const txn = await storage.readTxn([storage.storeNames.roomState]);
|
||||
const stateEvent = await txn.roomState.get(this.roomId, EventType.GroupCallMember, this.options.ownUserId);
|
||||
const stateContent = stateEvent?.event?.content ?? {
|
||||
const stateContent: CallMemberContent = stateEvent?.event?.content as CallMemberContent ?? {
|
||||
["m.calls"]: []
|
||||
};
|
||||
const callsInfo = stateContent["m.calls"];
|
||||
let callsInfo = stateContent["m.calls"];
|
||||
let callInfo = callsInfo.find(c => c["m.call_id"] === this.id);
|
||||
if (!callInfo) {
|
||||
callInfo = {
|
||||
@ -498,42 +544,42 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
};
|
||||
callsInfo.push(callInfo);
|
||||
}
|
||||
callInfo["m.devices"] = callInfo["m.devices"].filter(d => d["device_id"] !== this.options.ownDeviceId);
|
||||
const now = this.options.clock.now();
|
||||
callInfo["m.devices"] = callInfo["m.devices"].filter(d => {
|
||||
// remove our own device (to add it again below)
|
||||
if (d["device_id"] === this.options.ownDeviceId) {
|
||||
return false;
|
||||
}
|
||||
// also remove any expired devices (+ the validity period added again)
|
||||
if (memberExpiresAt(d) === undefined || isMemberExpired(d, now, CALL_MEMBER_VALIDITY_PERIOD_MS)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
});
|
||||
if (includeOwn) {
|
||||
callInfo["m.devices"].push({
|
||||
["device_id"]: this.options.ownDeviceId,
|
||||
["session_id"]: this.options.sessionId,
|
||||
["m.expires_ts"]: now + CALL_MEMBER_VALIDITY_PERIOD_MS,
|
||||
feeds: [{purpose: "m.usermedia"}]
|
||||
});
|
||||
|
||||
this._deviceIndex = callInfo["m.devices"].length;
|
||||
this._eventTimestamp = Date.now();
|
||||
|
||||
}
|
||||
// filter out empty call membership
|
||||
stateContent["m.calls"] = callsInfo.filter(c => c["m.devices"].length !== 0);
|
||||
return stateContent;
|
||||
}
|
||||
|
||||
private async _leaveCallMemberContent(): Promise<Record<string, any> | undefined> {
|
||||
const {storage} = this.options;
|
||||
const txn = await storage.readTxn([storage.storeNames.roomState]);
|
||||
const stateEvent = await txn.roomState.get(this.roomId, EventType.GroupCallMember, this.options.ownUserId);
|
||||
if (stateEvent) {
|
||||
const content = stateEvent.event.content;
|
||||
const callInfo = content["m.calls"]?.find(c => c["m.call_id"] === this.id);
|
||||
if (callInfo) {
|
||||
const devicesInfo = callInfo["m.devices"];
|
||||
const deviceIndex = devicesInfo.findIndex(d => d["device_id"] === this.options.ownDeviceId);
|
||||
if (deviceIndex !== -1) {
|
||||
devicesInfo.splice(deviceIndex, 1);
|
||||
return content;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private connectToMember(member: Member, joinedData: JoinedData, log: ILogItem) {
|
||||
const memberKey = getMemberKey(member.userId, member.deviceId);
|
||||
const logItem = joinedData.membersLogItem.child({l: "member", id: memberKey});
|
||||
logItem.set("sessionId", member.sessionId);
|
||||
const logItem = joinedData.membersLogItem.child({
|
||||
l: "member",
|
||||
id: memberKey,
|
||||
sessionId: member.sessionId
|
||||
});
|
||||
log.wrap({l: "connect", id: memberKey}, log => {
|
||||
const connectItem = member.connect(
|
||||
joinedData.localMedia,
|
||||
@ -544,11 +590,52 @@ export class GroupCall extends EventEmitter<{change: never}> {
|
||||
if (connectItem) {
|
||||
log.refDetached(connectItem);
|
||||
}
|
||||
})
|
||||
});
|
||||
}
|
||||
|
||||
protected emitChange() {
|
||||
this.emit("change");
|
||||
this.options.emitUpdate(this);
|
||||
}
|
||||
|
||||
private _setupRenewMembershipTimeout(callDeviceMembership: CallDeviceMembership, log: ILogItem) {
|
||||
const {joinedData} = this;
|
||||
if (!joinedData) {
|
||||
return;
|
||||
}
|
||||
joinedData.renewMembershipTimeout?.dispose();
|
||||
joinedData.renewMembershipTimeout = undefined;
|
||||
const expiresAt = memberExpiresAt(callDeviceMembership);
|
||||
if (typeof expiresAt !== "number") {
|
||||
return;
|
||||
}
|
||||
const expiresFromNow = expiresAt - this.options.clock.now();
|
||||
// renew 1 to 5 minutes (8.3% of 1h, but min 10s) before expiring
|
||||
// do it a bit beforehand and somewhat random to not collide with
|
||||
// other clients trying to renew as well
|
||||
const timeToRenewBeforeExpiration = Math.max(10000, Math.ceil((0.2 +(this.options.random() * 0.8)) * (0.08333 * CALL_MEMBER_VALIDITY_PERIOD_MS)));
|
||||
const renewFromNow = Math.max(0, expiresFromNow - timeToRenewBeforeExpiration);
|
||||
log.set("expiresIn", expiresFromNow);
|
||||
log.set("renewIn", renewFromNow);
|
||||
joinedData.renewMembershipTimeout = this.options.clock.createTimeout(renewFromNow);
|
||||
joinedData.renewMembershipTimeout.elapsed().then(
|
||||
() => {
|
||||
joinedData.logItem.wrap("renew membership", async log => {
|
||||
const memberContent = await this._createMemberPayload(true);
|
||||
log.set("payload", memberContent);
|
||||
// send m.call.member state event
|
||||
const request = this.options.hsApi.sendState(this.roomId, EventType.GroupCallMember, this.options.ownUserId, memberContent, {log});
|
||||
await request.response();
|
||||
});
|
||||
},
|
||||
() => { /* assume we're swallowing AbortError from dispose above */ }
|
||||
);
|
||||
}
|
||||
|
||||
dispose() {
|
||||
this.joinedData?.dispose();
|
||||
for (const member of this._members.values()) {
|
||||
member.dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -30,6 +30,7 @@ import type {RoomMember} from "../../room/members/RoomMember";
|
||||
import type {EncryptedMessage} from "../../e2ee/olm/Encryption";
|
||||
import type {ILogItem} from "../../../logging/types";
|
||||
import type {BaseObservableValue} from "../../../observable/value/BaseObservableValue";
|
||||
import type {Clock, Timeout} from "../../../platform/web/dom/Clock";
|
||||
|
||||
export type Options = Omit<PeerCallOptions, "emitUpdate" | "sendSignallingMessage" | "turnServer"> & {
|
||||
confId: string,
|
||||
@ -40,6 +41,7 @@ export type Options = Omit<PeerCallOptions, "emitUpdate" | "sendSignallingMessag
|
||||
hsApi: HomeServerApi,
|
||||
encryptDeviceMessage: (userId: string, deviceId: string, message: SignallingMessage<MGroupCallBase>, log: ILogItem) => Promise<EncryptedMessage | undefined>,
|
||||
emitUpdate: (participant: Member, params?: any) => void,
|
||||
clock: Clock
|
||||
}
|
||||
|
||||
const errorCodesWithoutRetry = [
|
||||
@ -65,18 +67,61 @@ class MemberConnection {
|
||||
public turnServer: BaseObservableValue<RTCIceServer>,
|
||||
public readonly logItem: ILogItem
|
||||
) {}
|
||||
|
||||
get canDequeueNextSignallingMessage() {
|
||||
if (this.queuedSignallingMessages.length === 0) {
|
||||
return false;
|
||||
}
|
||||
if (this.lastProcessedSeqNr === undefined) {
|
||||
return true;
|
||||
}
|
||||
const first = this.queuedSignallingMessages[0];
|
||||
// allow messages with both a seq we've just seen and
|
||||
// the next one to be dequeued as it can happen
|
||||
// that messages for other callIds (which could repeat seq)
|
||||
// are present in the queue
|
||||
return first.content.seq === this.lastProcessedSeqNr ||
|
||||
first.content.seq === this.lastProcessedSeqNr + 1;
|
||||
}
|
||||
|
||||
dispose() {
|
||||
this.peerCall?.dispose();
|
||||
this.localMedia.dispose();
|
||||
this.logItem.finish();
|
||||
}
|
||||
}
|
||||
|
||||
export class Member {
|
||||
private connection?: MemberConnection;
|
||||
private expireTimeout?: Timeout;
|
||||
|
||||
constructor(
|
||||
public member: RoomMember,
|
||||
private callDeviceMembership: CallDeviceMembership,
|
||||
private _deviceIndex: number,
|
||||
private _eventTimestamp: number,
|
||||
private readonly options: Options,
|
||||
) {}
|
||||
private options: Options,
|
||||
updateMemberLog: ILogItem
|
||||
) {
|
||||
this._renewExpireTimeout(updateMemberLog);
|
||||
}
|
||||
|
||||
private _renewExpireTimeout(log: ILogItem) {
|
||||
this.expireTimeout?.dispose();
|
||||
this.expireTimeout = undefined;
|
||||
const expiresAt = memberExpiresAt(this.callDeviceMembership);
|
||||
if (typeof expiresAt !== "number") {
|
||||
return;
|
||||
}
|
||||
const expiresFromNow = Math.max(0, expiresAt - this.options.clock.now());
|
||||
log?.set("expiresIn", expiresFromNow);
|
||||
// add 10ms to make sure isExpired returns true
|
||||
this.expireTimeout = this.options.clock.createTimeout(expiresFromNow + 10);
|
||||
this.expireTimeout.elapsed().then(
|
||||
() => { this.options.emitUpdate(this, "isExpired"); },
|
||||
(err) => { /* ignore abort error */ },
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Gives access the log item for this item once joined to the group call.
|
||||
@ -91,6 +136,11 @@ export class Member {
|
||||
return this.connection?.peerCall?.remoteMedia;
|
||||
}
|
||||
|
||||
get isExpired(): boolean {
|
||||
// never consider a peer we're connected to, to be expired
|
||||
return !this.isConnected && isMemberExpired(this.callDeviceMembership, this.options.clock.now());
|
||||
}
|
||||
|
||||
get remoteMuteSettings(): MuteSettings | undefined {
|
||||
return this.connection?.peerCall?.remoteMuteSettings;
|
||||
}
|
||||
@ -176,18 +226,15 @@ export class Member {
|
||||
return;
|
||||
}
|
||||
let disconnectLogItem;
|
||||
// if if not sending the hangup, still log disconnect
|
||||
connection.logItem.wrap("disconnect", async log => {
|
||||
disconnectLogItem = log;
|
||||
if (hangup) {
|
||||
await connection.peerCall?.hangup(CallErrorCode.UserHangup, log);
|
||||
} else {
|
||||
await connection.peerCall?.close(undefined, log);
|
||||
if (hangup && connection.peerCall) {
|
||||
await connection.peerCall.hangup(CallErrorCode.UserHangup, log);
|
||||
}
|
||||
connection.peerCall?.dispose();
|
||||
connection.localMedia?.dispose();
|
||||
this.connection = undefined;
|
||||
});
|
||||
connection.logItem.finish();
|
||||
connection.dispose();
|
||||
this.connection = undefined;
|
||||
return disconnectLogItem;
|
||||
}
|
||||
|
||||
@ -202,6 +249,7 @@ export class Member {
|
||||
this._deviceIndex = deviceIndex;
|
||||
this._eventTimestamp = eventTimestamp;
|
||||
|
||||
this._renewExpireTimeout(causeItem);
|
||||
if (this.connection) {
|
||||
this.connection.logItem.refDetached(causeItem);
|
||||
}
|
||||
@ -298,6 +346,7 @@ export class Member {
|
||||
}
|
||||
if (shouldReplace) {
|
||||
connection.peerCall = undefined;
|
||||
action = IncomingMessageAction.Handle;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -324,12 +373,7 @@ export class Member {
|
||||
|
||||
private dequeueSignallingMessages(connection: MemberConnection, peerCall: PeerCall, newMessage: SignallingMessage<MGroupCallBase>, syncLog: ILogItem): boolean {
|
||||
let hasNewMessageBeenDequeued = false;
|
||||
while (
|
||||
connection.queuedSignallingMessages.length && (
|
||||
connection.lastProcessedSeqNr === undefined ||
|
||||
connection.queuedSignallingMessages[0].content.seq === connection.lastProcessedSeqNr + 1
|
||||
)
|
||||
) {
|
||||
while (connection.canDequeueNextSignallingMessage) {
|
||||
const message = connection.queuedSignallingMessages.shift()!;
|
||||
if (message === newMessage) {
|
||||
hasNewMessageBeenDequeued = true;
|
||||
@ -370,4 +414,25 @@ export class Member {
|
||||
turnServer: connection.turnServer
|
||||
}), connection.logItem);
|
||||
}
|
||||
|
||||
dispose() {
|
||||
this.connection?.dispose();
|
||||
this.connection = undefined;
|
||||
this.expireTimeout?.dispose();
|
||||
this.expireTimeout = undefined;
|
||||
// ensure the emitUpdate callback can't be called anymore
|
||||
this.options = undefined as any as Options;
|
||||
}
|
||||
}
|
||||
|
||||
export function memberExpiresAt(callDeviceMembership: CallDeviceMembership): number | undefined {
|
||||
const expiresAt = callDeviceMembership["m.expires_ts"];
|
||||
if (Number.isSafeInteger(expiresAt)) {
|
||||
return expiresAt;
|
||||
}
|
||||
}
|
||||
|
||||
export function isMemberExpired(callDeviceMembership: CallDeviceMembership, now: number, margin: number = 0) {
|
||||
const expiresAt = memberExpiresAt(callDeviceMembership);
|
||||
return typeof expiresAt === "number" && ((expiresAt + margin) <= now);
|
||||
}
|
||||
|
@ -371,7 +371,7 @@ export class DeviceTracker {
|
||||
// verify signature
|
||||
const verifiedKeysPerUser = log.wrap("verify", log => this._filterVerifiedDeviceKeys(deviceKeyResponse["device_keys"], log));
|
||||
//// END EXTRACT
|
||||
|
||||
// TODO: what if verifiedKeysPerUser is empty or does not contain userId?
|
||||
const verifiedKeys = verifiedKeysPerUser
|
||||
.find(vkpu => vkpu.userId === userId).verifiedKeys
|
||||
.find(vk => vk["device_id"] === deviceId);
|
||||
|
@ -50,8 +50,8 @@ export class Disposables {
|
||||
}
|
||||
|
||||
untrack(disposable: Disposable): undefined {
|
||||
if (this.isDisposed) {
|
||||
console.warn("Disposables already disposed, cannot untrack");
|
||||
// already disposed
|
||||
if (!this._disposables) {
|
||||
return undefined;
|
||||
}
|
||||
const idx = this._disposables!.indexOf(disposable);
|
||||
|
@ -56,10 +56,10 @@
|
||||
version "3.2.8"
|
||||
resolved "https://gitlab.matrix.org/api/v4/projects/27/packages/npm/@matrix-org/olm/-/@matrix-org/olm-3.2.8.tgz#8d53636d045e1776e2a2ec6613e57330dd9ce856"
|
||||
|
||||
"@matrixdotorg/structured-logviewer@^0.0.1":
|
||||
version "0.0.1"
|
||||
resolved "https://registry.yarnpkg.com/@matrixdotorg/structured-logviewer/-/structured-logviewer-0.0.1.tgz#9c29470b552f874afbb1df16c6e8e9e0c55cbf59"
|
||||
integrity sha512-IdPYxAFDEoEs2G1ImKCkCxFI3xF1DDctP3N9JOtHRvIPbPPdTT9DyNqKTewCb5zwjNB1mGBrnWyURnHDiOOL3w==
|
||||
"@matrixdotorg/structured-logviewer@^0.0.3":
|
||||
version "0.0.3"
|
||||
resolved "https://registry.yarnpkg.com/@matrixdotorg/structured-logviewer/-/structured-logviewer-0.0.3.tgz#1555111159d83cde0cfd5ba1a571e1faa1a90871"
|
||||
integrity sha512-QqFglx0M8ix0IoRsJXDg1If26ltbYfuLjJ0MQrJYze3yz4ayEESRpQEA0YxJRVVtbco5M94tmrDpikokTFnn3A==
|
||||
|
||||
"@nodelib/fs.scandir@2.1.5":
|
||||
version "2.1.5"
|
||||
|
Loading…
x
Reference in New Issue
Block a user