From 80ee085c1dbe6ae808469608292fe66397503567 Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Thu, 30 Jan 2020 16:07:16 -0300 Subject: [PATCH 1/3] Memoize and batch all createOrUpdateSubscription executions in the same window --- app/lib/methods/subscriptions/rooms.js | 50 ++++++++++++++++++++++++-- 1 file changed, 47 insertions(+), 3 deletions(-) diff --git a/app/lib/methods/subscriptions/rooms.js b/app/lib/methods/subscriptions/rooms.js index 91be40fc9e0..8f62bf24803 100644 --- a/app/lib/methods/subscriptions/rooms.js +++ b/app/lib/methods/subscriptions/rooms.js @@ -16,8 +16,12 @@ let connectedListener; let disconnectedListener; let streamListener; let subServer; +let subQueue = {}; +let subTimer = null; +let roomQueue = {}; +let roomTimer = null; +const WINDOW_TIME = 1000; -// TODO: batch execution const createOrUpdateSubscription = async(subscription, room) => { try { const db = database.active; @@ -162,6 +166,36 @@ const createOrUpdateSubscription = async(subscription, room) => { } }; +const debouncedUpdateSub = (subscription) => { + if (!subTimer) { + subTimer = setTimeout(() => { + const subBatch = subQueue; + subQueue = {}; + Object.keys(subBatch).forEach((key) => { + createOrUpdateSubscription(subBatch[key]); + }); + subTimer = null; + subQueue = {}; + }, WINDOW_TIME); + } + subQueue[subscription.rid] = subscription; +}; + +const debouncedUpdateRoom = (room) => { + if (!roomTimer) { + roomTimer = setTimeout(() => { + const roomBatch = roomQueue; + roomQueue = {}; + Object.keys(roomBatch).forEach((key) => { + createOrUpdateSubscription(null, roomBatch[key]); + }); + roomTimer = null; + roomQueue = {}; + }, WINDOW_TIME); + } + roomQueue[room._id] = room; +}; + export default function subscribeRooms() { const handleConnection = () => { store.dispatch(roomsRequest()); @@ -202,12 +236,12 @@ export default function subscribeRooms() { log(e); } } else { - await createOrUpdateSubscription(data); + debouncedUpdateSub(data); } } if (/rooms/.test(ev)) { if (type === 'updated' || type === 'inserted') { - await createOrUpdateSubscription(null, data); + debouncedUpdateRoom(data); } } if (/message/.test(ev)) { @@ -257,6 +291,16 @@ export default function subscribeRooms() { streamListener.then(removeListener); streamListener = false; } + subQueue = {}; + roomQueue = {}; + if (subTimer) { + clearTimeout(subTimer); + subTimer = false; + } + if (roomTimer) { + clearTimeout(roomTimer); + roomTimer = false; + } }; connectedListener = this.sdk.onStreamData('connected', handleConnection); From 9638564a0cfd8b4d5938e3abed601229fd5ade5e Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Thu, 30 Jan 2020 16:21:09 -0300 Subject: [PATCH 2/3] Try inserting last message again --- app/lib/methods/subscriptions/rooms.js | 51 +++++++++++++------------- app/sagas/rooms.js | 22 +++++++++++ 2 files changed, 48 insertions(+), 25 deletions(-) diff --git a/app/lib/methods/subscriptions/rooms.js b/app/lib/methods/subscriptions/rooms.js index 8f62bf24803..16d0c283039 100644 --- a/app/lib/methods/subscriptions/rooms.js +++ b/app/lib/methods/subscriptions/rooms.js @@ -9,6 +9,7 @@ import random from '../../../utils/random'; import store from '../../createStore'; import { roomsRequest } from '../../../actions/rooms'; import { notificationReceived } from '../../../actions/notification'; +import buildMessage from '../helpers/buildMessage'; const removeListener = listener => listener.stop(); @@ -132,32 +133,32 @@ const createOrUpdateSubscription = async(subscription, room) => { } } - // if (tmp.lastMessage) { - // const lastMessage = buildMessage(tmp.lastMessage); - // const messagesCollection = db.collections.get('messages'); - // let messageRecord; - // try { - // messageRecord = await messagesCollection.find(lastMessage._id); - // } catch (error) { - // // Do nothing - // } + if (tmp.lastMessage) { + const lastMessage = buildMessage(tmp.lastMessage); + const messagesCollection = db.collections.get('messages'); + let messageRecord; + try { + messageRecord = await messagesCollection.find(lastMessage._id); + } catch (error) { + // Do nothing + } - // if (messageRecord) { - // batch.push( - // messageRecord.prepareUpdate(() => { - // Object.assign(messageRecord, lastMessage); - // }) - // ); - // } else { - // batch.push( - // messagesCollection.prepareCreate((m) => { - // m._raw = sanitizedRaw({ id: lastMessage._id }, messagesCollection.schema); - // m.subscription.id = lastMessage.rid; - // return Object.assign(m, lastMessage); - // }) - // ); - // } - // } + if (messageRecord) { + batch.push( + messageRecord.prepareUpdate(() => { + Object.assign(messageRecord, lastMessage); + }) + ); + } else { + batch.push( + messagesCollection.prepareCreate((m) => { + m._raw = sanitizedRaw({ id: lastMessage._id }, messagesCollection.schema); + m.subscription.id = lastMessage.rid; + return Object.assign(m, lastMessage); + }) + ); + } + } await db.batch(...batch); }); diff --git a/app/sagas/rooms.js b/app/sagas/rooms.js index 7476724bd4a..641a75e53ee 100644 --- a/app/sagas/rooms.js +++ b/app/sagas/rooms.js @@ -11,6 +11,8 @@ import database from '../lib/database'; import log from '../utils/log'; import mergeSubscriptionsRooms from '../lib/methods/helpers/mergeSubscriptionsRooms'; import RocketChat from '../lib/rocketchat'; +import buildMessage from '../lib/methods/helpers/buildMessage'; +import protectedFunction from '../lib/methods/helpers/protectedFunction'; const updateRooms = function* updateRooms({ server, newRoomsUpdatedAt }) { const serversDB = database.servers; @@ -38,6 +40,7 @@ const handleRoomsRequest = function* handleRoomsRequest() { const db = database.active; const subCollection = db.collections.get('subscriptions'); + const messagesCollection = db.collections.get('messages'); if (subscriptions.length) { const subsIds = subscriptions.map(sub => sub.rid); @@ -46,6 +49,14 @@ const handleRoomsRequest = function* handleRoomsRequest() { const subsToCreate = subscriptions.filter(i1 => !existingSubs.find(i2 => i1._id === i2._id)); // TODO: subsToDelete? + const lastMessages = subscriptions + .map(sub => sub.lastMessage && buildMessage(sub.lastMessage)) + .filter(lm => lm); + const lastMessagesIds = lastMessages.map(lm => lm._id); + const existingMessages = yield messagesCollection.query(Q.where('id', Q.oneOf(lastMessagesIds))).fetch(); + const messagesToUpdate = existingMessages.filter(i1 => lastMessages.find(i2 => i1.id === i2._id)); + const messagesToCreate = lastMessages.filter(i1 => !existingMessages.find(i2 => i1._id === i2.id)); + const allRecords = [ ...subsToCreate.map(subscription => subCollection.prepareCreate((s) => { s._raw = sanitizedRaw({ id: subscription.rid }, subCollection.schema); @@ -56,6 +67,17 @@ const handleRoomsRequest = function* handleRoomsRequest() { return subscription.prepareUpdate(() => { Object.assign(subscription, newSub); }); + }), + ...messagesToCreate.map(message => messagesCollection.prepareCreate(protectedFunction((m) => { + m._raw = sanitizedRaw({ id: message._id }, messagesCollection.schema); + m.subscription.id = message.rid; + return Object.assign(m, message); + }))), + ...messagesToUpdate.map((message) => { + const newMessage = lastMessages.find(m => m._id === message.id); + return message.prepareUpdate(protectedFunction(() => { + Object.assign(message, newMessage); + })); }) ]; From ad1c6c0da54e2e66286de9a23805d9f68ce1187d Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Thu, 30 Jan 2020 16:32:01 -0300 Subject: [PATCH 3/3] Fix timer --- app/lib/methods/subscriptions/rooms.js | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/app/lib/methods/subscriptions/rooms.js b/app/lib/methods/subscriptions/rooms.js index 16d0c283039..5cf610a0481 100644 --- a/app/lib/methods/subscriptions/rooms.js +++ b/app/lib/methods/subscriptions/rooms.js @@ -172,11 +172,10 @@ const debouncedUpdateSub = (subscription) => { subTimer = setTimeout(() => { const subBatch = subQueue; subQueue = {}; + subTimer = null; Object.keys(subBatch).forEach((key) => { createOrUpdateSubscription(subBatch[key]); }); - subTimer = null; - subQueue = {}; }, WINDOW_TIME); } subQueue[subscription.rid] = subscription; @@ -187,11 +186,10 @@ const debouncedUpdateRoom = (room) => { roomTimer = setTimeout(() => { const roomBatch = roomQueue; roomQueue = {}; + roomTimer = null; Object.keys(roomBatch).forEach((key) => { createOrUpdateSubscription(null, roomBatch[key]); }); - roomTimer = null; - roomQueue = {}; }, WINDOW_TIME); } roomQueue[room._id] = room;