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

118 lines
3.0 KiB
JavaScript
Raw Normal View History

2019-02-10 21:25:29 +01:00
import {
RequestAbortError,
HomeServerError,
StorageError
} from "./error.js";
import EventEmitter from "../event-emitter.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) {
return;
}
const allMemberships = ["join", "invite", "leave"];
for(const membership of allMemberships) {
const membershipSection = roomsSection[membership];
if (membershipSection) {
const rooms = Object.entries(membershipSection)
for (const [roomId, roomResponse] of rooms) {
roomCallback(roomId, roomResponse, membership);
}
}
}
2019-02-04 23:26:45 +01:00
}
2019-02-10 21:25:29 +01:00
export default class Sync extends EventEmitter {
2019-02-07 01:25:12 +01:00
constructor(hsApi, session, storage) {
2019-02-10 21:25:29 +01:00
super();
2019-02-07 01:25:12 +01:00
this._hsApi = hsApi;
2018-12-21 14:35:24 +01:00
this._session = session;
2019-02-03 22:17:24 +01:00
this._storage = storage;
2018-12-21 14:35:24 +01:00
this._isSyncing = false;
this._currentRequest = null;
}
2019-02-03 22:17:24 +01:00
// returns when initial sync is done
async start() {
2018-12-21 14:35:24 +01:00
if (this._isSyncing) {
return;
}
this._isSyncing = true;
2019-02-10 21:25:29 +01:00
let syncToken = this._session.syncToken;
// do initial sync if needed
if (!syncToken) {
2019-02-10 21:25:29 +01:00
// need to create limit filter here
syncToken = await this._syncRequest();
2018-12-21 14:35:24 +01:00
}
2019-02-03 22:17:24 +01:00
this._syncLoop(syncToken);
2018-12-21 14:35:24 +01:00
}
async _syncLoop(syncToken) {
2019-02-03 22:17:24 +01:00
// if syncToken is falsy, it will first do an initial sync ...
2018-12-21 14:35:24 +01:00
while(this._isSyncing) {
2019-02-03 22:17:24 +01:00
try {
2019-02-10 21:25:29 +01:00
console.log(`starting sync request with since ${syncToken} ...`);
syncToken = await this._syncRequest(syncToken, INCREMENTAL_TIMEOUT);
2019-02-03 22:17:24 +01:00
} catch (err) {
2019-02-10 21:25:29 +01:00
console.warn("stopping sync because of error");
this._isSyncing = false;
2019-02-03 22:17:24 +01:00
this.emit("error", err);
}
}
2019-02-10 21:25:29 +01:00
this.emit("stopped");
2019-02-03 22:17:24 +01:00
}
2019-02-10 21:25:29 +01:00
async _syncRequest(syncToken, timeout) {
this._currentRequest = this._hsApi.sync(syncToken, undefined, timeout);
const response = await this._currentRequest.response();
2019-02-03 22:17:24 +01:00
syncToken = response.next_batch;
2019-02-05 00:21:50 +01:00
const storeNames = this._storage.storeNames;
2019-02-10 21:25:29 +01:00
const syncTxn = this._storage.readWriteTxn([
storeNames.session,
2019-02-10 21:25:29 +01:00
storeNames.roomSummary,
storeNames.roomTimeline,
storeNames.roomState,
2019-02-05 00:21:50 +01:00
]);
2019-02-03 22:17:24 +01:00
try {
2019-02-10 21:25:29 +01:00
this._session.applySync(syncToken, response.account_data, syncTxn);
2018-12-21 14:35:24 +01:00
// to_device
// presence
2019-02-10 21:25:29 +01:00
if (response.rooms) {
parseRooms(response.rooms, (roomId, roomResponse, membership) => {
let room = this._session.getRoom(roomId);
if (!room) {
room = this._session.createRoom(roomId);
}
console.log(` * applying sync response to room ${roomId} ...`);
room.applySync(roomResponse, membership, syncTxn);
});
}
2019-02-03 22:17:24 +01:00
} catch(err) {
2019-02-10 21:25:29 +01:00
console.warn("aborting syncTxn because of error");
2019-02-03 22:17:24 +01:00
// avoid corrupting state by only
// storing the sync up till the point
// the exception occurred
2019-02-10 21:25:29 +01:00
syncTxn.abort();
2019-02-03 22:17:24 +01:00
throw err;
}
try {
2019-02-10 21:25:29 +01:00
await syncTxn.complete();
console.info("syncTxn committed!!");
2019-02-03 22:17:24 +01:00
} catch (err) {
throw new StorageError("unable to commit sync tranaction", err);
2018-12-21 14:35:24 +01:00
}
2019-02-03 22:17:24 +01:00
return syncToken;
2018-12-21 14:35:24 +01:00
}
stop() {
if (!this._isSyncing) {
return;
}
this._isSyncing = false;
if (this._currentRequest) {
this._currentRequest.abort();
this._currentRequest = null;
}
}
}