Dima/matrix-js-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Process join rooms and add local timeouts to /sync

Browse Source 
This actually works now, though there's a number of teething
issues which may be app-specific. That, and all the tests are
broken.
This commit is contained in:
Kegan Dougal
2015-12-10 14:14:56 +00:00
parent 4c7e6807d2
commit 6af2197183
2 changed files with 88 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/client.js
+2 -4
View File
@@ -138,9 +138,6 @@ function MatrixClient(opts) {
userId: (opts.userId || null)
};
this._http = new httpApi.MatrixHttpApi(httpOpts);
this._syncingRooms = {
// room_id: Promise
};
this.callList = {
// callId: MatrixCall
};
@@ -620,7 +617,8 @@ MatrixClient.prototype.joinRoom = function(roomIdOrAlias, opts, callback) {
var syncApi = new SyncApi(self);
var room = syncApi.createRoom(roomId);
if (opts.syncRoom) {
return syncApi.syncRoom(room);
// v2 will do this for us
// return syncApi.syncRoom(room);
}
return q(room);
}, function(err) {
lib/sync.js
+86 -265
View File
@@ -66,34 +66,6 @@ SyncApi.prototype.createRoom = function(roomId) {
return room;
};
/**
* @param {Room} room
* @return {Promise}
*/
SyncApi.prototype.syncRoom = function(room) {
var client = this.client;
var self = this;
if (client._syncingRooms[room.roomId]) {
return client._syncingRooms[room.roomId];
}
var defer = q.defer();
client._syncingRooms[room.roomId] = defer.promise;
client.roomInitialSync(room.roomId, this.opts.initialSyncLimit).done(
function(res) {
room.timeline = []; // blow away any previous messages.
self._processRoomEvents(room, res.state, res.messages);
room.recalculate(client.credentials.userId);
client.store.storeRoom(room);
client.emit("Room", room);
defer.resolve(room);
client._syncingRooms[room.roomId] = undefined;
}, function(err) {
defer.reject(err);
client._syncingRooms[room.roomId] = undefined;
});
return defer.promise;
};
/**
* Main entry point
*/
@@ -182,13 +154,41 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
since: client.store.getSyncToken() || undefined // do not send 'null'
};
if (attempt > 1) {
// we think the connection is dead. If it comes back up, we won't know
// about it till /sync returns. If the timeout= is high, this could
// be a long time. Set it to 1 when doing retries.
qps.timeout = 1;
}
if (client._guestRooms && client._isGuest) {
qps.room_id = JSON.stringify(client._guestRooms);
}
// Set up local timer and error handler for retries.
function errHandler(err) {
console.error("/sync error (%s attempts): %s", attempt, err);
attempt += 1;
startSyncingRetryTimer(client, attempt, function() {
self._sync(syncOptions, attempt);
});
updateSyncState(client, "ERROR", { error: err });
}
var discardResult = false;
var timeoutObj = setTimeout(function() {
discardResult = true;
errHandler("Locally timed out waiting for a response");
}, qps.timeout + (20 * 1000)); // 20s buffer
client._http.authedRequestWithPrefix(
undefined, "GET", "/sync", qps, undefined, httpApi.PREFIX_V2_ALPHA
).done(function(data) {
if (discardResult) {
return;
}
else {
clearTimeout(timeoutObj);
}
// data looks like:
// {
// next_batch: $token,
@@ -220,6 +220,10 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
// barfs on an event we can skip it rather than constantly polling with the same token.
client.store.setSyncToken(data.next_batch);
// TODO-arch:
// - Each event we pass through needs to be emitted via 'event', can we do this in one place?
// - The isBrandNewRoom boilerplate is boilerplatey.
try {
// handle presence events (User objects)
if (data.presence && utils.isArray(data.presence.events)) {
@@ -258,22 +262,33 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
// Handle invites
inviteRooms.forEach(function(inviteObj) {
var room = inviteObj.room;
var stateEvents = inviteObj.invite_state.events || [];
// add room_id back in f.e event
stateEvents = stateEvents.map(function(e) {
e.room_id = room.roomId;
return e;
});
var stateEvents = self._mapSyncEventsFormat(inviteObj.invite_state, room);
self._processRoomEvents(room, stateEvents);
if (inviteObj.isBrandNewRoom) {
room.recalculate(client.credentials.userId);
client.store.storeRoom(room);
client.emit("Room", room);
console.log("Storing %s %s", room.roomId, JSON.stringify(stateEvents));
}
stateEvents.forEach(function(e) { client.emit("event", e); });
});
// Handle joins
joinRooms.forEach(function(joinObj) {
var room = joinObj.room;
var stateEvents = self._mapSyncEventsFormat(joinObj.state, room);
var timelineEvents = self._mapSyncEventsFormat(joinObj.timeline, room);
self._processRoomEvents(
room, stateEvents, timelineEvents, joinObj.timeline.prev_batch
);
// TODO: Receipts, Typing, Tags from ephermeral + account_data
room.recalculate(client.credentials.userId);
if (joinObj.isBrandNewRoom) {
client.store.storeRoom(room);
client.emit("Room", room);
}
stateEvents.forEach(function(e) { client.emit("event", e); });
timelineEvents.forEach(function(e) { client.emit("event", e); });
});
// Ignore leave rooms for now (TODO: Honour includeArchived opt)
}
@@ -286,14 +301,6 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
var i, j;
// intercept the results and put them into our store
if (!(client.store instanceof StubStore)) {
utils.forEach(
utils.map(data.presence, client.getEventMapper()),
function(e) {
var user = createNewUser(client, e.getContent().user_id);
user.setPresenceEvent(e);
client.store.storeUser(user);
});
// group receipts by room ID.
var receiptsByRoom = {};
data.receipts = data.receipts || [];
@@ -307,27 +314,6 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
);
for (i = 0; i < data.rooms.length; i++) {
var room = createNewRoom(client, data.rooms[i].room_id);
if (!data.rooms[i].state) {
data.rooms[i].state = [];
}
if (data.rooms[i].membership === "invite") {
var inviteEvent = data.rooms[i].invite;
if (!inviteEvent) {
// fallback for servers which don't serve the invite key yet
inviteEvent = {
event_id: "$fake_" + room.roomId,
content: {
membership: "invite"
},
state_key: client.credentials.userId,
user_id: data.rooms[i].inviter,
room_id: room.roomId,
type: "m.room.member"
};
}
data.rooms[i].state.push(inviteEvent);
}
_processRoomEvents(
client, room, data.rooms[i].state, data.rooms[i].messages
@@ -359,30 +345,6 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
}
}
if (data) {
var events = [];
for (i = 0; i < data.presence.length; i++) {
events.push(new MatrixEvent(data.presence[i]));
}
for (i = 0; i < data.rooms.length; i++) {
if (data.rooms[i].state) {
for (j = 0; j < data.rooms[i].state.length; j++) {
events.push(new MatrixEvent(data.rooms[i].state[j]));
}
}
if (data.rooms[i].messages) {
for (j = 0; j < data.rooms[i].messages.chunk.length; j++) {
events.push(
new MatrixEvent(data.rooms[i].messages.chunk[j])
);
}
}
}
utils.forEach(events, function(e) {
client.emit("event", e);
});
}
*/
@@ -393,14 +355,7 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
}
updateSyncState(client, "SYNCING");
self._sync(syncOptions);
}, function(err) {
console.error("/sync error (%s attempts): %s", attempt, err);
attempt += 1;
startSyncingRetryTimer(client, attempt, function() {
self._sync(syncOptions, attempt);
});
updateSyncState(client, "ERROR", { error: err });
});
}, errHandler);
};
SyncApi.prototype._mapSyncResponseToRoomArray = function(obj) {
@@ -423,6 +378,19 @@ SyncApi.prototype._mapSyncResponseToRoomArray = function(obj) {
});
}
SyncApi.prototype._mapSyncEventsFormat = function(obj, room) {
if (!obj || !utils.isArray(obj.events)) {
return [];
}
var mapper = this.client.getEventMapper();
return obj.events.map(function(e) {
if (room) {
e.room_id = room.roomId;
}
return mapper(e);
});
};
SyncApi.prototype._resolveInvites = function(room) {
if (!room || !this.opts.resolveInvitesToProfiles) {
return;
@@ -465,190 +433,43 @@ SyncApi.prototype._resolveInvites = function(room) {
});
}
SyncApi.prototype._processRoomEvents = function(room, stateEventList, messageChunk) {
/**
* @param {Room} room
* @param {MatrixEvent[]} stateEventList A list of state events. This is the state
* at the *START* of the timeline list if it is supplied.
* @param {MatrixEvent[]=} timelineEventList A list of timeline events. Lower index
* is earlier in time. Higher index is later.
* @param {string=} paginationToken
*/
SyncApi.prototype._processRoomEvents = function(room, stateEventList, timelineEventList,
paginationToken) {
timelineEventList = timelineEventList || [];
var client = this.client;
// "old" and "current" state are the same initially; they
// start diverging if the user paginates.
// We must deep copy otherwise membership changes in old state
// will leak through to current state!
var oldStateEvents = utils.map(
utils.deepCopy(stateEventList), client.getEventMapper()
utils.deepCopy(
stateEventList.map(function(mxEvent) { return mxEvent.event; })
), client.getEventMapper()
);
var stateEvents = utils.map(stateEventList, client.getEventMapper());
var stateEvents = stateEventList;
// set the state of the room to as it was before the timeline executes
room.oldState.setStateEvents(oldStateEvents);
room.currentState.setStateEvents(stateEvents);
this._resolveInvites(room);
// add events to the timeline *after* setting the state
// events so messages use the right display names. Initial sync
// returns messages in chronological order, so we need to reverse
// it to get most recent -> oldest. We need it in that order in
// order to diverge old/current state correctly.
room.addEventsToTimeline(
utils.map(
messageChunk ? messageChunk.chunk : [],
client.getEventMapper()
).reverse(), true
);
if (messageChunk) {
room.oldState.paginationToken = messageChunk.start;
// execute the timeline events, this will begin to diverge the current state
// if the timeline has any state events in it.
room.addEventsToTimeline(timelineEventList);
if (paginationToken) {
room.oldState.paginationToken = paginationToken;
}
}
/**
* This is an internal method.
* @param {Number=} attempt The attempt number
*/ /*
SyncApi.prototype._pollForEvents = function(attempt) {
var client = this.client;
var self = this;
attempt = attempt || 1;
if (!client.clientRunning) {
return;
}
var timeoutMs = client._config.pollTimeout;
if (attempt > 1) {
// we think the connection is dead. If it comes back up, we won't know
// about it till /events returns. If the timeout= is high, this could
// be a long time. Set it to 1 when doing retries.
timeoutMs = 1;
}
var discardResult = false;
var timeoutObj = setTimeout(function() {
discardResult = true;
console.error("/events request timed out.");
self._pollForEvents();
}, timeoutMs + (20 * 1000)); // 20s buffer
var queryParams = {
from: client.store.getSyncToken(),
timeout: timeoutMs
};
if (client._guestRooms && client._isGuest) {
queryParams.room_id = client._guestRooms;
}
client._http.authedRequest(undefined, "GET", "/events", queryParams).done(
function(data) {
if (discardResult) {
return;
}
else {
clearTimeout(timeoutObj);
}
if (client._syncState !== "SYNCING") {
updateSyncState(client, "SYNCING");
}
try {
var events = [];
if (data) {
events = utils.map(data.chunk, client.getEventMapper());
}
if (!(client.store instanceof StubStore)) {
var roomIdsWithNewInvites = {};
// bucket events based on room.
var i = 0;
var roomIdToEvents = {};
for (i = 0; i < events.length; i++) {
var roomId = events[i].getRoomId();
// possible to have no room ID e.g. for presence events.
if (roomId) {
if (!roomIdToEvents[roomId]) {
roomIdToEvents[roomId] = [];
}
roomIdToEvents[roomId].push(events[i]);
if (events[i].getType() === "m.room.member" &&
events[i].getContent().membership === "invite") {
roomIdsWithNewInvites[roomId] = true;
}
}
else if (events[i].getType() === "m.presence") {
var usr = client.store.getUser(events[i].getContent().user_id);
if (usr) {
usr.setPresenceEvent(events[i]);
}
else {
usr = createNewUser(client, events[i].getContent().user_id);
usr.setPresenceEvent(events[i]);
client.store.storeUser(usr);
}
}
}
// add events to room
var roomIds = utils.keys(roomIdToEvents);
utils.forEach(roomIds, function(roomId) {
var room = client.store.getRoom(roomId);
var isBrandNewRoom = false;
if (!room) {
room = createNewRoom(client, roomId);
isBrandNewRoom = true;
}
var wasJoined = room.hasMembershipState(
client.credentials.userId, "join"
);
room.addEvents(roomIdToEvents[roomId], "replace");
room.recalculate(client.credentials.userId);
// store the Room for things like invite events so developers
// can update the UI
if (isBrandNewRoom) {
client.store.storeRoom(room);
client.emit("Room", room);
}
var justJoined = room.hasMembershipState(
client.credentials.userId, "join"
);
if (!wasJoined && justJoined) {
// we've just transitioned into a join state for this room,
// so sync state.
_syncRoom(client, room);
}
});
Object.keys(roomIdsWithNewInvites).forEach(function(inviteRoomId) {
_resolveInvites(client, client.store.getRoom(inviteRoomId));
});
}
if (data) {
client.store.setSyncToken(data.end);
utils.forEach(events, function(e) {
client.emit("event", e);
});
}
}
catch (e) {
console.error("Event stream error:");
console.error(e);
}
self._pollForEvents();
}, function(err) {
console.error("/events error: %s", JSON.stringify(err));
if (discardResult) {
return;
}
else {
clearTimeout(timeoutObj);
}
attempt += 1;
startSyncingRetryTimer(client, attempt, function() {
self._pollForEvents(attempt);
});
updateSyncState(client, "ERROR", { error: err });
});
}; */
function retryTimeMsForAttempt(attempt) {
// 2,4,8,16,32,64,128,128,128,... seconds
// max 2^7 secs = 2.1 mins