diff --git a/README.md b/README.md index 2549b11..e054e77 100644 --- a/README.md +++ b/README.md @@ -1,43 +1,23 @@ -# y-websocket :tophat: -> WebSocket Provider for Yjs - -The Websocket Provider implements a classical client server model. Clients -connect to a single endpoint over Websocket. The server distributes awareness -information and document updates among clients. - -This repository contains a simple in-memory backend that can persist to -databases, but it can't be scaled easily. The -[y-redis](https://github.com/yjs/y-redis/) repository contains an alternative -backend that is scalable, provides auth*, and can persist to different backends. +# y-websocket-server :tophat: +> Simple backend for [y-websocket](https://github.com/yjs/y-websocket) The Websocket Provider is a solid choice if you want a central source that handles authentication and authorization. Websockets also send header information and cookies, so you can use existing authentication mechanisms with this server. -* Supports cross-tab communication. When you open the same document in the same -browser, changes on the document are exchanged via cross-tab communication -([Broadcast -Channel](https://developer.mozilla.org/en-US/docs/Web/API/Broadcast_Channel_API) -and -[localStorage](https://developer.mozilla.org/en-US/docs/Web/API/Window/localStorage) -as fallback). -* Supports exchange of awareness information (e.g. cursors). - - - ## Quick Start ### Install dependencies ```sh -npm i y-websocket +npm i @y/websocket-server ``` ### Start a y-websocket server -This repository implements a basic server that you can adopt to your specific use-case. [(source code)](./bin/) +This repository implements a basic server that you can adopt to your specific use-case. [(source code)](./src/) Start a y-websocket server: @@ -59,73 +39,6 @@ wsProvider.on('status', event => { }) ``` -#### Client Code in Node.js - -The WebSocket provider requires a [`WebSocket`](https://developer.mozilla.org/en-US/docs/Web/API/WebSocket) object to create connection to a server. You can polyfill WebSocket support in Node.js using the [`ws` package](https://www.npmjs.com/package/ws). - -```js -const wsProvider = new WebsocketProvider('ws://localhost:1234', 'my-roomname', doc, { WebSocketPolyfill: require('ws') }) -``` - -## API - -```js -import { WebsocketProvider } from 'y-websocket' -``` - -
- wsProvider = new WebsocketProvider(serverUrl: string, room: string, ydoc: Y.Doc [, wsOpts: WsOpts]) -
Create a new websocket-provider instance. As long as this provider, or the connected ydoc, is not destroyed, the changes will be synced to other clients via the connected server. Optionally, you may specify a configuration object. The following default values of wsOpts can be overwritten.
-
- -```js -wsOpts = { - // Set this to `false` if you want to connect manually using wsProvider.connect() - connect: true, - // Specify a query-string / url parameters that will be url-encoded and attached to the `serverUrl` - // I.e. params = { auth: "bearer" } will be transformed to "?auth=bearer" - params: {}, // Object - // You may polyill the Websocket object (https://developer.mozilla.org/en-US/docs/Web/API/WebSocket). - // E.g. In nodejs, you could specify WebsocketPolyfill = require('ws') - WebsocketPolyfill: Websocket, - // Specify an existing Awareness instance - see https://github.com/yjs/y-protocols - awareness: new awarenessProtocol.Awareness(ydoc), - // Specify the maximum amount to wait between reconnects (we use exponential backoff). - maxBackoffTime: 2500 -} -``` - -
- wsProvider.wsconnected: boolean -
True if this instance is currently connected to the server.
- wsProvider.wsconnecting: boolean -
True if this instance is currently connecting to the server.
- wsProvider.shouldConnect: boolean -
If false, the client will not try to reconnect.
- wsProvider.bcconnected: boolean -
True if this instance is currently communicating to other browser-windows via BroadcastChannel.
- wsProvider.synced: boolean -
True if this instance is currently connected and synced with the server.
- wsProvider.params : boolean -
The specified url parameters. This can be safely updated, the new values - will be used when a new connction is established. If this contains an - auth token, it should be updated regularly.
- wsProvider.disconnect() -
Disconnect from the server and don't try to reconnect.
- wsProvider.connect() -
Establish a websocket connection to the websocket-server. Call this if you recently disconnected or if you set wsOpts.connect = false.
- wsProvider.destroy() -
Destroy this wsProvider instance. Disconnects from the server and removes all event handlers.
- wsProvider.on('sync', function(isSynced: boolean)) -
Add an event listener for the sync event that is fired when the client received content from the server.
- wsProvider.on('status', function({ status: 'disconnected' | 'connecting' | 'connected' })) -
Receive updates about the current connection status.
- wsProvider.on('connection-close', function(WSClosedEvent)) -
Fires when the underlying websocket connection is closed. It forwards the websocket event to this event handler.
- wsProvider.on('connection-error', function(WSErrorEvent)) -
Fires when the underlying websocket connection closes with an error. It forwards the websocket event to this event handler.
-
- ## Websocket Server Start a y-websocket server: @@ -143,7 +56,7 @@ Persist document updates in a LevelDB database. See [LevelDB Persistence](https://github.com/yjs/y-leveldb) for more info. ```sh -HOST=localhost PORT=1234 YPERSISTENCE=./dbDir node ./node_modules/y-websocket/bin/server.cjs +HOST=localhost PORT=1234 YPERSISTENCE=./dbDir npx y-websocket ``` ### Websocket Server with HTTP callback diff --git a/package.json b/package.json index 6595d9d..8af6de6 100644 --- a/package.json +++ b/package.json @@ -1,10 +1,7 @@ { - "name": "y-websocket", - "version": "2.1.0", - "description": "Websockets provider for Yjs", - "main": "./dist/y-websocket.cjs", - "module": "./src/y-websocket.js", - "types": "./dist/src/y-websocket.d.ts", + "name": "@y/websocket-server", + "version": "0.1.0", + "description": "Backend for y-websocket", "type": "module", "sideEffects": false, "funding": { @@ -12,36 +9,40 @@ "url": "https://github.com/sponsors/dmonad" }, "scripts": { - "start": "node ./bin/server.cjs", + "start": "node ./src/server.cjs", "dist": "rm -rf dist && rollup -c && tsc", "lint": "standard && tsc", "test": "npm run lint", - "preversion": "npm run lint && npm run dist && test -e dist/src/y-websocket.d.ts && test -e dist/y-websocket.cjs" + "preversion": "npm run lint && npm run dist && test -e dist/src/server.d.ts && test -e dist/server.cjs" }, "bin": { - "y-websocket-server": "./bin/server.cjs", - "y-websocket": "./bin/server.cjs" + "y-websocket-server": "./src/server.js", + "y-websocket": "./src/server.js" }, "files": [ "dist/*", - "bin/*", "src/*" ], "exports": { "./package.json": "./package.json", - "./bin/utils": "./bin/utils.cjs", - "./bin/callback": "./bin/callback.cjs", - ".": { - "module": "./src/y-websocket.js", - "import": "./src/y-websocket.js", - "require": "./dist/y-websocket.cjs", - "types": "./dist/src/y-websocket.d.ts", - "default": "./dist/y-websocket.js" + "./utils": { + "module": "./src/utils.js", + "import": "./src/utils.js", + "require": "./dist/utils.cjs", + "types": "./dist/src/utils.d.ts", + "default": "./src/utils.js" + }, + "./callback": { + "module": "./src/callback.js", + "import": "./src/callback.js", + "require": "./dist/callback.cjs", + "types": "./dist/src/callback.d.ts", + "default": "./src/callback.js" } }, "repository": { "type": "git", - "url": "git+https://github.com/yjs/y-websocket.git" + "url": "git+https://github.com/yjs/y-websocket-server.git" }, "keywords": [ "Yjs" @@ -49,9 +50,9 @@ "author": "Kevin Jahns ", "license": "MIT", "bugs": { - "url": "https://github.com/yjs/y-websocket/issues" + "url": "https://github.com/yjs/y-websocket-server/issues" }, - "homepage": "https://github.com/yjs/y-websocket#readme", + "homepage": "https://github.com/yjs/y-websocket-server#readme", "standard": { "ignore": [ "/dist", diff --git a/rollup.config.mjs b/rollup.config.mjs index 8302a5e..f47d139 100644 --- a/rollup.config.mjs +++ b/rollup.config.mjs @@ -1,5 +1,5 @@ export default [{ - input: ['./src/y-websocket.js', './bin/server.js', './bin/utils.js'], + input: ['./src/server.js', './src/utils.js', './src/callback.js'], external: id => /^(lib0|yjs|y-protocols|ws|lodash\.debounce|http)/.test(id), output: [{ dir: 'dist', diff --git a/bin/callback.js b/src/callback.js similarity index 100% rename from bin/callback.js rename to src/callback.js diff --git a/bin/server.js b/src/server.js similarity index 100% rename from bin/server.js rename to src/server.js diff --git a/bin/utils.js b/src/utils.js similarity index 100% rename from bin/utils.js rename to src/utils.js diff --git a/src/y-websocket.js b/src/y-websocket.js deleted file mode 100644 index c0fc343..0000000 --- a/src/y-websocket.js +++ /dev/null @@ -1,514 +0,0 @@ -/** - * @module provider/websocket - */ - -/* eslint-env browser */ - -import * as Y from 'yjs' // eslint-disable-line -import * as bc from 'lib0/broadcastchannel' -import * as time from 'lib0/time' -import * as encoding from 'lib0/encoding' -import * as decoding from 'lib0/decoding' -import * as syncProtocol from 'y-protocols/sync' -import * as authProtocol from 'y-protocols/auth' -import * as awarenessProtocol from 'y-protocols/awareness' -import { ObservableV2 } from 'lib0/observable' -import * as math from 'lib0/math' -import * as url from 'lib0/url' -import * as env from 'lib0/environment' - -export const messageSync = 0 -export const messageQueryAwareness = 3 -export const messageAwareness = 1 -export const messageAuth = 2 - -/** - * encoder, decoder, provider, emitSynced, messageType - * @type {Array} - */ -const messageHandlers = [] - -messageHandlers[messageSync] = ( - encoder, - decoder, - provider, - emitSynced, - _messageType -) => { - encoding.writeVarUint(encoder, messageSync) - const syncMessageType = syncProtocol.readSyncMessage( - decoder, - encoder, - provider.doc, - provider - ) - if ( - emitSynced && syncMessageType === syncProtocol.messageYjsSyncStep2 && - !provider.synced - ) { - provider.synced = true - } -} - -messageHandlers[messageQueryAwareness] = ( - encoder, - _decoder, - provider, - _emitSynced, - _messageType -) => { - encoding.writeVarUint(encoder, messageAwareness) - encoding.writeVarUint8Array( - encoder, - awarenessProtocol.encodeAwarenessUpdate( - provider.awareness, - Array.from(provider.awareness.getStates().keys()) - ) - ) -} - -messageHandlers[messageAwareness] = ( - _encoder, - decoder, - provider, - _emitSynced, - _messageType -) => { - awarenessProtocol.applyAwarenessUpdate( - provider.awareness, - decoding.readVarUint8Array(decoder), - provider - ) -} - -messageHandlers[messageAuth] = ( - _encoder, - decoder, - provider, - _emitSynced, - _messageType -) => { - authProtocol.readAuthMessage( - decoder, - provider.doc, - (_ydoc, reason) => permissionDeniedHandler(provider, reason) - ) -} - -// @todo - this should depend on awareness.outdatedTime -const messageReconnectTimeout = 30000 - -/** - * @param {WebsocketProvider} provider - * @param {string} reason - */ -const permissionDeniedHandler = (provider, reason) => - console.warn(`Permission denied to access ${provider.url}.\n${reason}`) - -/** - * @param {WebsocketProvider} provider - * @param {Uint8Array} buf - * @param {boolean} emitSynced - * @return {encoding.Encoder} - */ -const readMessage = (provider, buf, emitSynced) => { - const decoder = decoding.createDecoder(buf) - const encoder = encoding.createEncoder() - const messageType = decoding.readVarUint(decoder) - const messageHandler = provider.messageHandlers[messageType] - if (/** @type {any} */ (messageHandler)) { - messageHandler(encoder, decoder, provider, emitSynced, messageType) - } else { - console.error('Unable to compute message') - } - return encoder -} - -/** - * Outsource this function so that a new websocket connection is created immediately. - * I suspect that the `ws.onclose` event is not always fired if there are network issues. - * - * @param {WebsocketProvider} provider - * @param {WebSocket} ws - * @param {CloseEvent | null} event - */ -const closeWebsocketConnection = (provider, ws, event) => { - if (ws === provider.ws) { - provider.emit('connection-close', [event, provider]) - provider.ws = null - ws.close() - provider.wsconnecting = false - if (provider.wsconnected) { - provider.wsconnected = false - provider.synced = false - // update awareness (all users except local left) - awarenessProtocol.removeAwarenessStates( - provider.awareness, - Array.from(provider.awareness.getStates().keys()).filter((client) => - client !== provider.doc.clientID - ), - provider - ) - provider.emit('status', [{ - status: 'disconnected' - }]) - } else { - provider.wsUnsuccessfulReconnects++ - } - // Start with no reconnect timeout and increase timeout by - // using exponential backoff starting with 100ms - setTimeout( - setupWS, - math.min( - math.pow(2, provider.wsUnsuccessfulReconnects) * 100, - provider.maxBackoffTime - ), - provider - ) - } -} - -/** - * @param {WebsocketProvider} provider - */ -const setupWS = (provider) => { - if (provider.shouldConnect && provider.ws === null) { - const websocket = new provider._WS(provider.url, provider.protocols) - websocket.binaryType = 'arraybuffer' - provider.ws = websocket - provider.wsconnecting = true - provider.wsconnected = false - provider.synced = false - - websocket.onmessage = (event) => { - provider.wsLastMessageReceived = time.getUnixTime() - const encoder = readMessage(provider, new Uint8Array(event.data), true) - if (encoding.length(encoder) > 1) { - websocket.send(encoding.toUint8Array(encoder)) - } - } - websocket.onerror = (event) => { - provider.emit('connection-error', [event, provider]) - } - websocket.onclose = (event) => { - closeWebsocketConnection(provider, websocket, event) - } - websocket.onopen = () => { - provider.wsLastMessageReceived = time.getUnixTime() - provider.wsconnecting = false - provider.wsconnected = true - provider.wsUnsuccessfulReconnects = 0 - provider.emit('status', [{ - status: 'connected' - }]) - // always send sync step 1 when connected - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, messageSync) - syncProtocol.writeSyncStep1(encoder, provider.doc) - websocket.send(encoding.toUint8Array(encoder)) - // broadcast local awareness state - if (provider.awareness.getLocalState() !== null) { - const encoderAwarenessState = encoding.createEncoder() - encoding.writeVarUint(encoderAwarenessState, messageAwareness) - encoding.writeVarUint8Array( - encoderAwarenessState, - awarenessProtocol.encodeAwarenessUpdate(provider.awareness, [ - provider.doc.clientID - ]) - ) - websocket.send(encoding.toUint8Array(encoderAwarenessState)) - } - } - provider.emit('status', [{ - status: 'connecting' - }]) - } -} - -/** - * @param {WebsocketProvider} provider - * @param {ArrayBuffer} buf - */ -const broadcastMessage = (provider, buf) => { - const ws = provider.ws - if (provider.wsconnected && ws && ws.readyState === ws.OPEN) { - ws.send(buf) - } - if (provider.bcconnected) { - bc.publish(provider.bcChannel, buf, provider) - } -} - -/** - * Websocket Provider for Yjs. Creates a websocket connection to sync the shared document. - * The document name is attached to the provided url. I.e. the following example - * creates a websocket connection to http://localhost:1234/my-document-name - * - * @example - * import * as Y from 'yjs' - * import { WebsocketProvider } from 'y-websocket' - * const doc = new Y.Doc() - * const provider = new WebsocketProvider('http://localhost:1234', 'my-document-name', doc) - * - * @extends {ObservableV2<{ 'connection-close': (event: CloseEvent | null, provider: WebsocketProvider) => any, 'status': (event: { status: 'connected' | 'disconnected' | 'connecting' }) => any, 'connection-error': (event: Event, provider: WebsocketProvider) => any, 'sync': (state: boolean) => any }>} - */ -export class WebsocketProvider extends ObservableV2 { - /** - * @param {string} serverUrl - * @param {string} roomname - * @param {Y.Doc} doc - * @param {object} opts - * @param {boolean} [opts.connect] - * @param {awarenessProtocol.Awareness} [opts.awareness] - * @param {Object} [opts.params] specify url parameters - * @param {Array} [opts.protocols] specify websocket protocols - * @param {typeof WebSocket} [opts.WebSocketPolyfill] Optionall provide a WebSocket polyfill - * @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 - */ - constructor (serverUrl, roomname, doc, { - connect = true, - awareness = new awarenessProtocol.Awareness(doc), - params = {}, - protocols = [], - WebSocketPolyfill = WebSocket, - resyncInterval = -1, - maxBackoffTime = 2500, - disableBc = false - } = {}) { - super() - // ensure that serverUrl does not end with / - while (serverUrl[serverUrl.length - 1] === '/') { - serverUrl = serverUrl.slice(0, serverUrl.length - 1) - } - this.serverUrl = serverUrl - this.bcChannel = serverUrl + '/' + roomname - this.maxBackoffTime = maxBackoffTime - /** - * The specified url parameters. This can be safely updated. The changed parameters will be used - * when a new connection is established. - * @type {Object} - */ - this.params = params - this.protocols = protocols - this.roomname = roomname - this.doc = doc - this._WS = WebSocketPolyfill - this.awareness = awareness - this.wsconnected = false - this.wsconnecting = false - this.bcconnected = false - this.disableBc = disableBc - this.wsUnsuccessfulReconnects = 0 - this.messageHandlers = messageHandlers.slice() - /** - * @type {boolean} - */ - this._synced = false - /** - * @type {WebSocket?} - */ - this.ws = null - this.wsLastMessageReceived = 0 - /** - * Whether to connect to other peers or not - * @type {boolean} - */ - this.shouldConnect = connect - - /** - * @type {number} - */ - this._resyncInterval = 0 - if (resyncInterval > 0) { - this._resyncInterval = /** @type {any} */ (setInterval(() => { - if (this.ws && this.ws.readyState === WebSocket.OPEN) { - // resend sync step 1 - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, messageSync) - syncProtocol.writeSyncStep1(encoder, doc) - this.ws.send(encoding.toUint8Array(encoder)) - } - }, resyncInterval)) - } - - /** - * @param {ArrayBuffer} data - * @param {any} origin - */ - this._bcSubscriber = (data, origin) => { - if (origin !== this) { - const encoder = readMessage(this, new Uint8Array(data), false) - if (encoding.length(encoder) > 1) { - bc.publish(this.bcChannel, encoding.toUint8Array(encoder), this) - } - } - } - /** - * Listens to Yjs updates and sends them to remote peers (ws and broadcastchannel) - * @param {Uint8Array} update - * @param {any} origin - */ - this._updateHandler = (update, origin) => { - if (origin !== this) { - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, messageSync) - syncProtocol.writeUpdate(encoder, update) - broadcastMessage(this, encoding.toUint8Array(encoder)) - } - } - this.doc.on('update', this._updateHandler) - /** - * @param {any} changed - * @param {any} _origin - */ - this._awarenessUpdateHandler = ({ added, updated, removed }, _origin) => { - const changedClients = added.concat(updated).concat(removed) - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, messageAwareness) - encoding.writeVarUint8Array( - encoder, - awarenessProtocol.encodeAwarenessUpdate(awareness, changedClients) - ) - broadcastMessage(this, encoding.toUint8Array(encoder)) - } - this._exitHandler = () => { - awarenessProtocol.removeAwarenessStates( - this.awareness, - [doc.clientID], - 'app closed' - ) - } - if (env.isNode && typeof process !== 'undefined') { - process.on('exit', this._exitHandler) - } - awareness.on('update', this._awarenessUpdateHandler) - this._checkInterval = /** @type {any} */ (setInterval(() => { - if ( - this.wsconnected && - messageReconnectTimeout < - time.getUnixTime() - this.wsLastMessageReceived - ) { - // no message received in a long time - not even your own awareness - // updates (which are updated every 15 seconds) - closeWebsocketConnection(this, /** @type {WebSocket} */ (this.ws), null) - } - }, messageReconnectTimeout / 10)) - if (connect) { - this.connect() - } - } - - get url () { - const encodedParams = url.encodeQueryParams(this.params) - return this.serverUrl + '/' + this.roomname + - (encodedParams.length === 0 ? '' : '?' + encodedParams) - } - - /** - * @type {boolean} - */ - get synced () { - return this._synced - } - - set synced (state) { - if (this._synced !== state) { - this._synced = state - // @ts-ignore - this.emit('synced', [state]) - this.emit('sync', [state]) - } - } - - destroy () { - if (this._resyncInterval !== 0) { - clearInterval(this._resyncInterval) - } - clearInterval(this._checkInterval) - this.disconnect() - if (env.isNode && typeof process !== 'undefined') { - process.off('exit', this._exitHandler) - } - this.awareness.off('update', this._awarenessUpdateHandler) - this.doc.off('update', this._updateHandler) - super.destroy() - } - - connectBc () { - if (this.disableBc) { - return - } - if (!this.bcconnected) { - bc.subscribe(this.bcChannel, this._bcSubscriber) - this.bcconnected = true - } - // send sync step1 to bc - // write sync step 1 - const encoderSync = encoding.createEncoder() - encoding.writeVarUint(encoderSync, messageSync) - syncProtocol.writeSyncStep1(encoderSync, this.doc) - bc.publish(this.bcChannel, encoding.toUint8Array(encoderSync), this) - // broadcast local state - const encoderState = encoding.createEncoder() - encoding.writeVarUint(encoderState, messageSync) - syncProtocol.writeSyncStep2(encoderState, this.doc) - bc.publish(this.bcChannel, encoding.toUint8Array(encoderState), this) - // write queryAwareness - const encoderAwarenessQuery = encoding.createEncoder() - encoding.writeVarUint(encoderAwarenessQuery, messageQueryAwareness) - bc.publish( - this.bcChannel, - encoding.toUint8Array(encoderAwarenessQuery), - this - ) - // broadcast local awareness state - const encoderAwarenessState = encoding.createEncoder() - encoding.writeVarUint(encoderAwarenessState, messageAwareness) - encoding.writeVarUint8Array( - encoderAwarenessState, - awarenessProtocol.encodeAwarenessUpdate(this.awareness, [ - this.doc.clientID - ]) - ) - bc.publish( - this.bcChannel, - encoding.toUint8Array(encoderAwarenessState), - this - ) - } - - disconnectBc () { - // broadcast message with local awareness state set to null (indicating disconnect) - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, messageAwareness) - encoding.writeVarUint8Array( - encoder, - awarenessProtocol.encodeAwarenessUpdate(this.awareness, [ - this.doc.clientID - ], new Map()) - ) - broadcastMessage(this, encoding.toUint8Array(encoder)) - if (this.bcconnected) { - bc.unsubscribe(this.bcChannel, this._bcSubscriber) - this.bcconnected = false - } - } - - disconnect () { - this.shouldConnect = false - this.disconnectBc() - if (this.ws !== null) { - closeWebsocketConnection(this, this.ws, null) - } - } - - connect () { - this.shouldConnect = true - if (!this.wsconnected && this.ws === null) { - setupWS(this) - this.connectBc() - } - } -} diff --git a/tsconfig.json b/tsconfig.json index e15dee3..4e3a645 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -58,5 +58,5 @@ // "emitDecoratorMetadata": true, /* Enables experimental support for emitting type metadata for decorators. */ // "maxNodeModuleJsDepth": 5 }, - "include": ["./src/y-websocket.js", "./bin/**/*"] + "include": ["./src/**/*"] }