feat: complete Ancestor quote bot with production-ready Docker support
This commit is contained in:
unfunny
484
node_modules/matrix-js-sdk/lib/sliding-sync.js
generated
vendored
484
node_modules/matrix-js-sdk/lib/sliding-sync.js
generated
vendored
@@ -1,31 +1,40 @@
|
||||
import _defineProperty from "@babel/runtime/helpers/defineProperty";
|
||||
/*
|
||||
Copyright 2022-2024 The Matrix.org Foundation C.I.C.
|
||||
"use strict";
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
import { logger } from "./logger.js";
|
||||
import { TypedEventEmitter } from "./models/typed-event-emitter.js";
|
||||
import { sleep } from "./utils.js";
|
||||
var _interopRequireDefault = require("@babel/runtime/helpers/interopRequireDefault");
|
||||
Object.defineProperty(exports, "__esModule", {
|
||||
value: true
|
||||
});
|
||||
exports.SlidingSyncState = exports.SlidingSyncEvent = exports.SlidingSync = exports.MSC3575_WILDCARD = exports.MSC3575_STATE_KEY_ME = exports.MSC3575_STATE_KEY_LAZY = exports.ExtensionState = void 0;
|
||||
var _defineProperty2 = _interopRequireDefault(require("@babel/runtime/helpers/defineProperty"));
|
||||
var _logger = require("./logger");
|
||||
var _typedEventEmitter = require("./models/typed-event-emitter");
|
||||
var _utils = require("./utils");
|
||||
function ownKeys(object, enumerableOnly) { var keys = Object.keys(object); if (Object.getOwnPropertySymbols) { var symbols = Object.getOwnPropertySymbols(object); enumerableOnly && (symbols = symbols.filter(function (sym) { return Object.getOwnPropertyDescriptor(object, sym).enumerable; })), keys.push.apply(keys, symbols); } return keys; }
|
||||
function _objectSpread(target) { for (var i = 1; i < arguments.length; i++) { var source = null != arguments[i] ? arguments[i] : {}; i % 2 ? ownKeys(Object(source), !0).forEach(function (key) { (0, _defineProperty2.default)(target, key, source[key]); }) : Object.getOwnPropertyDescriptors ? Object.defineProperties(target, Object.getOwnPropertyDescriptors(source)) : ownKeys(Object(source)).forEach(function (key) { Object.defineProperty(target, key, Object.getOwnPropertyDescriptor(source, key)); }); } return target; } /*
|
||||
Copyright 2022 The Matrix.org Foundation C.I.C.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
// /sync requests allow you to set a timeout= but the request may continue
|
||||
// beyond that and wedge forever, so we need to track how long we are willing
|
||||
// to keep open the connection. This constant is *ADDED* to the timeout= value
|
||||
// to determine the max time we're willing to wait.
|
||||
const BUFFER_PERIOD_MS = 10 * 1000;
|
||||
export const MSC3575_WILDCARD = "*";
|
||||
export const MSC3575_STATE_KEY_ME = "$ME";
|
||||
export const MSC3575_STATE_KEY_LAZY = "$LAZY";
|
||||
const MSC3575_WILDCARD = "*";
|
||||
exports.MSC3575_WILDCARD = MSC3575_WILDCARD;
|
||||
const MSC3575_STATE_KEY_ME = "$ME";
|
||||
exports.MSC3575_STATE_KEY_ME = MSC3575_STATE_KEY_ME;
|
||||
const MSC3575_STATE_KEY_LAZY = "$LAZY";
|
||||
|
||||
/**
|
||||
* Represents a subscription to a room or set of rooms. Controls which events are returned.
|
||||
@@ -43,44 +52,31 @@ export const MSC3575_STATE_KEY_LAZY = "$LAZY";
|
||||
* A complete Sliding Sync request.
|
||||
*/
|
||||
|
||||
/**
|
||||
* New format of hero introduced in MSC4186 with display name and avatar URL
|
||||
* in addition to just user_id (as it is on the wire, with underscores)
|
||||
* as opposed to Hero in room-summary.ts which has fields in camelCase
|
||||
* (and also a flag to note what format the hero came from).
|
||||
*/
|
||||
|
||||
/**
|
||||
* A complete Sliding Sync response
|
||||
*/
|
||||
|
||||
export let SlidingSyncState = /*#__PURE__*/function (SlidingSyncState) {
|
||||
/**
|
||||
* Fired by SlidingSyncEvent.Lifecycle event immediately before processing the response.
|
||||
*/
|
||||
exports.MSC3575_STATE_KEY_LAZY = MSC3575_STATE_KEY_LAZY;
|
||||
let SlidingSyncState = /*#__PURE__*/function (SlidingSyncState) {
|
||||
SlidingSyncState["RequestFinished"] = "FINISHED";
|
||||
/**
|
||||
* Fired by SlidingSyncEvent.Lifecycle event immediately after all room data listeners have been
|
||||
* invoked, but before list listeners.
|
||||
*/
|
||||
SlidingSyncState["Complete"] = "COMPLETE";
|
||||
return SlidingSyncState;
|
||||
}({});
|
||||
|
||||
/**
|
||||
* Internal Class. SlidingList represents a single list in sliding sync. The list can have filters,
|
||||
* multiple sliding windows, and maintains the index-\>room_id mapping.
|
||||
*/
|
||||
exports.SlidingSyncState = SlidingSyncState;
|
||||
class SlidingList {
|
||||
/**
|
||||
* Construct a new sliding list.
|
||||
* @param list - The range, sort and filter values to use for this list.
|
||||
*/
|
||||
constructor(list) {
|
||||
_defineProperty(this, "list", void 0);
|
||||
_defineProperty(this, "isModified", void 0);
|
||||
(0, _defineProperty2.default)(this, "list", void 0);
|
||||
(0, _defineProperty2.default)(this, "isModified", void 0);
|
||||
// returned data
|
||||
_defineProperty(this, "joinedCount", 0);
|
||||
(0, _defineProperty2.default)(this, "roomIndexToRoomId", {});
|
||||
(0, _defineProperty2.default)(this, "joinedCount", 0);
|
||||
this.replaceList(list);
|
||||
}
|
||||
|
||||
@@ -106,14 +102,17 @@ class SlidingList {
|
||||
* @param list - The new list parameters
|
||||
*/
|
||||
replaceList(list) {
|
||||
list.filters = list.filters ?? {};
|
||||
list.ranges = list.ranges ?? [];
|
||||
list.filters = list.filters || {};
|
||||
list.ranges = list.ranges || [];
|
||||
this.list = JSON.parse(JSON.stringify(list));
|
||||
this.isModified = true;
|
||||
|
||||
// reset values as the join count may be very different (if filters changed) including the rooms
|
||||
// (e.g. sort orders or sliding window ranges changed)
|
||||
|
||||
// the constantly changing sliding window ranges. Not an array for performance reasons
|
||||
// E.g. tracking ranges 0-99, 500-599, we don't want to have a 600 element array
|
||||
this.roomIndexToRoomId = {};
|
||||
// the total number of joined rooms according to the server, always >= len(roomIndexToRoomId)
|
||||
this.joinedCount = 0;
|
||||
}
|
||||
@@ -133,60 +132,66 @@ class SlidingList {
|
||||
}
|
||||
return list;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a given index is within the list range. This is required even though the /sync API
|
||||
* provides explicit updates with index positions because of the following situation:
|
||||
* 0 1 2 3 4 5 6 7 8 indexes
|
||||
* a b c d e f COMMANDS: SYNC 0 2 a b c; SYNC 6 8 d e f;
|
||||
* a b c d _ f COMMAND: DELETE 7;
|
||||
* e a b c d f COMMAND: INSERT 0 e;
|
||||
* c=3 is wrong as we are not tracking it, ergo we need to see if `i` is in range else drop it
|
||||
* @param i - The index to check
|
||||
* @returns True if the index is within a sliding window
|
||||
*/
|
||||
isIndexInRange(i) {
|
||||
for (const r of this.list.ranges) {
|
||||
if (r[0] <= i && i <= r[1]) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* When onResponse extensions should be invoked: before or after processing the main response.
|
||||
*/
|
||||
export let ExtensionState = /*#__PURE__*/function (ExtensionState) {
|
||||
// Call onResponse before processing the response body. This is useful when your extension is
|
||||
// preparing the ground for the response body e.g. processing to-device messages before the
|
||||
// encrypted event arrives.
|
||||
let ExtensionState = /*#__PURE__*/function (ExtensionState) {
|
||||
ExtensionState["PreProcess"] = "ExtState.PreProcess";
|
||||
// Call onResponse after processing the response body. This is useful when your extension is
|
||||
// decorating data from the client, and you rely on MatrixClient.getRoom returning the Room object
|
||||
// e.g. room account data.
|
||||
ExtensionState["PostProcess"] = "ExtState.PostProcess";
|
||||
return ExtensionState;
|
||||
}({});
|
||||
|
||||
/**
|
||||
* An interface that must be satisfied to register extensions
|
||||
*/
|
||||
|
||||
exports.ExtensionState = ExtensionState;
|
||||
/**
|
||||
* Events which can be fired by the SlidingSync class. These are designed to provide different levels
|
||||
* of information when processing sync responses.
|
||||
* - RoomData: concerns rooms, useful for SlidingSyncSdk to update its knowledge of rooms.
|
||||
* - Lifecycle: concerns callbacks at various well-defined points in the sync process.
|
||||
* - List: concerns lists, useful for UI layers to re-render room lists.
|
||||
* Specifically, the order of event invocation is:
|
||||
* - Lifecycle (state=RequestFinished)
|
||||
* - RoomData (N times)
|
||||
* - Lifecycle (state=Complete)
|
||||
* - List (at most once per list)
|
||||
*/
|
||||
export let SlidingSyncEvent = /*#__PURE__*/function (SlidingSyncEvent) {
|
||||
/**
|
||||
* This event fires when there are updates for a room. Fired as and when rooms are encountered
|
||||
* in the response.
|
||||
*/
|
||||
let SlidingSyncEvent = /*#__PURE__*/function (SlidingSyncEvent) {
|
||||
SlidingSyncEvent["RoomData"] = "SlidingSync.RoomData";
|
||||
/**
|
||||
* This event fires at various points in the /sync loop lifecycle.
|
||||
* - SlidingSyncState.RequestFinished: Fires after we receive a valid response but before the
|
||||
* response has been processed. Perform any pre-process steps here. If there was a problem syncing,
|
||||
* `err` will be set (e.g network errors).
|
||||
* - SlidingSyncState.Complete: Fires after the response has been processed.
|
||||
*/
|
||||
SlidingSyncEvent["Lifecycle"] = "SlidingSync.Lifecycle";
|
||||
SlidingSyncEvent["List"] = "SlidingSync.List";
|
||||
return SlidingSyncEvent;
|
||||
}({});
|
||||
exports.SlidingSyncEvent = SlidingSyncEvent;
|
||||
/**
|
||||
* SlidingSync is a high-level data structure which controls the majority of sliding sync.
|
||||
* It has no hooks into JS SDK except for needing a MatrixClient to perform the HTTP request.
|
||||
* This means this class (and everything it uses) can be used in isolation from JS SDK if needed.
|
||||
* To hook this up with the JS SDK, you need to use SlidingSyncSdk.
|
||||
*/
|
||||
export class SlidingSync extends TypedEventEmitter {
|
||||
class SlidingSync extends _typedEventEmitter.TypedEventEmitter {
|
||||
/**
|
||||
* Create a new sliding sync instance
|
||||
* @param proxyBaseUrl - The base URL of the sliding sync proxy
|
||||
@@ -197,26 +202,31 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
*/
|
||||
constructor(proxyBaseUrl, lists, roomSubscriptionInfo, client, timeoutMS) {
|
||||
super();
|
||||
_defineProperty(this, "lists", void 0);
|
||||
_defineProperty(this, "listModifiedCount", 0);
|
||||
_defineProperty(this, "terminated", false);
|
||||
// flag set when resend() is called because we cannot rely on detecting AbortError in JS SDK :(
|
||||
_defineProperty(this, "needsResend", false);
|
||||
// map of extension name to req/resp handler
|
||||
_defineProperty(this, "extensions", {});
|
||||
_defineProperty(this, "desiredRoomSubscriptions", new Set());
|
||||
// the *desired* room subscriptions
|
||||
_defineProperty(this, "confirmedRoomSubscriptions", new Set());
|
||||
// map of custom subscription name to the subscription
|
||||
_defineProperty(this, "customSubscriptions", new Map());
|
||||
// map of room ID to custom subscription name
|
||||
_defineProperty(this, "roomIdToCustomSubscription", new Map());
|
||||
_defineProperty(this, "pendingReq", void 0);
|
||||
_defineProperty(this, "abortController", void 0);
|
||||
this.proxyBaseUrl = proxyBaseUrl;
|
||||
this.roomSubscriptionInfo = roomSubscriptionInfo;
|
||||
this.client = client;
|
||||
this.timeoutMS = timeoutMS;
|
||||
(0, _defineProperty2.default)(this, "lists", void 0);
|
||||
(0, _defineProperty2.default)(this, "listModifiedCount", 0);
|
||||
(0, _defineProperty2.default)(this, "terminated", false);
|
||||
// flag set when resend() is called because we cannot rely on detecting AbortError in JS SDK :(
|
||||
(0, _defineProperty2.default)(this, "needsResend", false);
|
||||
// the txn_id to send with the next request.
|
||||
(0, _defineProperty2.default)(this, "txnId", null);
|
||||
// a list (in chronological order of when they were sent) of objects containing the txn ID and
|
||||
// a defer to resolve/reject depending on whether they were successfully sent or not.
|
||||
(0, _defineProperty2.default)(this, "txnIdDefers", []);
|
||||
// map of extension name to req/resp handler
|
||||
(0, _defineProperty2.default)(this, "extensions", {});
|
||||
(0, _defineProperty2.default)(this, "desiredRoomSubscriptions", new Set());
|
||||
// the *desired* room subscriptions
|
||||
(0, _defineProperty2.default)(this, "confirmedRoomSubscriptions", new Set());
|
||||
// map of custom subscription name to the subscription
|
||||
(0, _defineProperty2.default)(this, "customSubscriptions", new Map());
|
||||
// map of room ID to custom subscription name
|
||||
(0, _defineProperty2.default)(this, "roomIdToCustomSubscription", new Map());
|
||||
(0, _defineProperty2.default)(this, "pendingReq", void 0);
|
||||
(0, _defineProperty2.default)(this, "abortController", void 0);
|
||||
this.lists = new Map();
|
||||
lists.forEach((list, key) => {
|
||||
this.lists.set(key, new SlidingList(list));
|
||||
@@ -232,7 +242,7 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
*/
|
||||
addCustomSubscription(name, sub) {
|
||||
if (this.customSubscriptions.has(name)) {
|
||||
logger.warn(`addCustomSubscription: ${name} already exists as a custom subscription, ignoring.`);
|
||||
_logger.logger.warn(`addCustomSubscription: ${name} already exists as a custom subscription, ignoring.`);
|
||||
return;
|
||||
}
|
||||
this.customSubscriptions.set(name, sub);
|
||||
@@ -267,7 +277,8 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
joinedCount: data.joinedCount
|
||||
joinedCount: data.joinedCount,
|
||||
roomIndexToRoomId: Object.assign({}, data.roomIndexToRoomId)
|
||||
};
|
||||
}
|
||||
|
||||
@@ -298,10 +309,10 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
setListRanges(key, ranges) {
|
||||
const list = this.lists.get(key);
|
||||
if (!list) {
|
||||
throw new Error("no list with key " + key);
|
||||
return Promise.reject(new Error("no list with key " + key));
|
||||
}
|
||||
list.updateListRange(ranges);
|
||||
this.resend();
|
||||
return this.resend();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -322,7 +333,7 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
this.lists.set(key, new SlidingList(list));
|
||||
}
|
||||
this.listModifiedCount += 1;
|
||||
this.resend();
|
||||
return this.resend();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -338,21 +349,27 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
* /sync request to resend new subscriptions. If the /sync stream has not started, this will
|
||||
* prepare the room subscriptions for when start() is called.
|
||||
* @param s - The new desired room subscriptions.
|
||||
* @returns A promise which resolves to the transaction ID when it has been received down sync
|
||||
* (or rejects with the transaction ID if the action was not applied e.g the request was cancelled
|
||||
* immediately after sending, in which case the action will be applied in the subsequent request)
|
||||
*/
|
||||
modifyRoomSubscriptions(s) {
|
||||
this.desiredRoomSubscriptions = s;
|
||||
this.resend();
|
||||
return this.resend();
|
||||
}
|
||||
|
||||
/**
|
||||
* Modify which events to retrieve for room subscriptions. Invalidates all room subscriptions
|
||||
* such that they will be sent up afresh.
|
||||
* @param rs - The new room subscription fields to fetch.
|
||||
* @returns A promise which resolves to the transaction ID when it has been received down sync
|
||||
* (or rejects with the transaction ID if the action was not applied e.g the request was cancelled
|
||||
* immediately after sending, in which case the action will be applied in the subsequent request)
|
||||
*/
|
||||
modifyRoomSubscriptionInfo(rs) {
|
||||
this.roomSubscriptionInfo = rs;
|
||||
this.confirmedRoomSubscriptions = new Set();
|
||||
this.resend();
|
||||
return this.resend();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -365,27 +382,26 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
}
|
||||
this.extensions[ext.name()] = ext;
|
||||
}
|
||||
async getExtensionRequest(isInitial) {
|
||||
getExtensionRequest(isInitial) {
|
||||
const ext = {};
|
||||
// oxlint-disable-next-line guard-for-in
|
||||
for (const extName in this.extensions) {
|
||||
ext[extName] = await this.extensions[extName].onRequest(isInitial);
|
||||
}
|
||||
Object.keys(this.extensions).forEach(extName => {
|
||||
ext[extName] = this.extensions[extName].onRequest(isInitial);
|
||||
});
|
||||
return ext;
|
||||
}
|
||||
async onPreExtensionsResponse(ext) {
|
||||
await Promise.all(Object.keys(ext).map(async extName => {
|
||||
onPreExtensionsResponse(ext) {
|
||||
Object.keys(ext).forEach(extName => {
|
||||
if (this.extensions[extName].when() == ExtensionState.PreProcess) {
|
||||
await this.extensions[extName].onResponse(ext[extName]);
|
||||
this.extensions[extName].onResponse(ext[extName]);
|
||||
}
|
||||
}));
|
||||
});
|
||||
}
|
||||
async onPostExtensionsResponse(ext) {
|
||||
await Promise.all(Object.keys(ext).map(async extName => {
|
||||
onPostExtensionsResponse(ext) {
|
||||
Object.keys(ext).forEach(extName => {
|
||||
if (this.extensions[extName].when() == ExtensionState.PostProcess) {
|
||||
await this.extensions[extName].onResponse(ext[extName]);
|
||||
this.extensions[extName].onResponse(ext[extName]);
|
||||
}
|
||||
}));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -412,24 +428,215 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
invokeLifecycleListeners(state, resp, err) {
|
||||
this.emit(SlidingSyncEvent.Lifecycle, state, resp, err);
|
||||
}
|
||||
shiftRight(listKey, hi, low) {
|
||||
const list = this.lists.get(listKey);
|
||||
if (!list) {
|
||||
return;
|
||||
}
|
||||
// l h
|
||||
// 0,1,2,3,4 <- before
|
||||
// 0,1,2,2,3 <- after, hi is deleted and low is duplicated
|
||||
for (let i = hi; i > low; i--) {
|
||||
if (list.isIndexInRange(i)) {
|
||||
list.roomIndexToRoomId[i] = list.roomIndexToRoomId[i - 1];
|
||||
}
|
||||
}
|
||||
}
|
||||
shiftLeft(listKey, hi, low) {
|
||||
const list = this.lists.get(listKey);
|
||||
if (!list) {
|
||||
return;
|
||||
}
|
||||
// l h
|
||||
// 0,1,2,3,4 <- before
|
||||
// 0,1,3,4,4 <- after, low is deleted and hi is duplicated
|
||||
for (let i = low; i < hi; i++) {
|
||||
if (list.isIndexInRange(i)) {
|
||||
list.roomIndexToRoomId[i] = list.roomIndexToRoomId[i + 1];
|
||||
}
|
||||
}
|
||||
}
|
||||
removeEntry(listKey, index) {
|
||||
const list = this.lists.get(listKey);
|
||||
if (!list) {
|
||||
return;
|
||||
}
|
||||
// work out the max index
|
||||
let max = -1;
|
||||
for (const n in list.roomIndexToRoomId) {
|
||||
if (Number(n) > max) {
|
||||
max = Number(n);
|
||||
}
|
||||
}
|
||||
if (max < 0 || index > max) {
|
||||
return;
|
||||
}
|
||||
// Everything higher than the gap needs to be shifted left.
|
||||
this.shiftLeft(listKey, max, index);
|
||||
delete list.roomIndexToRoomId[max];
|
||||
}
|
||||
addEntry(listKey, index) {
|
||||
const list = this.lists.get(listKey);
|
||||
if (!list) {
|
||||
return;
|
||||
}
|
||||
// work out the max index
|
||||
let max = -1;
|
||||
for (const n in list.roomIndexToRoomId) {
|
||||
if (Number(n) > max) {
|
||||
max = Number(n);
|
||||
}
|
||||
}
|
||||
if (max < 0 || index > max) {
|
||||
return;
|
||||
}
|
||||
// Everything higher than the gap needs to be shifted right, +1 so we don't delete the highest element
|
||||
this.shiftRight(listKey, max + 1, index);
|
||||
}
|
||||
processListOps(list, listKey) {
|
||||
let gapIndex = -1;
|
||||
const listData = this.lists.get(listKey);
|
||||
if (!listData) {
|
||||
return;
|
||||
}
|
||||
list.ops.forEach(op => {
|
||||
if (!listData) {
|
||||
return;
|
||||
}
|
||||
switch (op.op) {
|
||||
case "DELETE":
|
||||
{
|
||||
_logger.logger.debug("DELETE", listKey, op.index, ";");
|
||||
delete listData.roomIndexToRoomId[op.index];
|
||||
if (gapIndex !== -1) {
|
||||
// we already have a DELETE operation to process, so process it.
|
||||
this.removeEntry(listKey, gapIndex);
|
||||
}
|
||||
gapIndex = op.index;
|
||||
break;
|
||||
}
|
||||
case "INSERT":
|
||||
{
|
||||
_logger.logger.debug("INSERT", listKey, op.index, op.room_id, ";");
|
||||
if (listData.roomIndexToRoomId[op.index]) {
|
||||
// something is in this space, shift items out of the way
|
||||
if (gapIndex < 0) {
|
||||
// we haven't been told where to shift from, so make way for a new room entry.
|
||||
this.addEntry(listKey, op.index);
|
||||
} else if (gapIndex > op.index) {
|
||||
// the gap is further down the list, shift every element to the right
|
||||
// starting at the gap so we can just shift each element in turn:
|
||||
// [A,B,C,_] gapIndex=3, op.index=0
|
||||
// [A,B,C,C] i=3
|
||||
// [A,B,B,C] i=2
|
||||
// [A,A,B,C] i=1
|
||||
// Terminate. We'll assign into op.index next.
|
||||
this.shiftRight(listKey, gapIndex, op.index);
|
||||
} else if (gapIndex < op.index) {
|
||||
// the gap is further up the list, shift every element to the left
|
||||
// starting at the gap so we can just shift each element in turn
|
||||
this.shiftLeft(listKey, op.index, gapIndex);
|
||||
}
|
||||
}
|
||||
// forget the gap, we don't need it anymore. This is outside the check for
|
||||
// a room being present in this index position because INSERTs always universally
|
||||
// forget the gap, not conditionally based on the presence of a room in the INSERT
|
||||
// position. Without this, DELETE 0; INSERT 0; would do the wrong thing.
|
||||
gapIndex = -1;
|
||||
listData.roomIndexToRoomId[op.index] = op.room_id;
|
||||
break;
|
||||
}
|
||||
case "INVALIDATE":
|
||||
{
|
||||
const startIndex = op.range[0];
|
||||
for (let i = startIndex; i <= op.range[1]; i++) {
|
||||
delete listData.roomIndexToRoomId[i];
|
||||
}
|
||||
_logger.logger.debug("INVALIDATE", listKey, op.range[0], op.range[1], ";");
|
||||
break;
|
||||
}
|
||||
case "SYNC":
|
||||
{
|
||||
const startIndex = op.range[0];
|
||||
for (let i = startIndex; i <= op.range[1]; i++) {
|
||||
const roomId = op.room_ids[i - startIndex];
|
||||
if (!roomId) {
|
||||
break; // we are at the end of list
|
||||
}
|
||||
|
||||
listData.roomIndexToRoomId[i] = roomId;
|
||||
}
|
||||
_logger.logger.debug("SYNC", listKey, op.range[0], op.range[1], (op.room_ids || []).join(" "), ";");
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
if (gapIndex !== -1) {
|
||||
// we already have a DELETE operation to process, so process it
|
||||
// Everything higher than the gap needs to be shifted left.
|
||||
this.removeEntry(listKey, gapIndex);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resend a Sliding Sync request. Used when something has changed in the request.
|
||||
* Resend a Sliding Sync request. Used when something has changed in the request. Resolves with
|
||||
* the transaction ID of this request on success. Rejects with the transaction ID of this request
|
||||
* on failure.
|
||||
*/
|
||||
resend() {
|
||||
var _this$abortController;
|
||||
if (this.needsResend && this.txnIdDefers.length > 0) {
|
||||
// we already have a resend queued, so just return the same promise
|
||||
return this.txnIdDefers[this.txnIdDefers.length - 1].promise;
|
||||
}
|
||||
this.needsResend = true;
|
||||
this.abortController?.abort();
|
||||
this.txnId = this.client.makeTxnId();
|
||||
const d = (0, _utils.defer)();
|
||||
this.txnIdDefers.push(_objectSpread(_objectSpread({}, d), {}, {
|
||||
txnId: this.txnId
|
||||
}));
|
||||
(_this$abortController = this.abortController) === null || _this$abortController === void 0 ? void 0 : _this$abortController.abort();
|
||||
this.abortController = new AbortController();
|
||||
return d.promise;
|
||||
}
|
||||
resolveTransactionDefers(txnId) {
|
||||
if (!txnId) {
|
||||
return;
|
||||
}
|
||||
// find the matching index
|
||||
let txnIndex = -1;
|
||||
for (let i = 0; i < this.txnIdDefers.length; i++) {
|
||||
if (this.txnIdDefers[i].txnId === txnId) {
|
||||
txnIndex = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (txnIndex === -1) {
|
||||
// this shouldn't happen; we shouldn't be seeing txn_ids for things we don't know about,
|
||||
// whine about it.
|
||||
_logger.logger.warn(`resolveTransactionDefers: seen ${txnId} but it isn't a pending txn, ignoring.`);
|
||||
return;
|
||||
}
|
||||
// This list is sorted in time, so if the input txnId ACKs in the middle of this array,
|
||||
// then everything before it that hasn't been ACKed yet never will and we should reject them.
|
||||
for (let i = 0; i < txnIndex; i++) {
|
||||
this.txnIdDefers[i].reject(this.txnIdDefers[i].txnId);
|
||||
}
|
||||
this.txnIdDefers[txnIndex].resolve(txnId);
|
||||
// clear out settled promises, including the one we resolved.
|
||||
this.txnIdDefers = this.txnIdDefers.slice(txnIndex + 1);
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop syncing with the server.
|
||||
*/
|
||||
stop() {
|
||||
var _this$abortController2;
|
||||
this.terminated = true;
|
||||
this.abortController?.abort();
|
||||
(_this$abortController2 = this.abortController) === null || _this$abortController2 === void 0 ? void 0 : _this$abortController2.abort();
|
||||
// remove all listeners so things can be GC'd
|
||||
this.removeAllListeners(SlidingSyncEvent.Lifecycle);
|
||||
this.removeAllListeners(SlidingSyncEvent.List);
|
||||
this.removeAllListeners(SlidingSyncEvent.RoomData);
|
||||
}
|
||||
|
||||
@@ -437,14 +644,22 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
* Re-setup this connection e.g in the event of an expired session.
|
||||
*/
|
||||
resetup() {
|
||||
logger.warn("SlidingSync: resetting connection info");
|
||||
var _this$abortController3;
|
||||
_logger.logger.warn("SlidingSync: resetting connection info");
|
||||
// any pending txn ID defers will be forgotten already by the server, so clear them out
|
||||
this.txnIdDefers.forEach(d => {
|
||||
d.reject(d.txnId);
|
||||
});
|
||||
this.txnIdDefers = [];
|
||||
// resend sticky params and de-confirm all subscriptions
|
||||
this.lists.forEach(l => {
|
||||
l.setModified(true);
|
||||
});
|
||||
this.confirmedRoomSubscriptions = new Set(); // leave desired ones alone though!
|
||||
// reset the connection as we might be wedged
|
||||
this.resend();
|
||||
this.needsResend = true;
|
||||
(_this$abortController3 = this.abortController) === null || _this$abortController3 === void 0 ? void 0 : _this$abortController3.abort();
|
||||
this.abortController = new AbortController();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -455,18 +670,20 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
let currentPos;
|
||||
while (!this.terminated) {
|
||||
this.needsResend = false;
|
||||
let doNotUpdateList = false;
|
||||
let resp;
|
||||
try {
|
||||
const listModifiedCount = this.listModifiedCount;
|
||||
const reqLists = {};
|
||||
this.lists.forEach((l, key) => {
|
||||
reqLists[key] = l.getList(true);
|
||||
reqLists[key] = l.getList(false);
|
||||
});
|
||||
const reqBody = {
|
||||
lists: reqLists,
|
||||
pos: currentPos,
|
||||
timeout: this.timeoutMS,
|
||||
clientTimeout: this.timeoutMS + BUFFER_PERIOD_MS,
|
||||
extensions: await this.getExtensionRequest(currentPos === undefined)
|
||||
extensions: this.getExtensionRequest(currentPos === undefined)
|
||||
};
|
||||
// check if we are (un)subscribing to a room and modify request this one time for it
|
||||
const newSubscriptions = difference(this.desiredRoomSubscriptions, this.confirmedRoomSubscriptions);
|
||||
@@ -485,6 +702,10 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
reqBody.room_subscriptions[roomId] = sub;
|
||||
}
|
||||
}
|
||||
if (this.txnId) {
|
||||
reqBody.txn_id = this.txnId;
|
||||
this.txnId = null;
|
||||
}
|
||||
this.pendingReq = this.client.slidingSync(reqBody, this.proxyBaseUrl, this.abortController.signal);
|
||||
resp = await this.pendingReq;
|
||||
currentPos = resp.pos;
|
||||
@@ -495,14 +716,21 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
for (const roomId of unsubscriptions) {
|
||||
this.confirmedRoomSubscriptions.delete(roomId);
|
||||
}
|
||||
if (listModifiedCount !== this.listModifiedCount) {
|
||||
// the lists have been modified whilst we were waiting for 'await' to return, but the abort()
|
||||
// call did nothing. It is NOT SAFE to modify the list array now. We'll process the response but
|
||||
// not update list pointers.
|
||||
_logger.logger.debug("list modified during await call, not updating list");
|
||||
doNotUpdateList = true;
|
||||
}
|
||||
// mark all these lists as having been sent as sticky so we don't keep sending sticky params
|
||||
this.lists.forEach(l => {
|
||||
l.setModified(false);
|
||||
});
|
||||
// set default empty values so we don't need to null check
|
||||
resp.lists = resp.lists ?? {};
|
||||
resp.rooms = resp.rooms ?? {};
|
||||
resp.extensions = resp.extensions ?? {};
|
||||
resp.lists = resp.lists || {};
|
||||
resp.rooms = resp.rooms || {};
|
||||
resp.extensions = resp.extensions || {};
|
||||
Object.keys(resp.lists).forEach(key => {
|
||||
const list = this.lists.get(key);
|
||||
if (!list || !resp) {
|
||||
@@ -519,29 +747,47 @@ export class SlidingSync extends TypedEventEmitter {
|
||||
// so drop state and re-request
|
||||
this.resetup();
|
||||
currentPos = undefined;
|
||||
await sleep(50); // in case the 400 was for something else; don't tightloop
|
||||
await (0, _utils.sleep)(50); // in case the 400 was for something else; don't tightloop
|
||||
continue;
|
||||
} // else fallthrough to generic error handling
|
||||
} else if (this.needsResend || err.name === "AbortError") {
|
||||
continue; // don't sleep as we caused this error by abort()ing the request.
|
||||
}
|
||||
logger.error(err);
|
||||
await sleep(5000);
|
||||
|
||||
_logger.logger.error(err);
|
||||
await (0, _utils.sleep)(5000);
|
||||
}
|
||||
if (!resp) {
|
||||
continue;
|
||||
}
|
||||
await this.onPreExtensionsResponse(resp.extensions);
|
||||
|
||||
// oxlint-disable-next-line guard-for-in
|
||||
this.onPreExtensionsResponse(resp.extensions);
|
||||
for (const roomId in resp.rooms) {
|
||||
await this.invokeRoomDataListeners(roomId, resp.rooms[roomId]);
|
||||
}
|
||||
const listKeysWithUpdates = new Set();
|
||||
if (!doNotUpdateList) {
|
||||
for (const [key, list] of Object.entries(resp.lists)) {
|
||||
list.ops = list.ops || [];
|
||||
if (list.ops.length > 0) {
|
||||
listKeysWithUpdates.add(key);
|
||||
}
|
||||
this.processListOps(list, key);
|
||||
}
|
||||
}
|
||||
this.invokeLifecycleListeners(SlidingSyncState.Complete, resp);
|
||||
await this.onPostExtensionsResponse(resp.extensions);
|
||||
this.onPostExtensionsResponse(resp.extensions);
|
||||
listKeysWithUpdates.forEach(listKey => {
|
||||
const list = this.lists.get(listKey);
|
||||
if (!list) {
|
||||
return;
|
||||
}
|
||||
this.emit(SlidingSyncEvent.List, listKey, list.joinedCount, Object.assign({}, list.roomIndexToRoomId));
|
||||
});
|
||||
this.resolveTransactionDefers(resp.txn_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
exports.SlidingSync = SlidingSync;
|
||||
const difference = (setA, setB) => {
|
||||
const diff = new Set(setA);
|
||||
for (const elem of setB) {
|
||||
|
||||
Reference in New Issue
Block a user