diff --git a/lib/client.js b/lib/client.js index d54140b67..4c2b036ce 100644 --- a/lib/client.js +++ b/lib/client.js @@ -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) { diff --git a/lib/sync.js b/lib/sync.js index eb0fb0068..a1a5e7614 100644 --- a/lib/sync.js +++ b/lib/sync.js @@ -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