From 9208e699e2450f3e15c112d0de24f4ba51ca6ebd Mon Sep 17 00:00:00 2001 From: Kevin Jahns Date: Sat, 30 Mar 2019 11:09:37 +0100 Subject: [PATCH] init --- bin/server.js | 153 +++++++++++++++++++++++++++++++++++ package-lock.json | 37 +++++++++ package.json | 33 ++++++++ rollup.config.js | 18 +++++ src/y-websocket.js | 198 +++++++++++++++++++++++++++++++++++++++++++++ tsconfig.json | 59 ++++++++++++++ 6 files changed, 498 insertions(+) create mode 100755 bin/server.js create mode 100644 package-lock.json create mode 100644 package.json create mode 100644 rollup.config.js create mode 100644 src/y-websocket.js create mode 100644 tsconfig.json diff --git a/bin/server.js b/bin/server.js new file mode 100755 index 0000000..afee6c4 --- /dev/null +++ b/bin/server.js @@ -0,0 +1,153 @@ +/* +Unlike stated in the LICENSE file, it is not necessary to include the copyright notice and permission notice when you copy code from this file. +*/ + +const Y = require('yjs') +const WebSocket = require('ws') +const http = require('http') +const encoding = require('lib0/dist/encoding.js') +const decoding = require('lib0/dist/decoding.js') + +const port = process.env.PORT || 1234 + +// disable gc when using snapshots! +const gcEnabled = process.env.GC !== 'false' && process.env.GC !== '0' +const persistenceDir = process.env.YPERSISTENCE +let persistence = null +if (typeof persistenceDir === 'string') { + const LevelDbPersistence = require('y-leveldb').LevelDbPersistence + persistence = new LevelDbPersistence(persistenceDir) +} + +const server = http.createServer((req, res) => { + res.writeHead(200, { 'Content-Type': 'text/plain' }) + res.end('okay') +}) + +const wss = new WebSocket.Server({ noServer: true }) + +const docs = new Map() + +const messageSync = 0 +const messageAwareness = 1 +const messageAuth = 2 + +const afterTransaction = (doc, transaction) => { + if (transaction.encodedStructsLen > 0) { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageSync) + Y.syncProtocol.writeUpdate(encoder, transaction.encodedStructsLen, transaction.encodedStructs) + const message = encoding.toBuffer(encoder) + doc.conns.forEach((_, conn) => conn.send(message)) + } +} + +class WSSharedDoc extends Y.Y { + constructor () { + super({ gc: gcEnabled }) + this.mux = Y.createMutex() + /** + * Maps from conn to set of controlled user ids. Delete all user ids from awareness when this conn is closed + * @type {Map>} + */ + this.conns = new Map() + this.awareness = new Map() + this.awarenessClock = new Map() + this.on('afterTransaction', afterTransaction) + } +} + +const messageListener = (conn, doc, message) => { + const encoder = encoding.createEncoder() + const decoder = decoding.createDecoder(message) + const messageType = decoding.readVarUint(decoder) + switch (messageType) { + case messageSync: + encoding.writeVarUint(encoder, messageSync) + Y.syncProtocol.readSyncMessage(decoder, encoder, doc) + if (encoding.length(encoder) > 1) { + conn.send(encoding.toBuffer(encoder)) + } + break + case messageAwareness: { + encoding.writeVarUint(encoder, messageAwareness) + const updates = Y.awarenessProtocol.forwardAwarenessMessage(decoder, encoder) + updates.forEach(update => { + doc.awareness.set(update.userID, update.state) + doc.awarenessClock.set(update.userID, update.clock) + doc.conns.get(conn).add(update.userID) + }) + const buff = encoding.toBuffer(encoder) + doc.conns.forEach((_, c) => { + c.send(buff) + }) + break + } + } +} + +const setupConnection = (conn, req) => { + conn.binaryType = 'arraybuffer' + // get doc, create if it does not exist yet + const docName = req.url.slice(1) + let doc = docs.get(docName) + if (doc === undefined) { + doc = new WSSharedDoc() + if (persistence !== null) { + persistence.bindState(docName, doc) + } + docs.set(docName, doc) + } + doc.conns.set(conn, new Set()) + // listen and reply to events + conn.on('message', message => messageListener(conn, doc, message)) + conn.on('close', () => { + const controlledIds = doc.conns.get(conn) + doc.conns.delete(conn) + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageAwareness) + Y.awarenessProtocol.writeUsersStateChange(encoder, Array.from(controlledIds).map(userID => { + const clock = (doc.awarenessClock.get(userID) || 0) + 1 + doc.awareness.delete(userID) + doc.awarenessClock.delete(userID) + return { userID, state: null, clock } + })) + const buf = encoding.toBuffer(encoder) + doc.conns.forEach((_, conn) => conn.send(buf)) + if (doc.conns.size === 0 && persistence !== null) { + // if persisted, we store state and destroy ydocument + persistence.writeState(docName, doc).then(() => { + doc.destroy() + }) + docs.delete(docName) + } + }) + // send sync step 1 + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageSync) + Y.syncProtocol.writeSyncStep1(encoder, doc) + conn.send(encoding.toBuffer(encoder)) + if (doc.awareness.size > 0) { + const encoder = encoding.createEncoder() + const userStates = [] + doc.awareness.forEach((state, userID) => { + userStates.push({ state, userID, clock: (doc.awarenessClock.get(userID) || 0) }) + }) + encoding.writeVarUint(encoder, messageAwareness) + Y.awarenessProtocol.writeUsersStateChange(encoder, userStates) + conn.send(encoding.toBuffer(encoder)) + } +} + +wss.on('connection', setupConnection) + +server.on('upgrade', (request, socket, head) => { + // You may check auth of request here.. + wss.handleUpgrade(request, socket, head, ws => { + wss.emit('connection', ws, request) + }) +}) + +server.listen(port) + +console.log('running on port', port) diff --git a/package-lock.json b/package-lock.json new file mode 100644 index 0000000..4087efb --- /dev/null +++ b/package-lock.json @@ -0,0 +1,37 @@ +{ + "name": "y-websocket", + "version": "1.0.0", + "lockfileVersion": 1, + "requires": true, + "dependencies": { + "@types/estree": { + "version": "0.0.39", + "resolved": "https://registry.npmjs.org/@types/estree/-/estree-0.0.39.tgz", + "integrity": "sha512-EYNwp3bU+98cpU4lAWYYL7Zz+2gryWH1qbdDTidVd6hkiR6weksdbMadyXKXNPEkQFhXM+hVO9ZygomHXp+AIw==", + "dev": true + }, + "@types/node": { + "version": "10.12.21", + "resolved": "https://registry.npmjs.org/@types/node/-/node-10.12.21.tgz", + "integrity": "sha512-CBgLNk4o3XMnqMc0rhb6lc77IwShMEglz05deDcn2lQxyXEZivfwgYJu7SMha9V5XcrP6qZuevTHV/QrN2vjKQ==", + "dev": true + }, + "acorn": { + "version": "6.0.7", + "resolved": "https://registry.npmjs.org/acorn/-/acorn-6.0.7.tgz", + "integrity": "sha512-HNJNgE60C9eOTgn974Tlp3dpLZdUr+SoxxDwPaY9J/kDNOLQTkaDgwBUXAF4SSsrAwD9RpdxuHK/EbuF+W9Ahw==", + "dev": true + }, + "rollup": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/rollup/-/rollup-1.1.2.tgz", + "integrity": "sha512-OkdMxqMl8pWoQc5D8y1cIinYQPPLV8ZkfLgCzL6SytXeNA2P7UHynEQXI9tYxuAjAMsSyvRaWnyJDLHMxq0XAg==", + "dev": true, + "requires": { + "@types/estree": "0.0.39", + "@types/node": "*", + "acorn": "^6.0.5" + } + } + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..4f29cce --- /dev/null +++ b/package.json @@ -0,0 +1,33 @@ +{ + "name": "y-websocket", + "version": "1.0.0", + "description": "Yjs Websocket Provider", + "main": "./dist/y-websocket.js", + "module": "./src/y-websocket.js", + "bin": { + "y-websocket": "./bin/server.js" + }, + "dependencies": {}, + "devDependencies": { + "funlib": "*", + "y-protocols": "*", + "rollup": "^1.1.2" + }, + "scripts": { + "test": "echo \"Error: no test specified\" && exit 1", + "dist": "rollup -c" + }, + "repository": { + "type": "git", + "url": "git+https://github.com/y-js/y-websocket.git" + }, + "keywords": [ + "Yjs" + ], + "author": "Kevin Jahns ", + "license": "MIT", + "bugs": { + "url": "https://github.com/y-js/y-websocket/issues" + }, + "homepage": "https://github.com/y-js/y-websocket#readme" +} diff --git a/rollup.config.js b/rollup.config.js new file mode 100644 index 0000000..e569ea4 --- /dev/null +++ b/rollup.config.js @@ -0,0 +1,18 @@ +export default { + input: './src/y-websocket.js', + external: id => /^(funlib|yjs|y-protocols)/.test(id), + output: [{ + name: 'y-websocket', + file: 'dist/y-websocket.js', + format: 'cjs', + sourcemap: true, + paths: path => { + if (/^funlib\//.test(path)) { + return `lib0/dist${path.slice(6)}` + } else if (/^y\-protocols\//.test(path)) { + return `y-protocols/dist${path.slice(11)}` + } + return path + } + }] +} diff --git a/src/y-websocket.js b/src/y-websocket.js new file mode 100644 index 0000000..e31ff4d --- /dev/null +++ b/src/y-websocket.js @@ -0,0 +1,198 @@ +/* +Unlike stated in the LICENSE file, it is not necessary to include the copyright notice and permission notice when you copy code from this file. +*/ + +/** + * @module provider/websocket + */ + +/* eslint-env browser */ + +import * as Y from 'yjs' +import * as bc from 'lib0/broadcastchannel.js' +import * as encoding from 'lib0/encoding.js' +import * as decoding from 'lib0/decoding.js' +import * as syncProtocol from 'y-protocols/sync.js' +import * as authProtocol from 'y-protocols/auth.js' +import * as awarenessProtocol from 'y-protocols/awareness.js' +import * as mutex from 'lib0/mutex.js' + +const messageSync = 0 +const messageAwareness = 1 +const messageAuth = 2 + +const reconnectTimeout = 3000 + +/** + * @param {WebsocketsSharedDocument} doc + * @param {string} reason + */ +const permissionDeniedHandler = (doc, reason) => console.warn(`Permission denied to access ${doc.url}.\n${reason}`) + +/** + * @param {WebsocketsSharedDocument} doc + * @param {ArrayBuffer} buf + * @return {encoding.Encoder} + */ +const readMessage = (doc, buf) => { + const decoder = decoding.createDecoder(buf) + const encoder = encoding.createEncoder() + const messageType = decoding.readVarUint(decoder) + switch (messageType) { + case messageSync: + encoding.writeVarUint(encoder, messageSync) + doc.mux(() => + syncProtocol.readSyncMessage(decoder, encoder, doc) + ) + break + case messageAwareness: + awarenessProtocol.readAwarenessMessage(decoder, doc) + break + case messageAuth: + authProtocol.readAuthMessage(decoder, doc, permissionDeniedHandler) + } + return encoder +} + +const setupWS = (doc, url) => { + const websocket = new WebSocket(url) + websocket.binaryType = 'arraybuffer' + doc.ws = websocket + websocket.onmessage = event => { + const encoder = readMessage(doc, event.data) + if (encoding.length(encoder) > 1) { + websocket.send(encoding.toBuffer(encoder)) + } + } + websocket.onclose = () => { + doc.ws = null + doc.wsconnected = false + // update awareness (all users left) + const removed = [] + doc.getAwarenessInfo().forEach((_, userid) => { + removed.push(userid) + }) + doc.awareness = new Map() + doc.emit('awareness', [{ + added: [], updated: [], removed + }]) + doc.emit('status', [{ + status: 'disconnected' + }]) + setTimeout(setupWS, reconnectTimeout, doc, url) + } + websocket.onopen = () => { + doc.wsconnected = true + doc.emit('status', [{ + status: 'connected' + }]) + // always send sync step 1 when connected + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageSync) + syncProtocol.writeSyncStep1(encoder, doc) + websocket.send(encoding.toBuffer(encoder)) + // force send stored awareness info + doc.setAwarenessField(null, null) + } +} + +const broadcastUpdate = (y, transaction) => { + if (transaction.encodedStructsLen > 0) { + y.mux(() => { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageSync) + syncProtocol.writeUpdate(encoder, transaction.encodedStructsLen, transaction.encodedStructs) + const buf = encoding.toBuffer(encoder) + if (y.wsconnected) { + y.ws.send(buf) + } + bc.publish(y.url, buf) + }) + } +} + +class WebsocketsSharedDocument extends Y.Y { + constructor (url, opts) { + super(opts) + this.url = url + this.wsconnected = false + this.mux = mutex.createMutex() + this.ws = null + this._localAwarenessState = {} + this.awareness = new Map() + this.awarenessClock = new Map() + setupWS(this, url) + this.on('afterTransaction', broadcastUpdate) + this._bcSubscriber = data => { + const encoder = readMessage(this, data) // already muxed + this.mux(() => { + if (encoding.length(encoder) > 1) { + bc.publish(url, encoding.toBuffer(encoder)) + } + }) + } + bc.subscribe(url, this._bcSubscriber) + // send sync step1 to bc + this.mux(() => { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageSync) + syncProtocol.writeSyncStep1(encoder, this) + bc.publish(url, encoding.toBuffer(encoder)) + }) + } + getLocalAwarenessInfo () { + return this._localAwarenessState + } + getAwarenessInfo () { + return this.awareness + } + setAwarenessField (field, value) { + if (field !== null) { + this._localAwarenessState[field] = value + } + if (this.wsconnected) { + const clock = (this.awarenessClock.get(this.userID) || 0) + 1 + this.awarenessClock.set(this.userID, clock) + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageAwareness) + awarenessProtocol.writeUsersStateChange(encoder, [{ userID: this.userID, state: this._localAwarenessState, clock }]) + const buf = encoding.toBuffer(encoder) + this.ws.send(buf) + } + } +} + +/** + * Websocket Provider for Yjs. Creates a single websocket connection to each 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 { WebsocketProvider } from 'yjs/provider/websocket/client.js' + * const provider = new WebsocketProvider('http://localhost:1234') + * const ydocument = provider.get('my-document-name') + */ +export class WebsocketProvider { + constructor (url) { + // ensure that url is always ends with / + while (url[url.length - 1] === '/') { + url = url.slice(0, url.length - 1) + } + this.url = url + '/' + /** + * @type {Map} + */ + this.docs = new Map() + } + /** + * @param {string} name + * @return {WebsocketsSharedDocument} + */ + get (name, opts) { + let doc = this.docs.get(name) + if (doc === undefined) { + doc = new WebsocketsSharedDocument(this.url + name, opts) + } + return doc + } +} diff --git a/tsconfig.json b/tsconfig.json new file mode 100644 index 0000000..98c6f8a --- /dev/null +++ b/tsconfig.json @@ -0,0 +1,59 @@ +{ + "compilerOptions": { + /* Basic Options */ + "target": "es2018", + "lib": ["es2018", "dom"], /* Specify library files to be included in the compilation. */ + "allowJs": true, /* Allow javascript files to be compiled. */ + "checkJs": true, /* Report errors in .js files. */ + // "jsx": "preserve", /* Specify JSX code generation: 'preserve', 'react-native', or 'react'. */ + // "declaration": true, /* Generates corresponding '.d.ts' file. */ + // "declarationMap": true, /* Generates a sourcemap for each corresponding '.d.ts' file. */ + // "sourceMap": true, /* Generates corresponding '.map' file. */ + // "outFile": "./", /* Concatenate and emit output to single file. */ + // "outDir": "./build", /* Redirect output structure to the directory. */ + // "rootDir": "./", /* Specify the root directory of input files. Use to control the output directory structure with --outDir. */ + // "composite": true, /* Enable project compilation */ + // "removeComments": true, /* Do not emit comments to output. */ + "noEmit": true, /* Do not emit outputs. */ + // "importHelpers": true, /* Import emit helpers from 'tslib'. */ + // "downlevelIteration": true, /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */ + // "isolatedModules": true, /* Transpile each file as a separate module (similar to 'ts.transpileModule'). */ + + /* Strict Type-Checking Options */ + "strict": true, /* Enable all strict type-checking options. */ + "noImplicitAny": false, /* Raise error on expressions and declarations with an implied 'any' type. */ + // "strictNullChecks": true, /* Enable strict null checks. */ + // "strictFunctionTypes": true, /* Enable strict checking of function types. */ + // "strictPropertyInitialization": true, /* Enable strict checking of property initialization in classes. */ + // "noImplicitThis": true, /* Raise error on 'this' expressions with an implied 'any' type. */ + // "alwaysStrict": true, /* Parse in strict mode and emit "use strict" for each source file. */ + + /* Additional Checks */ + // "noUnusedLocals": true, /* Report errors on unused locals. */ + // "noUnusedParameters": true, /* Report errors on unused parameters. */ + // "noImplicitReturns": true, /* Report error when not all code paths in function return a value. */ + // "noFallthroughCasesInSwitch": true, /* Report errors for fallthrough cases in switch statement. */ + + /* Module Resolution Options */ + "moduleResolution": "node", /* Specify module resolution strategy: 'node' (Node.js) or 'classic' (TypeScript pre-1.6). */ + // "baseUrl": "./", /* Base directory to resolve non-absolute module names. */ + // "paths": {}, /* A series of entries which re-map imports to lookup locations relative to the 'baseUrl'. */ + // "rootDirs": [], /* List of root folders whose combined content represents the structure of the project at runtime. */ + // "typeRoots": [], /* List of folders to include type definitions from. */ + // "types": [], /* Type declaration files to be included in compilation. */ + // "allowSyntheticDefaultImports": true, /* Allow default imports from modules with no default export. This does not affect code emit, just typechecking. */ + "esModuleInterop": true, /* Enables emit interoperability between CommonJS and ES Modules via creation of namespace objects for all imports. Implies 'allowSyntheticDefaultImports'. */ + // "preserveSymlinks": true, /* Do not resolve the real path of symlinks. */ + + /* Source Map Options */ + // "sourceRoot": "", /* Specify the location where debugger should locate TypeScript files instead of source locations. */ + // "mapRoot": "", /* Specify the location where debugger should locate map files instead of generated locations. */ + // "inlineSourceMap": true, /* Emit a single file with source maps instead of having a separate file. */ + // "inlineSources": true, /* Emit the source alongside the sourcemaps within a single file; requires '--inlineSourceMap' or '--sourceMap' to be set. */ + + /* Experimental Options */ + // "experimentalDecorators": true, /* Enables experimental support for ES7 decorators. */ + // "emitDecoratorMetadata": true, /* Enables experimental support for emitting type metadata for decorators. */ + "maxNodeModuleJsDepth": 5 + } +}