diff --git a/src/y-websocket.js b/src/y-websocket.js index 69ae05fc..0a4704cc 100644 --- a/src/y-websocket.js +++ b/src/y-websocket.js @@ -266,6 +266,7 @@ export class WebsocketProvider extends ObservableV2 { * @param {number} [opts.resyncInterval] Request server state every `resyncInterval` milliseconds * @param {number} [opts.maxBackoffTime] Maximum amount of time to wait before trying to reconnect (we try to reconnect using exponential backoff) * @param {boolean} [opts.disableBc] Disable cross-tab BroadcastChannel communication + * @param {number} [opts.throttle] Throttle interval for sending updates (in ms) */ constructor (serverUrl, roomname, doc, { connect = true, @@ -275,7 +276,8 @@ export class WebsocketProvider extends ObservableV2 { WebSocketPolyfill = WebSocket, resyncInterval = -1, maxBackoffTime = 2500, - disableBc = false + disableBc = false, + throttle = 0 } = {}) { super() // ensure that serverUrl does not end with / @@ -302,6 +304,7 @@ export class WebsocketProvider extends ObservableV2 { this.disableBc = disableBc this.wsUnsuccessfulReconnects = 0 this.messageHandlers = messageHandlers.slice() + this.throttle = throttle /** * @type {boolean} */ @@ -358,7 +361,64 @@ export class WebsocketProvider extends ObservableV2 { broadcastMessage(this, encoding.toUint8Array(encoder)) } } - this.doc.on('update', this._updateHandler) + + /** + * queued updates if throttling is enabled + * @type {*[]} + * @private + */ + this._updateQueue = []; + + /** + * timer for throttling + * @type {any} + * @private + */ + this._throttleTimer = undefined; + + /** + * send the queued updates + * @private + */ + this._batchBroadcastUpdates = () => { + if (this._updateQueue.length > 0) { + const mergedUpdate = this._updateQueue.length === 1 + ? this._updateQueue[0] + : Y.mergeUpdates(this._updateQueue); + + this._updateQueue = []; + + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, messageSync); + syncProtocol.writeUpdate(encoder, mergedUpdate); + + broadcastMessage(this, encoding.toUint8Array(encoder)); + } + }; + + /** + * Listens to Yjs updates and adds them to the update queue for throttling + * @param {Uint8Array} update + * @param {any} origin + */ + this._throttledUpdateHandler = (update, origin) => { + if (origin !== this) { + this._updateQueue.push(update) + if (!this._throttleTimer) { + this._throttleTimer = setTimeout(() => { + delete this._throttleTimer; + this._batchBroadcastUpdates(); + }, this.throttle); + } + } + } + + if (this.throttle > 0) { + this.doc.on('update', this._throttledUpdateHandler) + } else { + this.doc.on('update', this._updateHandler) + } + /** * @param {any} changed * @param {any} _origin @@ -498,6 +558,11 @@ export class WebsocketProvider extends ObservableV2 { disconnect () { this.shouldConnect = false + if (this._throttleTimer) { + clearTimeout(this._throttleTimer); + delete this._throttleTimer; + this._batchBroadcastUpdates(); + } this.disconnectBc() if (this.ws !== null) { closeWebsocketConnection(this, this.ws, null)