implemented websocket provider

provider/websocket/WebSocketProvider.js

@@ -0,0 +1,85 @@
+/* eslint-env browser */
+
+import * as Y from '../../src/index.js'
+export * from '../../src/index.js'
+
+const reconnectTimeout = 100
+
+const setupWS = (doc, url) => {
+  const websocket = new WebSocket(url)
+  websocket.binaryType = 'arraybuffer'
+  doc.ws = websocket
+  websocket.onmessage = event => {
+    const decoder = Y.createDecoder(event.data)
+    const encoder = Y.createEncoder()
+    doc.mux(() =>
+      Y.readMessage(decoder, encoder, doc)
+    )
+    if (Y.length(encoder) > 0) {
+      websocket.send(Y.toBuffer(encoder))
+    }
+  }
+  websocket.onclose = () => {
+    doc.ws = null
+    doc.wsconnected = false
+    doc.emit('status', {
+      status: 'connected'
+    })
+    setTimeout(setupWS, reconnectTimeout, doc, url)
+  }
+  websocket.onopen = () => {
+    doc.wsconnected = true
+    doc.emit('status', {
+      status: 'disconnected'
+    })
+    // always send sync step 1 when connected
+    const encoder = Y.createEncoder()
+    Y.writeSyncStep1(encoder, doc)
+    websocket.send(Y.toBuffer(encoder))
+  }
+}
+
+const broadcastUpdate = (y, transaction) => {
+  if (y.wsconnected && transaction.encodedStructsLen > 0) {
+    y.mux(() => {
+      const encoder = Y.createEncoder()
+      Y.writeUpdate(encoder, transaction.encodedStructsLen, transaction.encodedStructs)
+      y.ws.send(Y.toBuffer(encoder))
+    })
+  }
+}
+
+class WebsocketsSharedDocument extends Y.Y {
+  constructor (url) {
+    super()
+    this.wsconnected = false
+    this.mux = Y.createMutex()
+    setupWS(this, url)
+    this.on('afterTransaction', broadcastUpdate)
+  }
+}
+
+export default 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<string, WebsocketsSharedDocument>}
+     */
+    this.docs = new Map()
+  }
+  /**
+   * @param {string} name
+   * @return {WebsocketsSharedDocument}
+   */
+  get (name) {
+    let doc = this.docs.get(name)
+    if (doc === undefined) {
+      doc = new WebsocketsSharedDocument(this.url + name)
+    }
+    return doc
+  }
+}

provider/websocket/server.js

@@ -0,0 +1,53 @@
+const Y = require('../../build/node/index.js')
+const WebSocket = require('ws')
+const wss = new WebSocket.Server({ port: 1234 })
+const docs = new Map()
+
+const afterTransaction = (doc, transaction) => {
+  if (transaction.encodedStructsLen > 0) {
+    const encoder = Y.createEncoder()
+    Y.writeUpdate(encoder, transaction.encodedStructsLen, transaction.encodedStructs)
+    const message = Y.toBuffer(encoder)
+    doc.conns.forEach(conn => conn.send(message))
+  }
+}
+
+class WSSharedDoc extends Y.Y {
+  constructor () {
+    super()
+    this.mux = Y.createMutex()
+    this.conns = new Set()
+    this.on('afterTransaction', afterTransaction)
+  }
+}
+
+const messageListener = (conn, doc, message) => {
+  const encoder = Y.createEncoder()
+  const decoder = Y.createDecoder(message)
+  Y.readMessage(decoder, encoder, doc)
+  if (Y.length(encoder) > 0) {
+    conn.send(Y.toBuffer(encoder))
+  }
+}
+
+const setupConnection = (conn, req) => {
+  conn.binaryType = 'arraybuffer'
+  // get doc, create if it does not exist yet
+  let doc = docs.get(req.url.slice(1))
+  if (doc === undefined) {
+    doc = new WSSharedDoc()
+    docs.set(req.url.slice(1), doc)
+  }
+  doc.conns.add(conn)
+  // listen and reply to events
+  conn.on('message', message => messageListener(conn, doc, message))
+  conn.on('close', () =>
+    doc.conns.delete(conn)
+  )
+  // send sync step 1
+  const encoder = Y.createEncoder()
+  Y.writeSyncStep1(encoder, doc)
+  conn.send(Y.toBuffer(encoder))
+}
+
+wss.on('connection', setupConnection)