vector-im-hydrogen-web/src/matrix/sync.js

140 lines
4.8 KiB
JavaScript
Raw Normal View History

import {RequestAbortError} from "./error.js";
2019-02-20 23:48:16 +01:00
import EventEmitter from "../EventEmitter.js";
2018-12-21 14:35:24 +01:00
2019-02-10 21:25:29 +01:00
const INCREMENTAL_TIMEOUT = 30000;
const SYNC_EVENT_LIMIT = 10;
2018-12-21 14:35:24 +01:00
2019-02-10 21:25:29 +01:00
function parseRooms(roomsSection, roomCallback) {
if (roomsSection) {
const allMemberships = ["join", "invite", "leave"];
for(const membership of allMemberships) {
const membershipSection = roomsSection[membership];
if (membershipSection) {
return Object.entries(membershipSection).map(([roomId, roomResponse]) => {
return roomCallback(roomId, roomResponse, membership);
});
2019-05-12 20:26:46 +02:00
}
}
}
return [];
2019-02-04 23:26:45 +01:00
}
2019-02-10 21:25:29 +01:00
export default class Sync extends EventEmitter {
constructor({hsApi, session, storage}) {
2019-05-12 20:26:46 +02:00
super();
this._hsApi = hsApi;
this._session = session;
this._storage = storage;
this._isSyncing = false;
this._currentRequest = null;
}
get isSyncing() {
return this._isSyncing;
}
2019-05-12 20:26:46 +02:00
// returns when initial sync is done
async start() {
if (this._isSyncing) {
return;
}
this._isSyncing = true;
this.emit("status", "started");
2019-05-12 20:26:46 +02:00
let syncToken = this._session.syncToken;
// do initial sync if needed
if (!syncToken) {
// need to create limit filter here
syncToken = await this._syncRequest();
}
this._syncLoop(syncToken);
}
2018-12-21 14:35:24 +01:00
2019-05-12 20:26:46 +02:00
async _syncLoop(syncToken) {
// if syncToken is falsy, it will first do an initial sync ...
while(this._isSyncing) {
try {
console.log(`starting sync request with since ${syncToken} ...`);
syncToken = await this._syncRequest(syncToken, INCREMENTAL_TIMEOUT);
} catch (err) {
this._isSyncing = false;
if (!(err instanceof RequestAbortError)) {
2019-10-12 22:18:19 +02:00
console.error("stopping sync because of error");
console.error(err);
this.emit("status", "error", err);
2019-05-12 20:26:46 +02:00
}
}
}
this.emit("status", "stopped");
2019-05-12 20:26:46 +02:00
}
2019-02-03 22:17:24 +01:00
2019-05-12 20:26:46 +02:00
async _syncRequest(syncToken, timeout) {
2019-10-12 20:24:09 +02:00
let {syncFilterId} = this._session;
if (typeof syncFilterId !== "string") {
syncFilterId = (await this._hsApi.createFilter(this._session.user.id, {room: {state: {lazy_load_members: true}}}).response()).filter_id;
}
this._currentRequest = this._hsApi.sync(syncToken, syncFilterId, timeout);
2019-05-12 20:26:46 +02:00
const response = await this._currentRequest.response();
syncToken = response.next_batch;
const storeNames = this._storage.storeNames;
const syncTxn = await this._storage.readWriteTxn([
storeNames.session,
storeNames.roomSummary,
storeNames.roomState,
2019-05-12 20:26:46 +02:00
storeNames.timelineEvents,
2019-05-19 20:49:46 +02:00
storeNames.timelineFragments,
storeNames.pendingEvents,
2019-05-12 20:26:46 +02:00
]);
const roomChanges = [];
let sessionChanges;
try {
sessionChanges = this._session.writeSync(syncToken, syncFilterId, response.account_data, syncTxn);
// to_device
// presence
2019-05-12 20:26:46 +02:00
if (response.rooms) {
const promises = parseRooms(response.rooms, async (roomId, roomResponse, membership) => {
2019-05-12 20:26:46 +02:00
let room = this._session.rooms.get(roomId);
if (!room) {
room = this._session.createRoom(roomId);
}
console.log(` * applying sync response to room ${roomId} ...`);
const changes = await room.persistSync(roomResponse, membership, syncTxn);
roomChanges.push({room, changes});
2019-05-12 20:26:46 +02:00
});
await Promise.all(promises);
2019-05-12 20:26:46 +02:00
}
} catch(err) {
2019-06-02 14:59:30 +02:00
console.warn("aborting syncTxn because of error");
2019-05-12 20:26:46 +02:00
// avoid corrupting state by only
// storing the sync up till the point
// the exception occurred
syncTxn.abort();
throw err;
}
try {
await syncTxn.complete();
console.info("syncTxn committed!!");
} catch (err) {
2019-10-12 22:18:19 +02:00
console.error("unable to commit sync tranaction");
throw err;
2019-05-12 20:26:46 +02:00
}
this._session.afterSync(sessionChanges);
// emit room related events after txn has been closed
for(let {room, changes} of roomChanges) {
room.emitSync(changes);
}
2019-05-12 20:26:46 +02:00
return syncToken;
}
2018-12-21 14:35:24 +01:00
2019-05-12 20:26:46 +02:00
stop() {
if (!this._isSyncing) {
return;
}
this._isSyncing = false;
if (this._currentRequest) {
this._currentRequest.abort();
this._currentRequest = null;
}
}
2019-02-20 23:48:16 +01:00
}