FRE-600: Fix code review blockers

- Consolidated duplicate UndoManagers to single instance
- Fixed connection promise to only resolve on 'connected' status
- Fixed WebSocketProvider import (WebsocketProvider)
- Added proper doc.destroy() cleanup
- Renamed isPresenceInitialized property to avoid conflict

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
2026-04-25 00:08:01 -04:00
parent 65b552bb08
commit 7c684a42cc
48450 changed files with 5679671 additions and 383 deletions

21
node_modules/y-websocket/LICENSE generated vendored Normal file
View File

@@ -0,0 +1,21 @@
The MIT License (MIT)
Copyright (c) 2019 Kevin Jahns <kevin.jahns@protonmail.com>.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

146
node_modules/y-websocket/README.md generated vendored Normal file
View File

@@ -0,0 +1,146 @@
# 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.
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
```
### Start a y-websocket server
This repository implements a basic server that you can adopt to your specific use-case. [(source code)](./bin/)
Start a y-websocket server:
```sh
HOST=localhost PORT=1234 npx y-websocket
```
### Client Code:
```js
import * as Y from 'yjs'
import { WebsocketProvider } from 'y-websocket'
const doc = new Y.Doc()
const wsProvider = new WebsocketProvider('ws://localhost:1234', 'my-roomname', doc)
wsProvider.on('status', event => {
console.log(event.status) // logs "connected" or "disconnected"
})
```
#### 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'
```
<dl>
<b><code>wsProvider = new WebsocketProvider(serverUrl: string, room: string, ydoc: Y.Doc [, wsOpts: WsOpts])</code></b>
<dd>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. </dd>
</dl>
```js
wsOpts = {
// Set this to `false` if you want to connect manually using wsProvider.connect()
connect: true,
// Specify a query-string that will be url-encoded and attached to the `serverUrl`
// I.e. params = { auth: "bearer" } will be transformed to "?auth=bearer"
params: {}, // Object<string,string>
// 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
}
```
<dl>
<b><code>wsProvider.wsconnected: boolean</code></b>
<dd>True if this instance is currently connected to the server.</dd>
<b><code>wsProvider.wsconnecting: boolean</code></b>
<dd>True if this instance is currently connecting to the server.</dd>
<b><code>wsProvider.shouldConnect: boolean</code></b>
<dd>If false, the client will not try to reconnect.</dd>
<b><code>wsProvider.bcconnected: boolean</code></b>
<dd>True if this instance is currently communicating to other browser-windows via BroadcastChannel.</dd>
<b><code>wsProvider.synced: boolean</code></b>
<dd>True if this instance is currently connected and synced with the server.</dd>
<b><code>wsProvider.disconnect()</code></b>
<dd>Disconnect from the server and don't try to reconnect.</dd>
<b><code>wsProvider.connect()</code></b>
<dd>Establish a websocket connection to the websocket-server. Call this if you recently disconnected or if you set wsOpts.connect = false.</dd>
<b><code>wsProvider.destroy()</code></b>
<dd>Destroy this wsProvider instance. Disconnects from the server and removes all event handlers.</dd>
<b><code>wsProvider.on('sync', function(isSynced: boolean))</code></b>
<dd>Add an event listener for the sync event that is fired when the client received content from the server.</dd>
<b><code>wsProvider.on('status', function({ status: 'disconnected' | 'connecting' | 'connected' }))</code></b>
<dd>Receive updates about the current connection status.</dd>
<b><code>wsProvider.on('connection-close', function(WSClosedEvent))</code></b>
<dd>Fires when the underlying websocket connection is closed. It forwards the websocket event to this event handler.</dd>
<b><code>wsProvider.on('connection-error', function(WSErrorEvent))</code></b>
<dd>Fires when the underlying websocket connection closes with an error. It forwards the websocket event to this event handler.</dd>
</dl>
## Websocket Server
Start a y-websocket server:
```sh
HOST=localhost PORT=1234 npx y-websocket
```
Since npm symlinks the `y-websocket` executable from your local `./node_modules/.bin` folder, you can simply run npx. The `PORT` environment variable already defaults to 1234, and `HOST` defaults to `localhost`.
### Websocket Server with Persistence
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.js
```
### Websocket Server with HTTP callback
Send a debounced callback to an HTTP server (`POST`) on document update. Note that this implementation doesn't implement a retry logic in case the `CALLBACK_URL` does not work.
Can take the following ENV variables:
* `CALLBACK_URL` : Callback server URL
* `CALLBACK_DEBOUNCE_WAIT` : Debounce time between callbacks (in ms). Defaults to 2000 ms
* `CALLBACK_DEBOUNCE_MAXWAIT` : Maximum time to wait before callback. Defaults to 10 seconds
* `CALLBACK_TIMEOUT` : Timeout for the HTTP call. Defaults to 5 seconds
* `CALLBACK_OBJECTS` : JSON of shared objects to get data (`'{"SHARED_OBJECT_NAME":"SHARED_OBJECT_TYPE}'`)
```sh
CALLBACK_URL=http://localhost:3000/ CALLBACK_OBJECTS='{"prosemirror":"XmlFragment"}' npm start
```
This sends a debounced callback to `localhost:3000` 2 seconds after receiving an update (default `DEBOUNCE_WAIT`) with the data of an XmlFragment named `"prosemirror"` in the body.
## License
[The MIT License](./LICENSE) © Kevin Jahns

76
node_modules/y-websocket/bin/callback.js generated vendored Normal file
View File

@@ -0,0 +1,76 @@
const http = require('http')
const CALLBACK_URL = process.env.CALLBACK_URL ? new URL(process.env.CALLBACK_URL) : null
const CALLBACK_TIMEOUT = process.env.CALLBACK_TIMEOUT || 5000
const CALLBACK_OBJECTS = process.env.CALLBACK_OBJECTS ? JSON.parse(process.env.CALLBACK_OBJECTS) : {}
exports.isCallbackSet = !!CALLBACK_URL
/**
* @param {Uint8Array} update
* @param {any} origin
* @param {WSSharedDoc} doc
*/
exports.callbackHandler = (update, origin, doc) => {
const room = doc.name
const dataToSend = {
room,
data: {}
}
const sharedObjectList = Object.keys(CALLBACK_OBJECTS)
sharedObjectList.forEach(sharedObjectName => {
const sharedObjectType = CALLBACK_OBJECTS[sharedObjectName]
dataToSend.data[sharedObjectName] = {
type: sharedObjectType,
content: getContent(sharedObjectName, sharedObjectType, doc).toJSON()
}
})
callbackRequest(CALLBACK_URL, CALLBACK_TIMEOUT, dataToSend)
}
/**
* @param {URL} url
* @param {number} timeout
* @param {Object} data
*/
const callbackRequest = (url, timeout, data) => {
data = JSON.stringify(data)
const options = {
hostname: url.hostname,
port: url.port,
path: url.pathname,
timeout,
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Content-Length': data.length
}
}
const req = http.request(options)
req.on('timeout', () => {
console.warn('Callback request timed out.')
req.abort()
})
req.on('error', (e) => {
console.error('Callback request error.', e)
req.abort()
})
req.write(data)
req.end()
}
/**
* @param {string} objName
* @param {string} objType
* @param {WSSharedDoc} doc
*/
const getContent = (objName, objType, doc) => {
switch (objType) {
case 'Array': return doc.getArray(objName)
case 'Map': return doc.getMap(objName)
case 'Text': return doc.getText(objName)
case 'XmlFragment': return doc.getXmlFragment(objName)
case 'XmlElement': return doc.getXmlElement(objName)
default : return {}
}
}

35
node_modules/y-websocket/bin/server.js generated vendored Executable file
View File

@@ -0,0 +1,35 @@
#!/usr/bin/env node
/**
* @type {any}
*/
const WebSocket = require('ws')
const http = require('http')
const wss = new WebSocket.Server({ noServer: true })
const setupWSConnection = require('./utils.js').setupWSConnection
const host = process.env.HOST || 'localhost'
const port = process.env.PORT || 1234
const server = http.createServer((request, response) => {
response.writeHead(200, { 'Content-Type': 'text/plain' })
response.end('okay')
})
wss.on('connection', setupWSConnection)
server.on('upgrade', (request, socket, head) => {
// You may check auth of request here..
// See https://github.com/websockets/ws#client-authentication
/**
* @param {any} ws
*/
const handleAuth = ws => {
wss.emit('connection', ws, request)
}
wss.handleUpgrade(request, socket, head, handleAuth)
})
server.listen(port, host, () => {
console.log(`running at '${host}' on port ${port}`)
})

286
node_modules/y-websocket/bin/utils.js generated vendored Normal file
View File

@@ -0,0 +1,286 @@
const Y = require('yjs')
const syncProtocol = require('y-protocols/dist/sync.cjs')
const awarenessProtocol = require('y-protocols/dist/awareness.cjs')
const encoding = require('lib0/dist/encoding.cjs')
const decoding = require('lib0/dist/decoding.cjs')
const map = require('lib0/dist/map.cjs')
const debounce = require('lodash.debounce')
const callbackHandler = require('./callback.js').callbackHandler
const isCallbackSet = require('./callback.js').isCallbackSet
const CALLBACK_DEBOUNCE_WAIT = parseInt(process.env.CALLBACK_DEBOUNCE_WAIT) || 2000
const CALLBACK_DEBOUNCE_MAXWAIT = parseInt(process.env.CALLBACK_DEBOUNCE_MAXWAIT) || 10000
const wsReadyStateConnecting = 0
const wsReadyStateOpen = 1
const wsReadyStateClosing = 2 // eslint-disable-line
const wsReadyStateClosed = 3 // eslint-disable-line
// disable gc when using snapshots!
const gcEnabled = process.env.GC !== 'false' && process.env.GC !== '0'
const persistenceDir = process.env.YPERSISTENCE
/**
* @type {{bindState: function(string,WSSharedDoc):void, writeState:function(string,WSSharedDoc):Promise<any>, provider: any}|null}
*/
let persistence = null
if (typeof persistenceDir === 'string') {
console.info('Persisting documents to "' + persistenceDir + '"')
// @ts-ignore
const LeveldbPersistence = require('y-leveldb').LeveldbPersistence
const ldb = new LeveldbPersistence(persistenceDir)
persistence = {
provider: ldb,
bindState: async (docName, ydoc) => {
const persistedYdoc = await ldb.getYDoc(docName)
const newUpdates = Y.encodeStateAsUpdate(ydoc)
ldb.storeUpdate(docName, newUpdates)
Y.applyUpdate(ydoc, Y.encodeStateAsUpdate(persistedYdoc))
ydoc.on('update', update => {
ldb.storeUpdate(docName, update)
})
},
writeState: async (docName, ydoc) => {}
}
}
/**
* @param {{bindState: function(string,WSSharedDoc):void,
* writeState:function(string,WSSharedDoc):Promise<any>,provider:any}|null} persistence_
*/
exports.setPersistence = persistence_ => {
persistence = persistence_
}
/**
* @return {null|{bindState: function(string,WSSharedDoc):void,
* writeState:function(string,WSSharedDoc):Promise<any>}|null} used persistence layer
*/
exports.getPersistence = () => persistence
/**
* @type {Map<string,WSSharedDoc>}
*/
const docs = new Map()
// exporting docs so that others can use it
exports.docs = docs
const messageSync = 0
const messageAwareness = 1
// const messageAuth = 2
/**
* @param {Uint8Array} update
* @param {any} origin
* @param {WSSharedDoc} doc
*/
const updateHandler = (update, origin, doc) => {
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, messageSync)
syncProtocol.writeUpdate(encoder, update)
const message = encoding.toUint8Array(encoder)
doc.conns.forEach((_, conn) => send(doc, conn, message))
}
class WSSharedDoc extends Y.Doc {
/**
* @param {string} name
*/
constructor (name) {
super({ gc: gcEnabled })
this.name = name
/**
* Maps from conn to set of controlled user ids. Delete all user ids from awareness when this conn is closed
* @type {Map<Object, Set<number>>}
*/
this.conns = new Map()
/**
* @type {awarenessProtocol.Awareness}
*/
this.awareness = new awarenessProtocol.Awareness(this)
this.awareness.setLocalState(null)
/**
* @param {{ added: Array<number>, updated: Array<number>, removed: Array<number> }} changes
* @param {Object | null} conn Origin is the connection that made the change
*/
const awarenessChangeHandler = ({ added, updated, removed }, conn) => {
const changedClients = added.concat(updated, removed)
if (conn !== null) {
const connControlledIDs = /** @type {Set<number>} */ (this.conns.get(conn))
if (connControlledIDs !== undefined) {
added.forEach(clientID => { connControlledIDs.add(clientID) })
removed.forEach(clientID => { connControlledIDs.delete(clientID) })
}
}
// broadcast awareness update
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, messageAwareness)
encoding.writeVarUint8Array(encoder, awarenessProtocol.encodeAwarenessUpdate(this.awareness, changedClients))
const buff = encoding.toUint8Array(encoder)
this.conns.forEach((_, c) => {
send(this, c, buff)
})
}
this.awareness.on('update', awarenessChangeHandler)
this.on('update', updateHandler)
if (isCallbackSet) {
this.on('update', debounce(
callbackHandler,
CALLBACK_DEBOUNCE_WAIT,
{ maxWait: CALLBACK_DEBOUNCE_MAXWAIT }
))
}
}
}
/**
* Gets a Y.Doc by name, whether in memory or on disk
*
* @param {string} docname - the name of the Y.Doc to find or create
* @param {boolean} gc - whether to allow gc on the doc (applies only when created)
* @return {WSSharedDoc}
*/
const getYDoc = (docname, gc = true) => map.setIfUndefined(docs, docname, () => {
const doc = new WSSharedDoc(docname)
doc.gc = gc
if (persistence !== null) {
persistence.bindState(docname, doc)
}
docs.set(docname, doc)
return doc
})
exports.getYDoc = getYDoc
/**
* @param {any} conn
* @param {WSSharedDoc} doc
* @param {Uint8Array} message
*/
const messageListener = (conn, doc, message) => {
try {
const encoder = encoding.createEncoder()
const decoder = decoding.createDecoder(message)
const messageType = decoding.readVarUint(decoder)
switch (messageType) {
case messageSync:
encoding.writeVarUint(encoder, messageSync)
syncProtocol.readSyncMessage(decoder, encoder, doc, conn)
// If the `encoder` only contains the type of reply message and no
// message, there is no need to send the message. When `encoder` only
// contains the type of reply, its length is 1.
if (encoding.length(encoder) > 1) {
send(doc, conn, encoding.toUint8Array(encoder))
}
break
case messageAwareness: {
awarenessProtocol.applyAwarenessUpdate(doc.awareness, decoding.readVarUint8Array(decoder), conn)
break
}
}
} catch (err) {
console.error(err)
doc.emit('error', [err])
}
}
/**
* @param {WSSharedDoc} doc
* @param {any} conn
*/
const closeConn = (doc, conn) => {
if (doc.conns.has(conn)) {
/**
* @type {Set<number>}
*/
// @ts-ignore
const controlledIds = doc.conns.get(conn)
doc.conns.delete(conn)
awarenessProtocol.removeAwarenessStates(doc.awareness, Array.from(controlledIds), null)
if (doc.conns.size === 0 && persistence !== null) {
// if persisted, we store state and destroy ydocument
persistence.writeState(doc.name, doc).then(() => {
doc.destroy()
})
docs.delete(doc.name)
}
}
conn.close()
}
/**
* @param {WSSharedDoc} doc
* @param {any} conn
* @param {Uint8Array} m
*/
const send = (doc, conn, m) => {
if (conn.readyState !== wsReadyStateConnecting && conn.readyState !== wsReadyStateOpen) {
closeConn(doc, conn)
}
try {
conn.send(m, /** @param {any} err */ err => { err != null && closeConn(doc, conn) })
} catch (e) {
closeConn(doc, conn)
}
}
const pingTimeout = 30000
/**
* @param {any} conn
* @param {any} req
* @param {any} opts
*/
exports.setupWSConnection = (conn, req, { docName = req.url.slice(1).split('?')[0], gc = true } = {}) => {
conn.binaryType = 'arraybuffer'
// get doc, initialize if it does not exist yet
const doc = getYDoc(docName, gc)
doc.conns.set(conn, new Set())
// listen and reply to events
conn.on('message', /** @param {ArrayBuffer} message */ message => messageListener(conn, doc, new Uint8Array(message)))
// Check if connection is still alive
let pongReceived = true
const pingInterval = setInterval(() => {
if (!pongReceived) {
if (doc.conns.has(conn)) {
closeConn(doc, conn)
}
clearInterval(pingInterval)
} else if (doc.conns.has(conn)) {
pongReceived = false
try {
conn.ping()
} catch (e) {
closeConn(doc, conn)
clearInterval(pingInterval)
}
}
}, pingTimeout)
conn.on('close', () => {
closeConn(doc, conn)
clearInterval(pingInterval)
})
conn.on('pong', () => {
pongReceived = true
})
// put the following in a variables in a block so the interval handlers don't keep in in
// scope
{
// send sync step 1
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, messageSync)
syncProtocol.writeSyncStep1(encoder, doc)
send(doc, conn, encoding.toUint8Array(encoder))
const awarenessStates = doc.awareness.getStates()
if (awarenessStates.size > 0) {
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, messageAwareness)
encoding.writeVarUint8Array(encoder, awarenessProtocol.encodeAwarenessUpdate(doc.awareness, Array.from(awarenessStates.keys())))
send(doc, conn, encoding.toUint8Array(encoder))
}
}
}

120
node_modules/y-websocket/dist/src/y-websocket.d.ts generated vendored Normal file
View File

@@ -0,0 +1,120 @@
export const messageSync: 0;
export const messageQueryAwareness: 3;
export const messageAwareness: 1;
export const messageAuth: 2;
/**
* 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 {Observable<string>}
*/
export class WebsocketProvider extends Observable<string> {
/**
* @param {string} serverUrl
* @param {string} roomname
* @param {Y.Doc} doc
* @param {object} opts
* @param {boolean} [opts.connect]
* @param {awarenessProtocol.Awareness} [opts.awareness]
* @param {Object<string,string>} [opts.params]
* @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: string, roomname: string, doc: Y.Doc, { connect, awareness, params, WebSocketPolyfill, resyncInterval, maxBackoffTime, disableBc }?: {
connect?: boolean | undefined;
awareness?: awarenessProtocol.Awareness | undefined;
params?: {
[x: string]: string;
} | undefined;
WebSocketPolyfill?: {
new (url: string | URL, protocols?: string | string[] | undefined): WebSocket;
prototype: WebSocket;
readonly CLOSED: number;
readonly CLOSING: number;
readonly CONNECTING: number;
readonly OPEN: number;
} | undefined;
resyncInterval?: number | undefined;
maxBackoffTime?: number | undefined;
disableBc?: boolean | undefined;
});
maxBackoffTime: number;
bcChannel: string;
url: string;
roomname: string;
doc: Y.Doc;
_WS: {
new (url: string | URL, protocols?: string | string[] | undefined): WebSocket;
prototype: WebSocket;
readonly CLOSED: number;
readonly CLOSING: number;
readonly CONNECTING: number;
readonly OPEN: number;
};
awareness: awarenessProtocol.Awareness;
wsconnected: boolean;
wsconnecting: boolean;
bcconnected: boolean;
disableBc: boolean;
wsUnsuccessfulReconnects: number;
messageHandlers: ((arg0: encoding.Encoder, arg1: decoding.Decoder, arg2: WebsocketProvider, arg3: boolean, arg4: number) => void)[];
/**
* @type {boolean}
*/
_synced: boolean;
/**
* @type {WebSocket?}
*/
ws: WebSocket | null;
wsLastMessageReceived: number;
/**
* Whether to connect to other peers or not
* @type {boolean}
*/
shouldConnect: boolean;
/**
* @type {number}
*/
_resyncInterval: number;
/**
* @param {ArrayBuffer} data
* @param {any} origin
*/
_bcSubscriber: (data: ArrayBuffer, origin: any) => void;
/**
* Listens to Yjs updates and sends them to remote peers (ws and broadcastchannel)
* @param {Uint8Array} update
* @param {any} origin
*/
_updateHandler: (update: Uint8Array, origin: any) => void;
/**
* @param {any} changed
* @param {any} _origin
*/
_awarenessUpdateHandler: ({ added, updated, removed }: any, _origin: any) => void;
_exitHandler: () => void;
_checkInterval: any;
set synced(arg: boolean);
/**
* @type {boolean}
*/
get synced(): boolean;
connectBc(): void;
disconnectBc(): void;
disconnect(): void;
connect(): void;
}
import { Observable } from "lib0/observable";
import * as Y from "yjs";
import * as awarenessProtocol from "y-protocols/awareness";
import * as encoding from "lib0/encoding";
import * as decoding from "lib0/decoding";

520
node_modules/y-websocket/dist/y-websocket.cjs generated vendored Normal file
View File

@@ -0,0 +1,520 @@
'use strict';
require('yjs');
var bc = require('lib0/dist/broadcastchannel.cjs');
var time = require('lib0/dist/time.cjs');
var encoding = require('lib0/dist/encoding.cjs');
var decoding = require('lib0/dist/decoding.cjs');
var syncProtocol = require('y-protocols/dist/sync.cjs');
var authProtocol = require('y-protocols/dist/auth.cjs');
var awarenessProtocol = require('y-protocols/dist/awareness.cjs');
var observable = require('lib0/dist/observable.cjs');
var math = require('lib0/dist/math.cjs');
var url = require('lib0/dist/url.cjs');
var env = require('lib0/dist/environment.cjs');
function _interopNamespaceDefault(e) {
var n = Object.create(null);
if (e) {
Object.keys(e).forEach(function (k) {
if (k !== 'default') {
var d = Object.getOwnPropertyDescriptor(e, k);
Object.defineProperty(n, k, d.get ? d : {
enumerable: true,
get: function () { return e[k]; }
});
}
});
}
n.default = e;
return Object.freeze(n);
}
var bc__namespace = /*#__PURE__*/_interopNamespaceDefault(bc);
var time__namespace = /*#__PURE__*/_interopNamespaceDefault(time);
var encoding__namespace = /*#__PURE__*/_interopNamespaceDefault(encoding);
var decoding__namespace = /*#__PURE__*/_interopNamespaceDefault(decoding);
var syncProtocol__namespace = /*#__PURE__*/_interopNamespaceDefault(syncProtocol);
var authProtocol__namespace = /*#__PURE__*/_interopNamespaceDefault(authProtocol);
var awarenessProtocol__namespace = /*#__PURE__*/_interopNamespaceDefault(awarenessProtocol);
var math__namespace = /*#__PURE__*/_interopNamespaceDefault(math);
var url__namespace = /*#__PURE__*/_interopNamespaceDefault(url);
var env__namespace = /*#__PURE__*/_interopNamespaceDefault(env);
/**
* @module provider/websocket
*/
const messageSync = 0;
const messageQueryAwareness = 3;
const messageAwareness = 1;
const messageAuth = 2;
/**
* encoder, decoder, provider, emitSynced, messageType
* @type {Array<function(encoding.Encoder, decoding.Decoder, WebsocketProvider, boolean, number):void>}
*/
const messageHandlers = [];
messageHandlers[messageSync] = (
encoder,
decoder,
provider,
emitSynced,
_messageType
) => {
encoding__namespace.writeVarUint(encoder, messageSync);
const syncMessageType = syncProtocol__namespace.readSyncMessage(
decoder,
encoder,
provider.doc,
provider
);
if (
emitSynced && syncMessageType === syncProtocol__namespace.messageYjsSyncStep2 &&
!provider.synced
) {
provider.synced = true;
}
};
messageHandlers[messageQueryAwareness] = (
encoder,
_decoder,
provider,
_emitSynced,
_messageType
) => {
encoding__namespace.writeVarUint(encoder, messageAwareness);
encoding__namespace.writeVarUint8Array(
encoder,
awarenessProtocol__namespace.encodeAwarenessUpdate(
provider.awareness,
Array.from(provider.awareness.getStates().keys())
)
);
};
messageHandlers[messageAwareness] = (
_encoder,
decoder,
provider,
_emitSynced,
_messageType
) => {
awarenessProtocol__namespace.applyAwarenessUpdate(
provider.awareness,
decoding__namespace.readVarUint8Array(decoder),
provider
);
};
messageHandlers[messageAuth] = (
_encoder,
decoder,
provider,
_emitSynced,
_messageType
) => {
authProtocol__namespace.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__namespace.createDecoder(buf);
const encoder = encoding__namespace.createEncoder();
const messageType = decoding__namespace.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
};
/**
* @param {WebsocketProvider} provider
*/
const setupWS = (provider) => {
if (provider.shouldConnect && provider.ws === null) {
const websocket = new provider._WS(provider.url);
websocket.binaryType = 'arraybuffer';
provider.ws = websocket;
provider.wsconnecting = true;
provider.wsconnected = false;
provider.synced = false;
websocket.onmessage = (event) => {
provider.wsLastMessageReceived = time__namespace.getUnixTime();
const encoder = readMessage(provider, new Uint8Array(event.data), true);
if (encoding__namespace.length(encoder) > 1) {
websocket.send(encoding__namespace.toUint8Array(encoder));
}
};
websocket.onerror = (event) => {
provider.emit('connection-error', [event, provider]);
};
websocket.onclose = (event) => {
provider.emit('connection-close', [event, provider]);
provider.ws = null;
provider.wsconnecting = false;
if (provider.wsconnected) {
provider.wsconnected = false;
provider.synced = false;
// update awareness (all users except local left)
awarenessProtocol__namespace.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__namespace.min(
math__namespace.pow(2, provider.wsUnsuccessfulReconnects) * 100,
provider.maxBackoffTime
),
provider
);
};
websocket.onopen = () => {
provider.wsLastMessageReceived = time__namespace.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__namespace.createEncoder();
encoding__namespace.writeVarUint(encoder, messageSync);
syncProtocol__namespace.writeSyncStep1(encoder, provider.doc);
websocket.send(encoding__namespace.toUint8Array(encoder));
// broadcast local awareness state
if (provider.awareness.getLocalState() !== null) {
const encoderAwarenessState = encoding__namespace.createEncoder();
encoding__namespace.writeVarUint(encoderAwarenessState, messageAwareness);
encoding__namespace.writeVarUint8Array(
encoderAwarenessState,
awarenessProtocol__namespace.encodeAwarenessUpdate(provider.awareness, [
provider.doc.clientID
])
);
websocket.send(encoding__namespace.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__namespace.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 {Observable<string>}
*/
class WebsocketProvider extends observable.Observable {
/**
* @param {string} serverUrl
* @param {string} roomname
* @param {Y.Doc} doc
* @param {object} opts
* @param {boolean} [opts.connect]
* @param {awarenessProtocol.Awareness} [opts.awareness]
* @param {Object<string,string>} [opts.params]
* @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__namespace.Awareness(doc),
params = {},
WebSocketPolyfill = WebSocket,
resyncInterval = -1,
maxBackoffTime = 2500,
disableBc = false
} = {}) {
super();
// ensure that url is always ends with /
while (serverUrl[serverUrl.length - 1] === '/') {
serverUrl = serverUrl.slice(0, serverUrl.length - 1);
}
const encodedParams = url__namespace.encodeQueryParams(params);
this.maxBackoffTime = maxBackoffTime;
this.bcChannel = serverUrl + '/' + roomname;
this.url = serverUrl + '/' + roomname +
(encodedParams.length === 0 ? '' : '?' + encodedParams);
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__namespace.createEncoder();
encoding__namespace.writeVarUint(encoder, messageSync);
syncProtocol__namespace.writeSyncStep1(encoder, doc);
this.ws.send(encoding__namespace.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__namespace.length(encoder) > 1) {
bc__namespace.publish(this.bcChannel, encoding__namespace.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__namespace.createEncoder();
encoding__namespace.writeVarUint(encoder, messageSync);
syncProtocol__namespace.writeUpdate(encoder, update);
broadcastMessage(this, encoding__namespace.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__namespace.createEncoder();
encoding__namespace.writeVarUint(encoder, messageAwareness);
encoding__namespace.writeVarUint8Array(
encoder,
awarenessProtocol__namespace.encodeAwarenessUpdate(awareness, changedClients)
);
broadcastMessage(this, encoding__namespace.toUint8Array(encoder));
};
this._exitHandler = () => {
awarenessProtocol__namespace.removeAwarenessStates(
this.awareness,
[doc.clientID],
'app closed'
);
};
if (env__namespace.isNode && typeof process !== 'undefined') {
process.on('exit', this._exitHandler);
}
awareness.on('update', this._awarenessUpdateHandler);
this._checkInterval = /** @type {any} */ (setInterval(() => {
if (
this.wsconnected &&
messageReconnectTimeout <
time__namespace.getUnixTime() - this.wsLastMessageReceived
) {
// no message received in a long time - not even your own awareness
// updates (which are updated every 15 seconds)
/** @type {WebSocket} */ (this.ws).close();
}
}, messageReconnectTimeout / 10));
if (connect) {
this.connect();
}
}
/**
* @type {boolean}
*/
get synced () {
return this._synced
}
set synced (state) {
if (this._synced !== state) {
this._synced = state;
this.emit('synced', [state]);
this.emit('sync', [state]);
}
}
destroy () {
if (this._resyncInterval !== 0) {
clearInterval(this._resyncInterval);
}
clearInterval(this._checkInterval);
this.disconnect();
if (env__namespace.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__namespace.subscribe(this.bcChannel, this._bcSubscriber);
this.bcconnected = true;
}
// send sync step1 to bc
// write sync step 1
const encoderSync = encoding__namespace.createEncoder();
encoding__namespace.writeVarUint(encoderSync, messageSync);
syncProtocol__namespace.writeSyncStep1(encoderSync, this.doc);
bc__namespace.publish(this.bcChannel, encoding__namespace.toUint8Array(encoderSync), this);
// broadcast local state
const encoderState = encoding__namespace.createEncoder();
encoding__namespace.writeVarUint(encoderState, messageSync);
syncProtocol__namespace.writeSyncStep2(encoderState, this.doc);
bc__namespace.publish(this.bcChannel, encoding__namespace.toUint8Array(encoderState), this);
// write queryAwareness
const encoderAwarenessQuery = encoding__namespace.createEncoder();
encoding__namespace.writeVarUint(encoderAwarenessQuery, messageQueryAwareness);
bc__namespace.publish(
this.bcChannel,
encoding__namespace.toUint8Array(encoderAwarenessQuery),
this
);
// broadcast local awareness state
const encoderAwarenessState = encoding__namespace.createEncoder();
encoding__namespace.writeVarUint(encoderAwarenessState, messageAwareness);
encoding__namespace.writeVarUint8Array(
encoderAwarenessState,
awarenessProtocol__namespace.encodeAwarenessUpdate(this.awareness, [
this.doc.clientID
])
);
bc__namespace.publish(
this.bcChannel,
encoding__namespace.toUint8Array(encoderAwarenessState),
this
);
}
disconnectBc () {
// broadcast message with local awareness state set to null (indicating disconnect)
const encoder = encoding__namespace.createEncoder();
encoding__namespace.writeVarUint(encoder, messageAwareness);
encoding__namespace.writeVarUint8Array(
encoder,
awarenessProtocol__namespace.encodeAwarenessUpdate(this.awareness, [
this.doc.clientID
], new Map())
);
broadcastMessage(this, encoding__namespace.toUint8Array(encoder));
if (this.bcconnected) {
bc__namespace.unsubscribe(this.bcChannel, this._bcSubscriber);
this.bcconnected = false;
}
}
disconnect () {
this.shouldConnect = false;
this.disconnectBc();
if (this.ws !== null) {
this.ws.close();
}
}
connect () {
this.shouldConnect = true;
if (!this.wsconnected && this.ws === null) {
setupWS(this);
this.connectBc();
}
}
}
exports.WebsocketProvider = WebsocketProvider;
exports.messageAuth = messageAuth;
exports.messageAwareness = messageAwareness;
exports.messageQueryAwareness = messageQueryAwareness;
exports.messageSync = messageSync;
//# sourceMappingURL=y-websocket.cjs.map

1
node_modules/y-websocket/dist/y-websocket.cjs.map generated vendored Normal file

File diff suppressed because one or more lines are too long

21
node_modules/y-websocket/node_modules/ws/LICENSE generated vendored Normal file
View File

@@ -0,0 +1,21 @@
The MIT License (MIT)
Copyright (c) 2011 Einar Otto Stangvik <einaros@gmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

449
node_modules/y-websocket/node_modules/ws/README.md generated vendored Normal file
View File

@@ -0,0 +1,449 @@
# ws: a Node.js WebSocket library
[![Version npm](https://img.shields.io/npm/v/ws.svg?logo=npm)](https://www.npmjs.com/package/ws)
[![Linux Build](https://img.shields.io/travis/websockets/ws/master.svg?logo=travis)](https://travis-ci.org/websockets/ws)
[![Windows Build](https://img.shields.io/appveyor/ci/lpinca/ws/master.svg?logo=appveyor)](https://ci.appveyor.com/project/lpinca/ws)
[![Coverage Status](https://img.shields.io/coveralls/websockets/ws/master.svg)](https://coveralls.io/github/websockets/ws)
ws is a simple to use, blazing fast, and thoroughly tested WebSocket client and
server implementation.
Passes the quite extensive Autobahn test suite: [server][server-report],
[client][client-report].
**Note**: This module does not work in the browser. The client in the docs is a
reference to a back end with the role of a client in the WebSocket
communication. Browser clients must use the native
[`WebSocket`](https://developer.mozilla.org/en-US/docs/Web/API/WebSocket)
object. To make the same code work seamlessly on Node.js and the browser, you
can use one of the many wrappers available on npm, like
[isomorphic-ws](https://github.com/heineiuo/isomorphic-ws).
## Table of Contents
- [Protocol support](#protocol-support)
- [Installing](#installing)
- [Opt-in for performance and spec compliance](#opt-in-for-performance-and-spec-compliance)
- [API docs](#api-docs)
- [WebSocket compression](#websocket-compression)
- [Usage examples](#usage-examples)
- [Sending and receiving text data](#sending-and-receiving-text-data)
- [Sending binary data](#sending-binary-data)
- [Simple server](#simple-server)
- [External HTTP/S server](#external-https-server)
- [Multiple servers sharing a single HTTP/S server](#multiple-servers-sharing-a-single-https-server)
- [Server broadcast](#server-broadcast)
- [echo.websocket.org demo](#echowebsocketorg-demo)
- [Other examples](#other-examples)
- [Error handling best practices](#error-handling-best-practices)
- [FAQ](#faq)
- [How to get the IP address of the client?](#how-to-get-the-ip-address-of-the-client)
- [How to detect and close broken connections?](#how-to-detect-and-close-broken-connections)
- [How to connect via a proxy?](#how-to-connect-via-a-proxy)
- [Changelog](#changelog)
- [License](#license)
## Protocol support
- **HyBi drafts 07-12** (Use the option `protocolVersion: 8`)
- **HyBi drafts 13-17** (Current default, alternatively option
`protocolVersion: 13`)
## Installing
```
npm install ws
```
### Opt-in for performance and spec compliance
There are 2 optional modules that can be installed along side with the ws
module. These modules are binary addons which improve certain operations.
Prebuilt binaries are available for the most popular platforms so you don't
necessarily need to have a C++ compiler installed on your machine.
- `npm install --save-optional bufferutil`: Allows to efficiently perform
operations such as masking and unmasking the data payload of the WebSocket
frames.
- `npm install --save-optional utf-8-validate`: Allows to efficiently check if a
message contains valid UTF-8 as required by the spec.
## API docs
See [`/doc/ws.md`](./doc/ws.md) for Node.js-like docs for the ws classes.
## WebSocket compression
ws supports the [permessage-deflate extension][permessage-deflate] which enables
the client and server to negotiate a compression algorithm and its parameters,
and then selectively apply it to the data payloads of each WebSocket message.
The extension is disabled by default on the server and enabled by default on the
client. It adds a significant overhead in terms of performance and memory
consumption so we suggest to enable it only if it is really needed.
Note that Node.js has a variety of issues with high-performance compression,
where increased concurrency, especially on Linux, can lead to [catastrophic
memory fragmentation][node-zlib-bug] and slow performance. If you intend to use
permessage-deflate in production, it is worthwhile to set up a test
representative of your workload and ensure Node.js/zlib will handle it with
acceptable performance and memory usage.
Tuning of permessage-deflate can be done via the options defined below. You can
also use `zlibDeflateOptions` and `zlibInflateOptions`, which is passed directly
into the creation of [raw deflate/inflate streams][node-zlib-deflaterawdocs].
See [the docs][ws-server-options] for more options.
```js
const WebSocket = require('ws');
const wss = new WebSocket.Server({
port: 8080,
perMessageDeflate: {
zlibDeflateOptions: {
// See zlib defaults.
chunkSize: 1024,
memLevel: 7,
level: 3
},
zlibInflateOptions: {
chunkSize: 10 * 1024
},
// Other options settable:
clientNoContextTakeover: true, // Defaults to negotiated value.
serverNoContextTakeover: true, // Defaults to negotiated value.
serverMaxWindowBits: 10, // Defaults to negotiated value.
// Below options specified as default values.
concurrencyLimit: 10, // Limits zlib concurrency for perf.
threshold: 1024 // Size (in bytes) below which messages
// should not be compressed.
}
});
```
The client will only use the extension if it is supported and enabled on the
server. To always disable the extension on the client set the
`perMessageDeflate` option to `false`.
```js
const WebSocket = require('ws');
const ws = new WebSocket('ws://www.host.com/path', {
perMessageDeflate: false
});
```
## Usage examples
### Sending and receiving text data
```js
const WebSocket = require('ws');
const ws = new WebSocket('ws://www.host.com/path');
ws.on('open', function open() {
ws.send('something');
});
ws.on('message', function incoming(data) {
console.log(data);
});
```
### Sending binary data
```js
const WebSocket = require('ws');
const ws = new WebSocket('ws://www.host.com/path');
ws.on('open', function open() {
const array = new Float32Array(5);
for (var i = 0; i < array.length; ++i) {
array[i] = i / 2;
}
ws.send(array);
});
```
### Simple server
```js
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });
wss.on('connection', function connection(ws) {
ws.on('message', function incoming(message) {
console.log('received: %s', message);
});
ws.send('something');
});
```
### External HTTP/S server
```js
const fs = require('fs');
const https = require('https');
const WebSocket = require('ws');
const server = new https.createServer({
cert: fs.readFileSync('/path/to/cert.pem'),
key: fs.readFileSync('/path/to/key.pem')
});
const wss = new WebSocket.Server({ server });
wss.on('connection', function connection(ws) {
ws.on('message', function incoming(message) {
console.log('received: %s', message);
});
ws.send('something');
});
server.listen(8080);
```
### Multiple servers sharing a single HTTP/S server
```js
const http = require('http');
const WebSocket = require('ws');
const server = http.createServer();
const wss1 = new WebSocket.Server({ noServer: true });
const wss2 = new WebSocket.Server({ noServer: true });
wss1.on('connection', function connection(ws) {
// ...
});
wss2.on('connection', function connection(ws) {
// ...
});
server.on('upgrade', function upgrade(request, socket, head) {
const pathname = url.parse(request.url).pathname;
if (pathname === '/foo') {
wss1.handleUpgrade(request, socket, head, function done(ws) {
wss1.emit('connection', ws, request);
});
} else if (pathname === '/bar') {
wss2.handleUpgrade(request, socket, head, function done(ws) {
wss2.emit('connection', ws, request);
});
} else {
socket.destroy();
}
});
server.listen(8080);
```
### Server broadcast
```js
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });
// Broadcast to all.
wss.broadcast = function broadcast(data) {
wss.clients.forEach(function each(client) {
if (client.readyState === WebSocket.OPEN) {
client.send(data);
}
});
};
wss.on('connection', function connection(ws) {
ws.on('message', function incoming(data) {
// Broadcast to everyone else.
wss.clients.forEach(function each(client) {
if (client !== ws && client.readyState === WebSocket.OPEN) {
client.send(data);
}
});
});
});
```
### echo.websocket.org demo
```js
const WebSocket = require('ws');
const ws = new WebSocket('wss://echo.websocket.org/', {
origin: 'https://websocket.org'
});
ws.on('open', function open() {
console.log('connected');
ws.send(Date.now());
});
ws.on('close', function close() {
console.log('disconnected');
});
ws.on('message', function incoming(data) {
console.log(`Roundtrip time: ${Date.now() - data} ms`);
setTimeout(function timeout() {
ws.send(Date.now());
}, 500);
});
```
### Other examples
For a full example with a browser client communicating with a ws server, see the
examples folder.
Otherwise, see the test cases.
## Error handling best practices
```js
// If the WebSocket is closed before the following send is attempted
ws.send('something');
// Errors (both immediate and async write errors) can be detected in an optional
// callback. The callback is also the only way of being notified that data has
// actually been sent.
ws.send('something', function ack(error) {
// If error is not defined, the send has been completed, otherwise the error
// object will indicate what failed.
});
// Immediate errors can also be handled with `try...catch`, but **note** that
// since sends are inherently asynchronous, socket write failures will *not* be
// captured when this technique is used.
try {
ws.send('something');
} catch (e) {
/* handle error */
}
```
## FAQ
### How to get the IP address of the client?
The remote IP address can be obtained from the raw socket.
```js
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });
wss.on('connection', function connection(ws, req) {
const ip = req.connection.remoteAddress;
});
```
When the server runs behind a proxy like NGINX, the de-facto standard is to use
the `X-Forwarded-For` header.
```js
wss.on('connection', function connection(ws, req) {
const ip = req.headers['x-forwarded-for'].split(/\s*,\s*/)[0];
});
```
### How to detect and close broken connections?
Sometimes the link between the server and the client can be interrupted in a way
that keeps both the server and the client unaware of the broken state of the
connection (e.g. when pulling the cord).
In these cases ping messages can be used as a means to verify that the remote
endpoint is still responsive.
```js
const WebSocket = require('ws');
function noop() {}
function heartbeat() {
this.isAlive = true;
}
const wss = new WebSocket.Server({ port: 8080 });
wss.on('connection', function connection(ws) {
ws.isAlive = true;
ws.on('pong', heartbeat);
});
const interval = setInterval(function ping() {
wss.clients.forEach(function each(ws) {
if (ws.isAlive === false) return ws.terminate();
ws.isAlive = false;
ws.ping(noop);
});
}, 30000);
```
Pong messages are automatically sent in response to ping messages as required by
the spec.
Just like the server example above your clients might as well lose connection
without knowing it. You might want to add a ping listener on your clients to
prevent that. A simple implementation would be:
```js
const WebSocket = require('ws');
function heartbeat() {
clearTimeout(this.pingTimeout);
// Use `WebSocket#terminate()` and not `WebSocket#close()`. Delay should be
// equal to the interval at which your server sends out pings plus a
// conservative assumption of the latency.
this.pingTimeout = setTimeout(() => {
this.terminate();
}, 30000 + 1000);
}
const client = new WebSocket('wss://echo.websocket.org/');
client.on('open', heartbeat);
client.on('ping', heartbeat);
client.on('close', function clear() {
clearTimeout(this.pingTimeout);
});
```
### How to connect via a proxy?
Use a custom `http.Agent` implementation like [https-proxy-agent][] or
[socks-proxy-agent][].
## Changelog
We're using the GitHub [releases][changelog] for changelog entries.
## License
[MIT](LICENSE)
[https-proxy-agent]: https://github.com/TooTallNate/node-https-proxy-agent
[socks-proxy-agent]: https://github.com/TooTallNate/node-socks-proxy-agent
[client-report]: http://websockets.github.io/ws/autobahn/clients/
[server-report]: http://websockets.github.io/ws/autobahn/servers/
[permessage-deflate]: https://tools.ietf.org/html/rfc7692
[changelog]: https://github.com/websockets/ws/releases
[node-zlib-bug]: https://github.com/nodejs/node/issues/8871
[node-zlib-deflaterawdocs]:
https://nodejs.org/api/zlib.html#zlib_zlib_createdeflateraw_options
[ws-server-options]:
https://github.com/websockets/ws/blob/master/doc/ws.md#new-websocketserveroptions-callback

8
node_modules/y-websocket/node_modules/ws/browser.js generated vendored Normal file
View File

@@ -0,0 +1,8 @@
'use strict';
module.exports = function() {
throw new Error(
'ws does not work in the browser. Browser clients must use the native ' +
'WebSocket object'
);
};

9
node_modules/y-websocket/node_modules/ws/index.js generated vendored Normal file
View File

@@ -0,0 +1,9 @@
'use strict';
const WebSocket = require('./lib/websocket');
WebSocket.Server = require('./lib/websocket-server');
WebSocket.Receiver = require('./lib/receiver');
WebSocket.Sender = require('./lib/sender');
module.exports = WebSocket;

View File

@@ -0,0 +1,144 @@
'use strict';
const { EMPTY_BUFFER } = require('./constants');
/**
* Merges an array of buffers into a new buffer.
*
* @param {Buffer[]} list The array of buffers to concat
* @param {Number} totalLength The total length of buffers in the list
* @return {Buffer} The resulting buffer
* @public
*/
function concat(list, totalLength) {
if (list.length === 0) return EMPTY_BUFFER;
if (list.length === 1) return list[0];
const target = Buffer.allocUnsafe(totalLength);
var offset = 0;
for (var i = 0; i < list.length; i++) {
const buf = list[i];
buf.copy(target, offset);
offset += buf.length;
}
return target;
}
/**
* Masks a buffer using the given mask.
*
* @param {Buffer} source The buffer to mask
* @param {Buffer} mask The mask to use
* @param {Buffer} output The buffer where to store the result
* @param {Number} offset The offset at which to start writing
* @param {Number} length The number of bytes to mask.
* @public
*/
function _mask(source, mask, output, offset, length) {
for (var i = 0; i < length; i++) {
output[offset + i] = source[i] ^ mask[i & 3];
}
}
/**
* Unmasks a buffer using the given mask.
*
* @param {Buffer} buffer The buffer to unmask
* @param {Buffer} mask The mask to use
* @public
*/
function _unmask(buffer, mask) {
// Required until https://github.com/nodejs/node/issues/9006 is resolved.
const length = buffer.length;
for (var i = 0; i < length; i++) {
buffer[i] ^= mask[i & 3];
}
}
/**
* Converts a buffer to an `ArrayBuffer`.
*
* @param {Buffer} buf The buffer to convert
* @return {ArrayBuffer} Converted buffer
* @public
*/
function toArrayBuffer(buf) {
if (buf.byteLength === buf.buffer.byteLength) {
return buf.buffer;
}
return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
}
/**
* Converts `data` to a `Buffer`.
*
* @param {*} data The data to convert
* @return {Buffer} The buffer
* @throws {TypeError}
* @public
*/
function toBuffer(data) {
toBuffer.readOnly = true;
if (Buffer.isBuffer(data)) return data;
var buf;
if (data instanceof ArrayBuffer) {
buf = Buffer.from(data);
} else if (ArrayBuffer.isView(data)) {
buf = viewToBuffer(data);
} else {
buf = Buffer.from(data);
toBuffer.readOnly = false;
}
return buf;
}
/**
* Converts an `ArrayBuffer` view into a buffer.
*
* @param {(DataView|TypedArray)} view The view to convert
* @return {Buffer} Converted view
* @private
*/
function viewToBuffer(view) {
const buf = Buffer.from(view.buffer);
if (view.byteLength !== view.buffer.byteLength) {
return buf.slice(view.byteOffset, view.byteOffset + view.byteLength);
}
return buf;
}
try {
const bufferUtil = require('bufferutil');
const bu = bufferUtil.BufferUtil || bufferUtil;
module.exports = {
concat,
mask(source, mask, output, offset, length) {
if (length < 48) _mask(source, mask, output, offset, length);
else bu.mask(source, mask, output, offset, length);
},
toArrayBuffer,
toBuffer,
unmask(buffer, mask) {
if (buffer.length < 32) _unmask(buffer, mask);
else bu.unmask(buffer, mask);
}
};
} catch (e) /* istanbul ignore next */ {
module.exports = {
concat,
mask: _mask,
toArrayBuffer,
toBuffer,
unmask: _unmask
};
}

View File

@@ -0,0 +1,10 @@
'use strict';
module.exports = {
BINARY_TYPES: ['nodebuffer', 'arraybuffer', 'fragments'],
GUID: '258EAFA5-E914-47DA-95CA-C5AB0DC85B11',
kStatusCode: Symbol('status-code'),
kWebSocket: Symbol('websocket'),
EMPTY_BUFFER: Buffer.alloc(0),
NOOP: () => {}
};

View File

@@ -0,0 +1,170 @@
'use strict';
/**
* Class representing an event.
*
* @private
*/
class Event {
/**
* Create a new `Event`.
*
* @param {String} type The name of the event
* @param {Object} target A reference to the target to which the event was dispatched
*/
constructor(type, target) {
this.target = target;
this.type = type;
}
}
/**
* Class representing a message event.
*
* @extends Event
* @private
*/
class MessageEvent extends Event {
/**
* Create a new `MessageEvent`.
*
* @param {(String|Buffer|ArrayBuffer|Buffer[])} data The received data
* @param {WebSocket} target A reference to the target to which the event was dispatched
*/
constructor(data, target) {
super('message', target);
this.data = data;
}
}
/**
* Class representing a close event.
*
* @extends Event
* @private
*/
class CloseEvent extends Event {
/**
* Create a new `CloseEvent`.
*
* @param {Number} code The status code explaining why the connection is being closed
* @param {String} reason A human-readable string explaining why the connection is closing
* @param {WebSocket} target A reference to the target to which the event was dispatched
*/
constructor(code, reason, target) {
super('close', target);
this.wasClean = target._closeFrameReceived && target._closeFrameSent;
this.reason = reason;
this.code = code;
}
}
/**
* Class representing an open event.
*
* @extends Event
* @private
*/
class OpenEvent extends Event {
/**
* Create a new `OpenEvent`.
*
* @param {WebSocket} target A reference to the target to which the event was dispatched
*/
constructor(target) {
super('open', target);
}
}
/**
* Class representing an error event.
*
* @extends Event
* @private
*/
class ErrorEvent extends Event {
/**
* Create a new `ErrorEvent`.
*
* @param {Object} error The error that generated this event
* @param {WebSocket} target A reference to the target to which the event was dispatched
*/
constructor(error, target) {
super('error', target);
this.message = error.message;
this.error = error;
}
}
/**
* This provides methods for emulating the `EventTarget` interface. It's not
* meant to be used directly.
*
* @mixin
*/
const EventTarget = {
/**
* Register an event listener.
*
* @param {String} method A string representing the event type to listen for
* @param {Function} listener The listener to add
* @public
*/
addEventListener(method, listener) {
if (typeof listener !== 'function') return;
function onMessage(data) {
listener.call(this, new MessageEvent(data, this));
}
function onClose(code, message) {
listener.call(this, new CloseEvent(code, message, this));
}
function onError(error) {
listener.call(this, new ErrorEvent(error, this));
}
function onOpen() {
listener.call(this, new OpenEvent(this));
}
if (method === 'message') {
onMessage._listener = listener;
this.on(method, onMessage);
} else if (method === 'close') {
onClose._listener = listener;
this.on(method, onClose);
} else if (method === 'error') {
onError._listener = listener;
this.on(method, onError);
} else if (method === 'open') {
onOpen._listener = listener;
this.on(method, onOpen);
} else {
this.on(method, listener);
}
},
/**
* Remove an event listener.
*
* @param {String} method A string representing the event type to remove
* @param {Function} listener The listener to remove
* @public
*/
removeEventListener(method, listener) {
const listeners = this.listeners(method);
for (var i = 0; i < listeners.length; i++) {
if (listeners[i] === listener || listeners[i]._listener === listener) {
this.removeListener(method, listeners[i]);
}
}
}
};
module.exports = EventTarget;

View File

@@ -0,0 +1,222 @@
'use strict';
//
// Allowed token characters:
//
// '!', '#', '$', '%', '&', ''', '*', '+', '-',
// '.', 0-9, A-Z, '^', '_', '`', a-z, '|', '~'
//
// tokenChars[32] === 0 // ' '
// tokenChars[33] === 1 // '!'
// tokenChars[34] === 0 // '"'
// ...
//
// prettier-ignore
const tokenChars = [
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, // 0 - 15
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, // 16 - 31
0, 1, 0, 1, 1, 1, 1, 1, 0, 0, 1, 1, 0, 1, 1, 0, // 32 - 47
1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, // 48 - 63
0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, // 64 - 79
1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 1, 1, // 80 - 95
1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, // 96 - 111
1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 1, 0, 1, 0 // 112 - 127
];
/**
* Adds an offer to the map of extension offers or a parameter to the map of
* parameters.
*
* @param {Object} dest The map of extension offers or parameters
* @param {String} name The extension or parameter name
* @param {(Object|Boolean|String)} elem The extension parameters or the
* parameter value
* @private
*/
function push(dest, name, elem) {
if (Object.prototype.hasOwnProperty.call(dest, name)) dest[name].push(elem);
else dest[name] = [elem];
}
/**
* Parses the `Sec-WebSocket-Extensions` header into an object.
*
* @param {String} header The field value of the header
* @return {Object} The parsed object
* @public
*/
function parse(header) {
const offers = {};
if (header === undefined || header === '') return offers;
var params = {};
var mustUnescape = false;
var isEscaping = false;
var inQuotes = false;
var extensionName;
var paramName;
var start = -1;
var end = -1;
for (var i = 0; i < header.length; i++) {
const code = header.charCodeAt(i);
if (extensionName === undefined) {
if (end === -1 && tokenChars[code] === 1) {
if (start === -1) start = i;
} else if (code === 0x20 /* ' ' */ || code === 0x09 /* '\t' */) {
if (end === -1 && start !== -1) end = i;
} else if (code === 0x3b /* ';' */ || code === 0x2c /* ',' */) {
if (start === -1) {
throw new SyntaxError(`Unexpected character at index ${i}`);
}
if (end === -1) end = i;
const name = header.slice(start, end);
if (code === 0x2c) {
push(offers, name, params);
params = {};
} else {
extensionName = name;
}
start = end = -1;
} else {
throw new SyntaxError(`Unexpected character at index ${i}`);
}
} else if (paramName === undefined) {
if (end === -1 && tokenChars[code] === 1) {
if (start === -1) start = i;
} else if (code === 0x20 || code === 0x09) {
if (end === -1 && start !== -1) end = i;
} else if (code === 0x3b || code === 0x2c) {
if (start === -1) {
throw new SyntaxError(`Unexpected character at index ${i}`);
}
if (end === -1) end = i;
push(params, header.slice(start, end), true);
if (code === 0x2c) {
push(offers, extensionName, params);
params = {};
extensionName = undefined;
}
start = end = -1;
} else if (code === 0x3d /* '=' */ && start !== -1 && end === -1) {
paramName = header.slice(start, i);
start = end = -1;
} else {
throw new SyntaxError(`Unexpected character at index ${i}`);
}
} else {
//
// The value of a quoted-string after unescaping must conform to the
// token ABNF, so only token characters are valid.
// Ref: https://tools.ietf.org/html/rfc6455#section-9.1
//
if (isEscaping) {
if (tokenChars[code] !== 1) {
throw new SyntaxError(`Unexpected character at index ${i}`);
}
if (start === -1) start = i;
else if (!mustUnescape) mustUnescape = true;
isEscaping = false;
} else if (inQuotes) {
if (tokenChars[code] === 1) {
if (start === -1) start = i;
} else if (code === 0x22 /* '"' */ && start !== -1) {
inQuotes = false;
end = i;
} else if (code === 0x5c /* '\' */) {
isEscaping = true;
} else {
throw new SyntaxError(`Unexpected character at index ${i}`);
}
} else if (code === 0x22 && header.charCodeAt(i - 1) === 0x3d) {
inQuotes = true;
} else if (end === -1 && tokenChars[code] === 1) {
if (start === -1) start = i;
} else if (start !== -1 && (code === 0x20 || code === 0x09)) {
if (end === -1) end = i;
} else if (code === 0x3b || code === 0x2c) {
if (start === -1) {
throw new SyntaxError(`Unexpected character at index ${i}`);
}
if (end === -1) end = i;
var value = header.slice(start, end);
if (mustUnescape) {
value = value.replace(/\\/g, '');
mustUnescape = false;
}
push(params, paramName, value);
if (code === 0x2c) {
push(offers, extensionName, params);
params = {};
extensionName = undefined;
}
paramName = undefined;
start = end = -1;
} else {
throw new SyntaxError(`Unexpected character at index ${i}`);
}
}
}
if (start === -1 || inQuotes) {
throw new SyntaxError('Unexpected end of input');
}
if (end === -1) end = i;
const token = header.slice(start, end);
if (extensionName === undefined) {
push(offers, token, {});
} else {
if (paramName === undefined) {
push(params, token, true);
} else if (mustUnescape) {
push(params, paramName, token.replace(/\\/g, ''));
} else {
push(params, paramName, token);
}
push(offers, extensionName, params);
}
return offers;
}
/**
* Builds the `Sec-WebSocket-Extensions` header field value.
*
* @param {Object} extensions The map of extensions and parameters to format
* @return {String} A string representing the given object
* @public
*/
function format(extensions) {
return Object.keys(extensions)
.map((extension) => {
var configurations = extensions[extension];
if (!Array.isArray(configurations)) configurations = [configurations];
return configurations
.map((params) => {
return [extension]
.concat(
Object.keys(params).map((k) => {
var values = params[k];
if (!Array.isArray(values)) values = [values];
return values
.map((v) => (v === true ? k : `${k}=${v}`))
.join('; ');
})
)
.join('; ');
})
.join(', ');
})
.join(', ');
}
module.exports = { format, parse };

View File

@@ -0,0 +1,502 @@
'use strict';
const Limiter = require('async-limiter');
const zlib = require('zlib');
const bufferUtil = require('./buffer-util');
const { kStatusCode, NOOP } = require('./constants');
const TRAILER = Buffer.from([0x00, 0x00, 0xff, 0xff]);
const EMPTY_BLOCK = Buffer.from([0x00]);
const kPerMessageDeflate = Symbol('permessage-deflate');
const kTotalLength = Symbol('total-length');
const kCallback = Symbol('callback');
const kBuffers = Symbol('buffers');
const kError = Symbol('error');
//
// We limit zlib concurrency, which prevents severe memory fragmentation
// as documented in https://github.com/nodejs/node/issues/8871#issuecomment-250915913
// and https://github.com/websockets/ws/issues/1202
//
// Intentionally global; it's the global thread pool that's an issue.
//
let zlibLimiter;
/**
* permessage-deflate implementation.
*/
class PerMessageDeflate {
/**
* Creates a PerMessageDeflate instance.
*
* @param {Object} options Configuration options
* @param {Boolean} options.serverNoContextTakeover Request/accept disabling
* of server context takeover
* @param {Boolean} options.clientNoContextTakeover Advertise/acknowledge
* disabling of client context takeover
* @param {(Boolean|Number)} options.serverMaxWindowBits Request/confirm the
* use of a custom server window size
* @param {(Boolean|Number)} options.clientMaxWindowBits Advertise support
* for, or request, a custom client window size
* @param {Object} options.zlibDeflateOptions Options to pass to zlib on deflate
* @param {Object} options.zlibInflateOptions Options to pass to zlib on inflate
* @param {Number} options.threshold Size (in bytes) below which messages
* should not be compressed
* @param {Number} options.concurrencyLimit The number of concurrent calls to
* zlib
* @param {Boolean} isServer Create the instance in either server or client
* mode
* @param {Number} maxPayload The maximum allowed message length
*/
constructor(options, isServer, maxPayload) {
this._maxPayload = maxPayload | 0;
this._options = options || {};
this._threshold =
this._options.threshold !== undefined ? this._options.threshold : 1024;
this._isServer = !!isServer;
this._deflate = null;
this._inflate = null;
this.params = null;
if (!zlibLimiter) {
const concurrency =
this._options.concurrencyLimit !== undefined
? this._options.concurrencyLimit
: 10;
zlibLimiter = new Limiter({ concurrency });
}
}
/**
* @type {String}
*/
static get extensionName() {
return 'permessage-deflate';
}
/**
* Create an extension negotiation offer.
*
* @return {Object} Extension parameters
* @public
*/
offer() {
const params = {};
if (this._options.serverNoContextTakeover) {
params.server_no_context_takeover = true;
}
if (this._options.clientNoContextTakeover) {
params.client_no_context_takeover = true;
}
if (this._options.serverMaxWindowBits) {
params.server_max_window_bits = this._options.serverMaxWindowBits;
}
if (this._options.clientMaxWindowBits) {
params.client_max_window_bits = this._options.clientMaxWindowBits;
} else if (this._options.clientMaxWindowBits == null) {
params.client_max_window_bits = true;
}
return params;
}
/**
* Accept an extension negotiation offer/response.
*
* @param {Array} configurations The extension negotiation offers/reponse
* @return {Object} Accepted configuration
* @public
*/
accept(configurations) {
configurations = this.normalizeParams(configurations);
this.params = this._isServer
? this.acceptAsServer(configurations)
: this.acceptAsClient(configurations);
return this.params;
}
/**
* Releases all resources used by the extension.
*
* @public
*/
cleanup() {
if (this._inflate) {
this._inflate.close();
this._inflate = null;
}
if (this._deflate) {
this._deflate.close();
this._deflate = null;
}
}
/**
* Accept an extension negotiation offer.
*
* @param {Array} offers The extension negotiation offers
* @return {Object} Accepted configuration
* @private
*/
acceptAsServer(offers) {
const opts = this._options;
const accepted = offers.find((params) => {
if (
(opts.serverNoContextTakeover === false &&
params.server_no_context_takeover) ||
(params.server_max_window_bits &&
(opts.serverMaxWindowBits === false ||
(typeof opts.serverMaxWindowBits === 'number' &&
opts.serverMaxWindowBits > params.server_max_window_bits))) ||
(typeof opts.clientMaxWindowBits === 'number' &&
!params.client_max_window_bits)
) {
return false;
}
return true;
});
if (!accepted) {
throw new Error('None of the extension offers can be accepted');
}
if (opts.serverNoContextTakeover) {
accepted.server_no_context_takeover = true;
}
if (opts.clientNoContextTakeover) {
accepted.client_no_context_takeover = true;
}
if (typeof opts.serverMaxWindowBits === 'number') {
accepted.server_max_window_bits = opts.serverMaxWindowBits;
}
if (typeof opts.clientMaxWindowBits === 'number') {
accepted.client_max_window_bits = opts.clientMaxWindowBits;
} else if (
accepted.client_max_window_bits === true ||
opts.clientMaxWindowBits === false
) {
delete accepted.client_max_window_bits;
}
return accepted;
}
/**
* Accept the extension negotiation response.
*
* @param {Array} response The extension negotiation response
* @return {Object} Accepted configuration
* @private
*/
acceptAsClient(response) {
const params = response[0];
if (
this._options.clientNoContextTakeover === false &&
params.client_no_context_takeover
) {
throw new Error('Unexpected parameter "client_no_context_takeover"');
}
if (!params.client_max_window_bits) {
if (typeof this._options.clientMaxWindowBits === 'number') {
params.client_max_window_bits = this._options.clientMaxWindowBits;
}
} else if (
this._options.clientMaxWindowBits === false ||
(typeof this._options.clientMaxWindowBits === 'number' &&
params.client_max_window_bits > this._options.clientMaxWindowBits)
) {
throw new Error(
'Unexpected or invalid parameter "client_max_window_bits"'
);
}
return params;
}
/**
* Normalize parameters.
*
* @param {Array} configurations The extension negotiation offers/reponse
* @return {Array} The offers/response with normalized parameters
* @private
*/
normalizeParams(configurations) {
configurations.forEach((params) => {
Object.keys(params).forEach((key) => {
var value = params[key];
if (value.length > 1) {
throw new Error(`Parameter "${key}" must have only a single value`);
}
value = value[0];
if (key === 'client_max_window_bits') {
if (value !== true) {
const num = +value;
if (!Number.isInteger(num) || num < 8 || num > 15) {
throw new TypeError(
`Invalid value for parameter "${key}": ${value}`
);
}
value = num;
} else if (!this._isServer) {
throw new TypeError(
`Invalid value for parameter "${key}": ${value}`
);
}
} else if (key === 'server_max_window_bits') {
const num = +value;
if (!Number.isInteger(num) || num < 8 || num > 15) {
throw new TypeError(
`Invalid value for parameter "${key}": ${value}`
);
}
value = num;
} else if (
key === 'client_no_context_takeover' ||
key === 'server_no_context_takeover'
) {
if (value !== true) {
throw new TypeError(
`Invalid value for parameter "${key}": ${value}`
);
}
} else {
throw new Error(`Unknown parameter "${key}"`);
}
params[key] = value;
});
});
return configurations;
}
/**
* Decompress data. Concurrency limited by async-limiter.
*
* @param {Buffer} data Compressed data
* @param {Boolean} fin Specifies whether or not this is the last fragment
* @param {Function} callback Callback
* @public
*/
decompress(data, fin, callback) {
zlibLimiter.push((done) => {
this._decompress(data, fin, (err, result) => {
done();
callback(err, result);
});
});
}
/**
* Compress data. Concurrency limited by async-limiter.
*
* @param {Buffer} data Data to compress
* @param {Boolean} fin Specifies whether or not this is the last fragment
* @param {Function} callback Callback
* @public
*/
compress(data, fin, callback) {
zlibLimiter.push((done) => {
this._compress(data, fin, (err, result) => {
done();
callback(err, result);
});
});
}
/**
* Decompress data.
*
* @param {Buffer} data Compressed data
* @param {Boolean} fin Specifies whether or not this is the last fragment
* @param {Function} callback Callback
* @private
*/
_decompress(data, fin, callback) {
const endpoint = this._isServer ? 'client' : 'server';
if (!this._inflate) {
const key = `${endpoint}_max_window_bits`;
const windowBits =
typeof this.params[key] !== 'number'
? zlib.Z_DEFAULT_WINDOWBITS
: this.params[key];
this._inflate = zlib.createInflateRaw(
Object.assign({}, this._options.zlibInflateOptions, { windowBits })
);
this._inflate[kPerMessageDeflate] = this;
this._inflate[kTotalLength] = 0;
this._inflate[kBuffers] = [];
this._inflate.on('error', inflateOnError);
this._inflate.on('data', inflateOnData);
}
this._inflate[kCallback] = callback;
this._inflate.write(data);
if (fin) this._inflate.write(TRAILER);
this._inflate.flush(() => {
const err = this._inflate[kError];
if (err) {
this._inflate.close();
this._inflate = null;
callback(err);
return;
}
const data = bufferUtil.concat(
this._inflate[kBuffers],
this._inflate[kTotalLength]
);
if (fin && this.params[`${endpoint}_no_context_takeover`]) {
this._inflate.close();
this._inflate = null;
} else {
this._inflate[kTotalLength] = 0;
this._inflate[kBuffers] = [];
}
callback(null, data);
});
}
/**
* Compress data.
*
* @param {Buffer} data Data to compress
* @param {Boolean} fin Specifies whether or not this is the last fragment
* @param {Function} callback Callback
* @private
*/
_compress(data, fin, callback) {
if (!data || data.length === 0) {
process.nextTick(callback, null, EMPTY_BLOCK);
return;
}
const endpoint = this._isServer ? 'server' : 'client';
if (!this._deflate) {
const key = `${endpoint}_max_window_bits`;
const windowBits =
typeof this.params[key] !== 'number'
? zlib.Z_DEFAULT_WINDOWBITS
: this.params[key];
this._deflate = zlib.createDeflateRaw(
Object.assign({}, this._options.zlibDeflateOptions, { windowBits })
);
this._deflate[kTotalLength] = 0;
this._deflate[kBuffers] = [];
//
// An `'error'` event is emitted, only on Node.js < 10.0.0, if the
// `zlib.DeflateRaw` instance is closed while data is being processed.
// This can happen if `PerMessageDeflate#cleanup()` is called at the wrong
// time due to an abnormal WebSocket closure.
//
this._deflate.on('error', NOOP);
this._deflate.on('data', deflateOnData);
}
this._deflate.write(data);
this._deflate.flush(zlib.Z_SYNC_FLUSH, () => {
if (!this._deflate) {
//
// This `if` statement is only needed for Node.js < 10.0.0 because as of
// commit https://github.com/nodejs/node/commit/5e3f5164, the flush
// callback is no longer called if the deflate stream is closed while
// data is being processed.
//
return;
}
var data = bufferUtil.concat(
this._deflate[kBuffers],
this._deflate[kTotalLength]
);
if (fin) data = data.slice(0, data.length - 4);
if (fin && this.params[`${endpoint}_no_context_takeover`]) {
this._deflate.close();
this._deflate = null;
} else {
this._deflate[kTotalLength] = 0;
this._deflate[kBuffers] = [];
}
callback(null, data);
});
}
}
module.exports = PerMessageDeflate;
/**
* The listener of the `zlib.DeflateRaw` stream `'data'` event.
*
* @param {Buffer} chunk A chunk of data
* @private
*/
function deflateOnData(chunk) {
this[kBuffers].push(chunk);
this[kTotalLength] += chunk.length;
}
/**
* The listener of the `zlib.InflateRaw` stream `'data'` event.
*
* @param {Buffer} chunk A chunk of data
* @private
*/
function inflateOnData(chunk) {
this[kTotalLength] += chunk.length;
if (
this[kPerMessageDeflate]._maxPayload < 1 ||
this[kTotalLength] <= this[kPerMessageDeflate]._maxPayload
) {
this[kBuffers].push(chunk);
return;
}
this[kError] = new RangeError('Max payload size exceeded');
this[kError][kStatusCode] = 1009;
this.removeListener('data', inflateOnData);
this.reset();
}
/**
* The listener of the `zlib.InflateRaw` stream `'error'` event.
*
* @param {Error} err The emitted error
* @private
*/
function inflateOnError(err) {
//
// There is no need to call `Zlib#close()` as the handle is automatically
// closed when an error is emitted.
//
this[kPerMessageDeflate]._inflate = null;
err[kStatusCode] = 1007;
this[kCallback](err);
}

View File

@@ -0,0 +1,492 @@
'use strict';
const { Writable } = require('stream');
const PerMessageDeflate = require('./permessage-deflate');
const {
BINARY_TYPES,
EMPTY_BUFFER,
kStatusCode,
kWebSocket
} = require('./constants');
const { concat, toArrayBuffer, unmask } = require('./buffer-util');
const { isValidStatusCode, isValidUTF8 } = require('./validation');
const GET_INFO = 0;
const GET_PAYLOAD_LENGTH_16 = 1;
const GET_PAYLOAD_LENGTH_64 = 2;
const GET_MASK = 3;
const GET_DATA = 4;
const INFLATING = 5;
/**
* HyBi Receiver implementation.
*
* @extends stream.Writable
*/
class Receiver extends Writable {
/**
* Creates a Receiver instance.
*
* @param {String} binaryType The type for binary data
* @param {Object} extensions An object containing the negotiated extensions
* @param {Number} maxPayload The maximum allowed message length
*/
constructor(binaryType, extensions, maxPayload) {
super();
this._binaryType = binaryType || BINARY_TYPES[0];
this[kWebSocket] = undefined;
this._extensions = extensions || {};
this._maxPayload = maxPayload | 0;
this._bufferedBytes = 0;
this._buffers = [];
this._compressed = false;
this._payloadLength = 0;
this._mask = undefined;
this._fragmented = 0;
this._masked = false;
this._fin = false;
this._opcode = 0;
this._totalPayloadLength = 0;
this._messageLength = 0;
this._fragments = [];
this._state = GET_INFO;
this._loop = false;
}
/**
* Implements `Writable.prototype._write()`.
*
* @param {Buffer} chunk The chunk of data to write
* @param {String} encoding The character encoding of `chunk`
* @param {Function} cb Callback
*/
_write(chunk, encoding, cb) {
if (this._opcode === 0x08 && this._state == GET_INFO) return cb();
this._bufferedBytes += chunk.length;
this._buffers.push(chunk);
this.startLoop(cb);
}
/**
* Consumes `n` bytes from the buffered data.
*
* @param {Number} n The number of bytes to consume
* @return {Buffer} The consumed bytes
* @private
*/
consume(n) {
this._bufferedBytes -= n;
if (n === this._buffers[0].length) return this._buffers.shift();
if (n < this._buffers[0].length) {
const buf = this._buffers[0];
this._buffers[0] = buf.slice(n);
return buf.slice(0, n);
}
const dst = Buffer.allocUnsafe(n);
do {
const buf = this._buffers[0];
if (n >= buf.length) {
this._buffers.shift().copy(dst, dst.length - n);
} else {
buf.copy(dst, dst.length - n, 0, n);
this._buffers[0] = buf.slice(n);
}
n -= buf.length;
} while (n > 0);
return dst;
}
/**
* Starts the parsing loop.
*
* @param {Function} cb Callback
* @private
*/
startLoop(cb) {
var err;
this._loop = true;
do {
switch (this._state) {
case GET_INFO:
err = this.getInfo();
break;
case GET_PAYLOAD_LENGTH_16:
err = this.getPayloadLength16();
break;
case GET_PAYLOAD_LENGTH_64:
err = this.getPayloadLength64();
break;
case GET_MASK:
this.getMask();
break;
case GET_DATA:
err = this.getData(cb);
break;
default:
// `INFLATING`
this._loop = false;
return;
}
} while (this._loop);
cb(err);
}
/**
* Reads the first two bytes of a frame.
*
* @return {(RangeError|undefined)} A possible error
* @private
*/
getInfo() {
if (this._bufferedBytes < 2) {
this._loop = false;
return;
}
const buf = this.consume(2);
if ((buf[0] & 0x30) !== 0x00) {
this._loop = false;
return error(RangeError, 'RSV2 and RSV3 must be clear', true, 1002);
}
const compressed = (buf[0] & 0x40) === 0x40;
if (compressed && !this._extensions[PerMessageDeflate.extensionName]) {
this._loop = false;
return error(RangeError, 'RSV1 must be clear', true, 1002);
}
this._fin = (buf[0] & 0x80) === 0x80;
this._opcode = buf[0] & 0x0f;
this._payloadLength = buf[1] & 0x7f;
if (this._opcode === 0x00) {
if (compressed) {
this._loop = false;
return error(RangeError, 'RSV1 must be clear', true, 1002);
}
if (!this._fragmented) {
this._loop = false;
return error(RangeError, 'invalid opcode 0', true, 1002);
}
this._opcode = this._fragmented;
} else if (this._opcode === 0x01 || this._opcode === 0x02) {
if (this._fragmented) {
this._loop = false;
return error(RangeError, `invalid opcode ${this._opcode}`, true, 1002);
}
this._compressed = compressed;
} else if (this._opcode > 0x07 && this._opcode < 0x0b) {
if (!this._fin) {
this._loop = false;
return error(RangeError, 'FIN must be set', true, 1002);
}
if (compressed) {
this._loop = false;
return error(RangeError, 'RSV1 must be clear', true, 1002);
}
if (this._payloadLength > 0x7d) {
this._loop = false;
return error(
RangeError,
`invalid payload length ${this._payloadLength}`,
true,
1002
);
}
} else {
this._loop = false;
return error(RangeError, `invalid opcode ${this._opcode}`, true, 1002);
}
if (!this._fin && !this._fragmented) this._fragmented = this._opcode;
this._masked = (buf[1] & 0x80) === 0x80;
if (this._payloadLength === 126) this._state = GET_PAYLOAD_LENGTH_16;
else if (this._payloadLength === 127) this._state = GET_PAYLOAD_LENGTH_64;
else return this.haveLength();
}
/**
* Gets extended payload length (7+16).
*
* @return {(RangeError|undefined)} A possible error
* @private
*/
getPayloadLength16() {
if (this._bufferedBytes < 2) {
this._loop = false;
return;
}
this._payloadLength = this.consume(2).readUInt16BE(0);
return this.haveLength();
}
/**
* Gets extended payload length (7+64).
*
* @return {(RangeError|undefined)} A possible error
* @private
*/
getPayloadLength64() {
if (this._bufferedBytes < 8) {
this._loop = false;
return;
}
const buf = this.consume(8);
const num = buf.readUInt32BE(0);
//
// The maximum safe integer in JavaScript is 2^53 - 1. An error is returned
// if payload length is greater than this number.
//
if (num > Math.pow(2, 53 - 32) - 1) {
this._loop = false;
return error(
RangeError,
'Unsupported WebSocket frame: payload length > 2^53 - 1',
false,
1009
);
}
this._payloadLength = num * Math.pow(2, 32) + buf.readUInt32BE(4);
return this.haveLength();
}
/**
* Payload length has been read.
*
* @return {(RangeError|undefined)} A possible error
* @private
*/
haveLength() {
if (this._payloadLength && this._opcode < 0x08) {
this._totalPayloadLength += this._payloadLength;
if (this._totalPayloadLength > this._maxPayload && this._maxPayload > 0) {
this._loop = false;
return error(RangeError, 'Max payload size exceeded', false, 1009);
}
}
if (this._masked) this._state = GET_MASK;
else this._state = GET_DATA;
}
/**
* Reads mask bytes.
*
* @private
*/
getMask() {
if (this._bufferedBytes < 4) {
this._loop = false;
return;
}
this._mask = this.consume(4);
this._state = GET_DATA;
}
/**
* Reads data bytes.
*
* @param {Function} cb Callback
* @return {(Error|RangeError|undefined)} A possible error
* @private
*/
getData(cb) {
var data = EMPTY_BUFFER;
if (this._payloadLength) {
if (this._bufferedBytes < this._payloadLength) {
this._loop = false;
return;
}
data = this.consume(this._payloadLength);
if (this._masked) unmask(data, this._mask);
}
if (this._opcode > 0x07) return this.controlMessage(data);
if (this._compressed) {
this._state = INFLATING;
this.decompress(data, cb);
return;
}
if (data.length) {
//
// This message is not compressed so its lenght is the sum of the payload
// length of all fragments.
//
this._messageLength = this._totalPayloadLength;
this._fragments.push(data);
}
return this.dataMessage();
}
/**
* Decompresses data.
*
* @param {Buffer} data Compressed data
* @param {Function} cb Callback
* @private
*/
decompress(data, cb) {
const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];
perMessageDeflate.decompress(data, this._fin, (err, buf) => {
if (err) return cb(err);
if (buf.length) {
this._messageLength += buf.length;
if (this._messageLength > this._maxPayload && this._maxPayload > 0) {
return cb(
error(RangeError, 'Max payload size exceeded', false, 1009)
);
}
this._fragments.push(buf);
}
const er = this.dataMessage();
if (er) return cb(er);
this.startLoop(cb);
});
}
/**
* Handles a data message.
*
* @return {(Error|undefined)} A possible error
* @private
*/
dataMessage() {
if (this._fin) {
const messageLength = this._messageLength;
const fragments = this._fragments;
this._totalPayloadLength = 0;
this._messageLength = 0;
this._fragmented = 0;
this._fragments = [];
if (this._opcode === 2) {
var data;
if (this._binaryType === 'nodebuffer') {
data = concat(fragments, messageLength);
} else if (this._binaryType === 'arraybuffer') {
data = toArrayBuffer(concat(fragments, messageLength));
} else {
data = fragments;
}
this.emit('message', data);
} else {
const buf = concat(fragments, messageLength);
if (!isValidUTF8(buf)) {
this._loop = false;
return error(Error, 'invalid UTF-8 sequence', true, 1007);
}
this.emit('message', buf.toString());
}
}
this._state = GET_INFO;
}
/**
* Handles a control message.
*
* @param {Buffer} data Data to handle
* @return {(Error|RangeError|undefined)} A possible error
* @private
*/
controlMessage(data) {
if (this._opcode === 0x08) {
this._loop = false;
if (data.length === 0) {
this.emit('conclude', 1005, '');
this.end();
} else if (data.length === 1) {
return error(RangeError, 'invalid payload length 1', true, 1002);
} else {
const code = data.readUInt16BE(0);
if (!isValidStatusCode(code)) {
return error(RangeError, `invalid status code ${code}`, true, 1002);
}
const buf = data.slice(2);
if (!isValidUTF8(buf)) {
return error(Error, 'invalid UTF-8 sequence', true, 1007);
}
this.emit('conclude', code, buf.toString());
this.end();
}
} else if (this._opcode === 0x09) {
this.emit('ping', data);
} else {
this.emit('pong', data);
}
this._state = GET_INFO;
}
}
module.exports = Receiver;
/**
* Builds an error object.
*
* @param {(Error|RangeError)} ErrorCtor The error constructor
* @param {String} message The error message
* @param {Boolean} prefix Specifies whether or not to add a default prefix to
* `message`
* @param {Number} statusCode The status code
* @return {(Error|RangeError)} The error
* @private
*/
function error(ErrorCtor, message, prefix, statusCode) {
const err = new ErrorCtor(
prefix ? `Invalid WebSocket frame: ${message}` : message
);
Error.captureStackTrace(err, error);
err[kStatusCode] = statusCode;
return err;
}

358
node_modules/y-websocket/node_modules/ws/lib/sender.js generated vendored Normal file
View File

@@ -0,0 +1,358 @@
'use strict';
const { randomBytes } = require('crypto');
const PerMessageDeflate = require('./permessage-deflate');
const { EMPTY_BUFFER } = require('./constants');
const { isValidStatusCode } = require('./validation');
const { mask: applyMask, toBuffer } = require('./buffer-util');
/**
* HyBi Sender implementation.
*/
class Sender {
/**
* Creates a Sender instance.
*
* @param {net.Socket} socket The connection socket
* @param {Object} extensions An object containing the negotiated extensions
*/
constructor(socket, extensions) {
this._extensions = extensions || {};
this._socket = socket;
this._firstFragment = true;
this._compress = false;
this._bufferedBytes = 0;
this._deflating = false;
this._queue = [];
}
/**
* Frames a piece of data according to the HyBi WebSocket protocol.
*
* @param {Buffer} data The data to frame
* @param {Object} options Options object
* @param {Number} options.opcode The opcode
* @param {Boolean} options.readOnly Specifies whether `data` can be modified
* @param {Boolean} options.fin Specifies whether or not to set the FIN bit
* @param {Boolean} options.mask Specifies whether or not to mask `data`
* @param {Boolean} options.rsv1 Specifies whether or not to set the RSV1 bit
* @return {Buffer[]} The framed data as a list of `Buffer` instances
* @public
*/
static frame(data, options) {
const merge = options.mask && options.readOnly;
var offset = options.mask ? 6 : 2;
var payloadLength = data.length;
if (data.length >= 65536) {
offset += 8;
payloadLength = 127;
} else if (data.length > 125) {
offset += 2;
payloadLength = 126;
}
const target = Buffer.allocUnsafe(merge ? data.length + offset : offset);
target[0] = options.fin ? options.opcode | 0x80 : options.opcode;
if (options.rsv1) target[0] |= 0x40;
target[1] = payloadLength;
if (payloadLength === 126) {
target.writeUInt16BE(data.length, 2);
} else if (payloadLength === 127) {
target.writeUInt32BE(0, 2);
target.writeUInt32BE(data.length, 6);
}
if (!options.mask) return [target, data];
const mask = randomBytes(4);
target[1] |= 0x80;
target[offset - 4] = mask[0];
target[offset - 3] = mask[1];
target[offset - 2] = mask[2];
target[offset - 1] = mask[3];
if (merge) {
applyMask(data, mask, target, offset, data.length);
return [target];
}
applyMask(data, mask, data, 0, data.length);
return [target, data];
}
/**
* Sends a close message to the other peer.
*
* @param {(Number|undefined)} code The status code component of the body
* @param {String} data The message component of the body
* @param {Boolean} mask Specifies whether or not to mask the message
* @param {Function} cb Callback
* @public
*/
close(code, data, mask, cb) {
var buf;
if (code === undefined) {
buf = EMPTY_BUFFER;
} else if (typeof code !== 'number' || !isValidStatusCode(code)) {
throw new TypeError('First argument must be a valid error code number');
} else if (data === undefined || data === '') {
buf = Buffer.allocUnsafe(2);
buf.writeUInt16BE(code, 0);
} else {
buf = Buffer.allocUnsafe(2 + Buffer.byteLength(data));
buf.writeUInt16BE(code, 0);
buf.write(data, 2);
}
if (this._deflating) {
this.enqueue([this.doClose, buf, mask, cb]);
} else {
this.doClose(buf, mask, cb);
}
}
/**
* Frames and sends a close message.
*
* @param {Buffer} data The message to send
* @param {Boolean} mask Specifies whether or not to mask `data`
* @param {Function} cb Callback
* @private
*/
doClose(data, mask, cb) {
this.sendFrame(
Sender.frame(data, {
fin: true,
rsv1: false,
opcode: 0x08,
mask,
readOnly: false
}),
cb
);
}
/**
* Sends a ping message to the other peer.
*
* @param {*} data The message to send
* @param {Boolean} mask Specifies whether or not to mask `data`
* @param {Function} cb Callback
* @public
*/
ping(data, mask, cb) {
const buf = toBuffer(data);
if (this._deflating) {
this.enqueue([this.doPing, buf, mask, toBuffer.readOnly, cb]);
} else {
this.doPing(buf, mask, toBuffer.readOnly, cb);
}
}
/**
* Frames and sends a ping message.
*
* @param {*} data The message to send
* @param {Boolean} mask Specifies whether or not to mask `data`
* @param {Boolean} readOnly Specifies whether `data` can be modified
* @param {Function} cb Callback
* @private
*/
doPing(data, mask, readOnly, cb) {
this.sendFrame(
Sender.frame(data, {
fin: true,
rsv1: false,
opcode: 0x09,
mask,
readOnly
}),
cb
);
}
/**
* Sends a pong message to the other peer.
*
* @param {*} data The message to send
* @param {Boolean} mask Specifies whether or not to mask `data`
* @param {Function} cb Callback
* @public
*/
pong(data, mask, cb) {
const buf = toBuffer(data);
if (this._deflating) {
this.enqueue([this.doPong, buf, mask, toBuffer.readOnly, cb]);
} else {
this.doPong(buf, mask, toBuffer.readOnly, cb);
}
}
/**
* Frames and sends a pong message.
*
* @param {*} data The message to send
* @param {Boolean} mask Specifies whether or not to mask `data`
* @param {Boolean} readOnly Specifies whether `data` can be modified
* @param {Function} cb Callback
* @private
*/
doPong(data, mask, readOnly, cb) {
this.sendFrame(
Sender.frame(data, {
fin: true,
rsv1: false,
opcode: 0x0a,
mask,
readOnly
}),
cb
);
}
/**
* Sends a data message to the other peer.
*
* @param {*} data The message to send
* @param {Object} options Options object
* @param {Boolean} options.compress Specifies whether or not to compress `data`
* @param {Boolean} options.binary Specifies whether `data` is binary or text
* @param {Boolean} options.fin Specifies whether the fragment is the last one
* @param {Boolean} options.mask Specifies whether or not to mask `data`
* @param {Function} cb Callback
* @public
*/
send(data, options, cb) {
const buf = toBuffer(data);
const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];
var opcode = options.binary ? 2 : 1;
var rsv1 = options.compress;
if (this._firstFragment) {
this._firstFragment = false;
if (rsv1 && perMessageDeflate) {
rsv1 = buf.length >= perMessageDeflate._threshold;
}
this._compress = rsv1;
} else {
rsv1 = false;
opcode = 0;
}
if (options.fin) this._firstFragment = true;
if (perMessageDeflate) {
const opts = {
fin: options.fin,
rsv1,
opcode,
mask: options.mask,
readOnly: toBuffer.readOnly
};
if (this._deflating) {
this.enqueue([this.dispatch, buf, this._compress, opts, cb]);
} else {
this.dispatch(buf, this._compress, opts, cb);
}
} else {
this.sendFrame(
Sender.frame(buf, {
fin: options.fin,
rsv1: false,
opcode,
mask: options.mask,
readOnly: toBuffer.readOnly
}),
cb
);
}
}
/**
* Dispatches a data message.
*
* @param {Buffer} data The message to send
* @param {Boolean} compress Specifies whether or not to compress `data`
* @param {Object} options Options object
* @param {Number} options.opcode The opcode
* @param {Boolean} options.readOnly Specifies whether `data` can be modified
* @param {Boolean} options.fin Specifies whether or not to set the FIN bit
* @param {Boolean} options.mask Specifies whether or not to mask `data`
* @param {Boolean} options.rsv1 Specifies whether or not to set the RSV1 bit
* @param {Function} cb Callback
* @private
*/
dispatch(data, compress, options, cb) {
if (!compress) {
this.sendFrame(Sender.frame(data, options), cb);
return;
}
const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];
this._deflating = true;
perMessageDeflate.compress(data, options.fin, (_, buf) => {
this._deflating = false;
options.readOnly = false;
this.sendFrame(Sender.frame(buf, options), cb);
this.dequeue();
});
}
/**
* Executes queued send operations.
*
* @private
*/
dequeue() {
while (!this._deflating && this._queue.length) {
const params = this._queue.shift();
this._bufferedBytes -= params[1].length;
params[0].apply(this, params.slice(1));
}
}
/**
* Enqueues a send operation.
*
* @param {Array} params Send operation parameters.
* @private
*/
enqueue(params) {
this._bufferedBytes += params[1].length;
this._queue.push(params);
}
/**
* Sends a frame.
*
* @param {Buffer[]} list The frame to send
* @param {Function} cb Callback
* @private
*/
sendFrame(list, cb) {
if (list.length === 2) {
this._socket.cork();
this._socket.write(list[0]);
this._socket.write(list[1], cb);
this._socket.uncork();
} else {
this._socket.write(list[0], cb);
}
}
}
module.exports = Sender;

View File

@@ -0,0 +1,30 @@
'use strict';
try {
const isValidUTF8 = require('utf-8-validate');
exports.isValidUTF8 =
typeof isValidUTF8 === 'object'
? isValidUTF8.Validation.isValidUTF8 // utf-8-validate@<3.0.0
: isValidUTF8;
} catch (e) /* istanbul ignore next */ {
exports.isValidUTF8 = () => true;
}
/**
* Checks if a status code is allowed in a close frame.
*
* @param {Number} code The status code
* @return {Boolean} `true` if the status code is valid, else `false`
* @public
*/
exports.isValidStatusCode = (code) => {
return (
(code >= 1000 &&
code <= 1013 &&
code !== 1004 &&
code !== 1005 &&
code !== 1006) ||
(code >= 3000 && code <= 4999)
);
};

View File

@@ -0,0 +1,415 @@
'use strict';
const EventEmitter = require('events');
const crypto = require('crypto');
const http = require('http');
const PerMessageDeflate = require('./permessage-deflate');
const extension = require('./extension');
const WebSocket = require('./websocket');
const { GUID } = require('./constants');
const keyRegex = /^[+/0-9A-Za-z]{22}==$/;
/**
* Class representing a WebSocket server.
*
* @extends EventEmitter
*/
class WebSocketServer extends EventEmitter {
/**
* Create a `WebSocketServer` instance.
*
* @param {Object} options Configuration options
* @param {Number} options.backlog The maximum length of the queue of pending
* connections
* @param {Boolean} options.clientTracking Specifies whether or not to track
* clients
* @param {Function} options.handleProtocols An hook to handle protocols
* @param {String} options.host The hostname where to bind the server
* @param {Number} options.maxPayload The maximum allowed message size
* @param {Boolean} options.noServer Enable no server mode
* @param {String} options.path Accept only connections matching this path
* @param {(Boolean|Object)} options.perMessageDeflate Enable/disable
* permessage-deflate
* @param {Number} options.port The port where to bind the server
* @param {http.Server} options.server A pre-created HTTP/S server to use
* @param {Function} options.verifyClient An hook to reject connections
* @param {Function} callback A listener for the `listening` event
*/
constructor(options, callback) {
super();
options = Object.assign(
{
maxPayload: 100 * 1024 * 1024,
perMessageDeflate: false,
handleProtocols: null,
clientTracking: true,
verifyClient: null,
noServer: false,
backlog: null, // use default (511 as implemented in net.js)
server: null,
host: null,
path: null,
port: null
},
options
);
if (options.port == null && !options.server && !options.noServer) {
throw new TypeError(
'One of the "port", "server", or "noServer" options must be specified'
);
}
if (options.port != null) {
this._server = http.createServer((req, res) => {
const body = http.STATUS_CODES[426];
res.writeHead(426, {
'Content-Length': body.length,
'Content-Type': 'text/plain'
});
res.end(body);
});
this._server.listen(
options.port,
options.host,
options.backlog,
callback
);
} else if (options.server) {
this._server = options.server;
}
if (this._server) {
this._removeListeners = addListeners(this._server, {
listening: this.emit.bind(this, 'listening'),
error: this.emit.bind(this, 'error'),
upgrade: (req, socket, head) => {
this.handleUpgrade(req, socket, head, (ws) => {
this.emit('connection', ws, req);
});
}
});
}
if (options.perMessageDeflate === true) options.perMessageDeflate = {};
if (options.clientTracking) this.clients = new Set();
this.options = options;
}
/**
* Returns the bound address, the address family name, and port of the server
* as reported by the operating system if listening on an IP socket.
* If the server is listening on a pipe or UNIX domain socket, the name is
* returned as a string.
*
* @return {(Object|String|null)} The address of the server
* @public
*/
address() {
if (this.options.noServer) {
throw new Error('The server is operating in "noServer" mode');
}
if (!this._server) return null;
return this._server.address();
}
/**
* Close the server.
*
* @param {Function} cb Callback
* @public
*/
close(cb) {
if (cb) this.once('close', cb);
//
// Terminate all associated clients.
//
if (this.clients) {
for (const client of this.clients) client.terminate();
}
const server = this._server;
if (server) {
this._removeListeners();
this._removeListeners = this._server = null;
//
// Close the http server if it was internally created.
//
if (this.options.port != null) {
server.close(() => this.emit('close'));
return;
}
}
process.nextTick(emitClose, this);
}
/**
* See if a given request should be handled by this server instance.
*
* @param {http.IncomingMessage} req Request object to inspect
* @return {Boolean} `true` if the request is valid, else `false`
* @public
*/
shouldHandle(req) {
if (this.options.path) {
const index = req.url.indexOf('?');
const pathname = index !== -1 ? req.url.slice(0, index) : req.url;
if (pathname !== this.options.path) return false;
}
return true;
}
/**
* Handle a HTTP Upgrade request.
*
* @param {http.IncomingMessage} req The request object
* @param {net.Socket} socket The network socket between the server and client
* @param {Buffer} head The first packet of the upgraded stream
* @param {Function} cb Callback
* @public
*/
handleUpgrade(req, socket, head, cb) {
socket.on('error', socketOnError);
const key =
req.headers['sec-websocket-key'] !== undefined
? req.headers['sec-websocket-key'].trim()
: false;
const upgrade = req.headers.upgrade;
const version = +req.headers['sec-websocket-version'];
const extensions = {};
if (
req.method !== 'GET' ||
upgrade === undefined ||
upgrade.toLowerCase() !== 'websocket' ||
!key ||
!keyRegex.test(key) ||
(version !== 8 && version !== 13) ||
!this.shouldHandle(req)
) {
return abortHandshake(socket, 400);
}
if (this.options.perMessageDeflate) {
const perMessageDeflate = new PerMessageDeflate(
this.options.perMessageDeflate,
true,
this.options.maxPayload
);
try {
const offers = extension.parse(req.headers['sec-websocket-extensions']);
if (offers[PerMessageDeflate.extensionName]) {
perMessageDeflate.accept(offers[PerMessageDeflate.extensionName]);
extensions[PerMessageDeflate.extensionName] = perMessageDeflate;
}
} catch (err) {
return abortHandshake(socket, 400);
}
}
//
// Optionally call external client verification handler.
//
if (this.options.verifyClient) {
const info = {
origin:
req.headers[`${version === 8 ? 'sec-websocket-origin' : 'origin'}`],
secure: !!(req.connection.authorized || req.connection.encrypted),
req
};
if (this.options.verifyClient.length === 2) {
this.options.verifyClient(info, (verified, code, message, headers) => {
if (!verified) {
return abortHandshake(socket, code || 401, message, headers);
}
this.completeUpgrade(key, extensions, req, socket, head, cb);
});
return;
}
if (!this.options.verifyClient(info)) return abortHandshake(socket, 401);
}
this.completeUpgrade(key, extensions, req, socket, head, cb);
}
/**
* Upgrade the connection to WebSocket.
*
* @param {String} key The value of the `Sec-WebSocket-Key` header
* @param {Object} extensions The accepted extensions
* @param {http.IncomingMessage} req The request object
* @param {net.Socket} socket The network socket between the server and client
* @param {Buffer} head The first packet of the upgraded stream
* @param {Function} cb Callback
* @private
*/
completeUpgrade(key, extensions, req, socket, head, cb) {
//
// Destroy the socket if the client has already sent a FIN packet.
//
if (!socket.readable || !socket.writable) return socket.destroy();
const digest = crypto
.createHash('sha1')
.update(key + GUID)
.digest('base64');
const headers = [
'HTTP/1.1 101 Switching Protocols',
'Upgrade: websocket',
'Connection: Upgrade',
`Sec-WebSocket-Accept: ${digest}`
];
const ws = new WebSocket(null);
var protocol = req.headers['sec-websocket-protocol'];
if (protocol) {
protocol = protocol.split(',').map(trim);
//
// Optionally call external protocol selection handler.
//
if (this.options.handleProtocols) {
protocol = this.options.handleProtocols(protocol, req);
} else {
protocol = protocol[0];
}
if (protocol) {
headers.push(`Sec-WebSocket-Protocol: ${protocol}`);
ws.protocol = protocol;
}
}
if (extensions[PerMessageDeflate.extensionName]) {
const params = extensions[PerMessageDeflate.extensionName].params;
const value = extension.format({
[PerMessageDeflate.extensionName]: [params]
});
headers.push(`Sec-WebSocket-Extensions: ${value}`);
ws._extensions = extensions;
}
//
// Allow external modification/inspection of handshake headers.
//
this.emit('headers', headers, req);
socket.write(headers.concat('\r\n').join('\r\n'));
socket.removeListener('error', socketOnError);
ws.setSocket(socket, head, this.options.maxPayload);
if (this.clients) {
this.clients.add(ws);
ws.on('close', () => this.clients.delete(ws));
}
cb(ws);
}
}
module.exports = WebSocketServer;
/**
* Add event listeners on an `EventEmitter` using a map of <event, listener>
* pairs.
*
* @param {EventEmitter} server The event emitter
* @param {Object.<String, Function>} map The listeners to add
* @return {Function} A function that will remove the added listeners when called
* @private
*/
function addListeners(server, map) {
for (const event of Object.keys(map)) server.on(event, map[event]);
return function removeListeners() {
for (const event of Object.keys(map)) {
server.removeListener(event, map[event]);
}
};
}
/**
* Emit a `'close'` event on an `EventEmitter`.
*
* @param {EventEmitter} server The event emitter
* @private
*/
function emitClose(server) {
server.emit('close');
}
/**
* Handle premature socket errors.
*
* @private
*/
function socketOnError() {
this.destroy();
}
/**
* Close the connection when preconditions are not fulfilled.
*
* @param {net.Socket} socket The socket of the upgrade request
* @param {Number} code The HTTP response status code
* @param {String} [message] The HTTP response body
* @param {Object} [headers] Additional HTTP response headers
* @private
*/
function abortHandshake(socket, code, message, headers) {
if (socket.writable) {
message = message || http.STATUS_CODES[code];
headers = Object.assign(
{
Connection: 'close',
'Content-type': 'text/html',
'Content-Length': Buffer.byteLength(message)
},
headers
);
socket.write(
`HTTP/1.1 ${code} ${http.STATUS_CODES[code]}\r\n` +
Object.keys(headers)
.map((h) => `${h}: ${headers[h]}`)
.join('\r\n') +
'\r\n\r\n' +
message
);
}
socket.removeListener('error', socketOnError);
socket.destroy();
}
/**
* Remove whitespace characters from both ends of a string.
*
* @param {String} str The string
* @return {String} A new string representing `str` stripped of whitespace
* characters from both its beginning and end
* @private
*/
function trim(str) {
return str.trim();
}

View File

@@ -0,0 +1,895 @@
'use strict';
const EventEmitter = require('events');
const crypto = require('crypto');
const https = require('https');
const http = require('http');
const net = require('net');
const tls = require('tls');
const url = require('url');
const PerMessageDeflate = require('./permessage-deflate');
const EventTarget = require('./event-target');
const extension = require('./extension');
const Receiver = require('./receiver');
const Sender = require('./sender');
const {
BINARY_TYPES,
EMPTY_BUFFER,
GUID,
kStatusCode,
kWebSocket,
NOOP
} = require('./constants');
const readyStates = ['CONNECTING', 'OPEN', 'CLOSING', 'CLOSED'];
const protocolVersions = [8, 13];
const closeTimeout = 30 * 1000;
/**
* Class representing a WebSocket.
*
* @extends EventEmitter
*/
class WebSocket extends EventEmitter {
/**
* Create a new `WebSocket`.
*
* @param {(String|url.Url|url.URL)} address The URL to which to connect
* @param {(String|String[])} protocols The subprotocols
* @param {Object} options Connection options
*/
constructor(address, protocols, options) {
super();
this.readyState = WebSocket.CONNECTING;
this.protocol = '';
this._binaryType = BINARY_TYPES[0];
this._closeFrameReceived = false;
this._closeFrameSent = false;
this._closeMessage = '';
this._closeTimer = null;
this._closeCode = 1006;
this._extensions = {};
this._receiver = null;
this._sender = null;
this._socket = null;
if (address !== null) {
this._isServer = false;
this._redirects = 0;
if (Array.isArray(protocols)) {
protocols = protocols.join(', ');
} else if (typeof protocols === 'object' && protocols !== null) {
options = protocols;
protocols = undefined;
}
initAsClient(this, address, protocols, options);
} else {
this._isServer = true;
}
}
get CONNECTING() {
return WebSocket.CONNECTING;
}
get CLOSING() {
return WebSocket.CLOSING;
}
get CLOSED() {
return WebSocket.CLOSED;
}
get OPEN() {
return WebSocket.OPEN;
}
/**
* This deviates from the WHATWG interface since ws doesn't support the
* required default "blob" type (instead we define a custom "nodebuffer"
* type).
*
* @type {String}
*/
get binaryType() {
return this._binaryType;
}
set binaryType(type) {
if (!BINARY_TYPES.includes(type)) return;
this._binaryType = type;
//
// Allow to change `binaryType` on the fly.
//
if (this._receiver) this._receiver._binaryType = type;
}
/**
* @type {Number}
*/
get bufferedAmount() {
if (!this._socket) return 0;
//
// `socket.bufferSize` is `undefined` if the socket is closed.
//
return (this._socket.bufferSize || 0) + this._sender._bufferedBytes;
}
/**
* @type {String}
*/
get extensions() {
return Object.keys(this._extensions).join();
}
/**
* Set up the socket and the internal resources.
*
* @param {net.Socket} socket The network socket between the server and client
* @param {Buffer} head The first packet of the upgraded stream
* @param {Number} maxPayload The maximum allowed message size
* @private
*/
setSocket(socket, head, maxPayload) {
const receiver = new Receiver(
this._binaryType,
this._extensions,
maxPayload
);
this._sender = new Sender(socket, this._extensions);
this._receiver = receiver;
this._socket = socket;
receiver[kWebSocket] = this;
socket[kWebSocket] = this;
receiver.on('conclude', receiverOnConclude);
receiver.on('drain', receiverOnDrain);
receiver.on('error', receiverOnError);
receiver.on('message', receiverOnMessage);
receiver.on('ping', receiverOnPing);
receiver.on('pong', receiverOnPong);
socket.setTimeout(0);
socket.setNoDelay();
if (head.length > 0) socket.unshift(head);
socket.on('close', socketOnClose);
socket.on('data', socketOnData);
socket.on('end', socketOnEnd);
socket.on('error', socketOnError);
this.readyState = WebSocket.OPEN;
this.emit('open');
}
/**
* Emit the `'close'` event.
*
* @private
*/
emitClose() {
this.readyState = WebSocket.CLOSED;
if (!this._socket) {
this.emit('close', this._closeCode, this._closeMessage);
return;
}
if (this._extensions[PerMessageDeflate.extensionName]) {
this._extensions[PerMessageDeflate.extensionName].cleanup();
}
this._receiver.removeAllListeners();
this.emit('close', this._closeCode, this._closeMessage);
}
/**
* Start a closing handshake.
*
* +----------+ +-----------+ +----------+
* - - -|ws.close()|-->|close frame|-->|ws.close()|- - -
* | +----------+ +-----------+ +----------+ |
* +----------+ +-----------+ |
* CLOSING |ws.close()|<--|close frame|<--+-----+ CLOSING
* +----------+ +-----------+ |
* | | | +---+ |
* +------------------------+-->|fin| - - - -
* | +---+ | +---+
* - - - - -|fin|<---------------------+
* +---+
*
* @param {Number} code Status code explaining why the connection is closing
* @param {String} data A string explaining why the connection is closing
* @public
*/
close(code, data) {
if (this.readyState === WebSocket.CLOSED) return;
if (this.readyState === WebSocket.CONNECTING) {
const msg = 'WebSocket was closed before the connection was established';
return abortHandshake(this, this._req, msg);
}
if (this.readyState === WebSocket.CLOSING) {
if (this._closeFrameSent && this._closeFrameReceived) this._socket.end();
return;
}
this.readyState = WebSocket.CLOSING;
this._sender.close(code, data, !this._isServer, (err) => {
//
// This error is handled by the `'error'` listener on the socket. We only
// want to know if the close frame has been sent here.
//
if (err) return;
this._closeFrameSent = true;
if (this._closeFrameReceived) this._socket.end();
});
//
// Specify a timeout for the closing handshake to complete.
//
this._closeTimer = setTimeout(
this._socket.destroy.bind(this._socket),
closeTimeout
);
}
/**
* Send a ping.
*
* @param {*} data The data to send
* @param {Boolean} mask Indicates whether or not to mask `data`
* @param {Function} cb Callback which is executed when the ping is sent
* @public
*/
ping(data, mask, cb) {
if (typeof data === 'function') {
cb = data;
data = mask = undefined;
} else if (typeof mask === 'function') {
cb = mask;
mask = undefined;
}
if (this.readyState !== WebSocket.OPEN) {
const err = new Error(
`WebSocket is not open: readyState ${this.readyState} ` +
`(${readyStates[this.readyState]})`
);
if (cb) return cb(err);
throw err;
}
if (typeof data === 'number') data = data.toString();
if (mask === undefined) mask = !this._isServer;
this._sender.ping(data || EMPTY_BUFFER, mask, cb);
}
/**
* Send a pong.
*
* @param {*} data The data to send
* @param {Boolean} mask Indicates whether or not to mask `data`
* @param {Function} cb Callback which is executed when the pong is sent
* @public
*/
pong(data, mask, cb) {
if (typeof data === 'function') {
cb = data;
data = mask = undefined;
} else if (typeof mask === 'function') {
cb = mask;
mask = undefined;
}
if (this.readyState !== WebSocket.OPEN) {
const err = new Error(
`WebSocket is not open: readyState ${this.readyState} ` +
`(${readyStates[this.readyState]})`
);
if (cb) return cb(err);
throw err;
}
if (typeof data === 'number') data = data.toString();
if (mask === undefined) mask = !this._isServer;
this._sender.pong(data || EMPTY_BUFFER, mask, cb);
}
/**
* Send a data message.
*
* @param {*} data The message to send
* @param {Object} options Options object
* @param {Boolean} options.compress Specifies whether or not to compress `data`
* @param {Boolean} options.binary Specifies whether `data` is binary or text
* @param {Boolean} options.fin Specifies whether the fragment is the last one
* @param {Boolean} options.mask Specifies whether or not to mask `data`
* @param {Function} cb Callback which is executed when data is written out
* @public
*/
send(data, options, cb) {
if (typeof options === 'function') {
cb = options;
options = {};
}
if (this.readyState !== WebSocket.OPEN) {
const err = new Error(
`WebSocket is not open: readyState ${this.readyState} ` +
`(${readyStates[this.readyState]})`
);
if (cb) return cb(err);
throw err;
}
if (typeof data === 'number') data = data.toString();
const opts = Object.assign(
{
binary: typeof data !== 'string',
mask: !this._isServer,
compress: true,
fin: true
},
options
);
if (!this._extensions[PerMessageDeflate.extensionName]) {
opts.compress = false;
}
this._sender.send(data || EMPTY_BUFFER, opts, cb);
}
/**
* Forcibly close the connection.
*
* @public
*/
terminate() {
if (this.readyState === WebSocket.CLOSED) return;
if (this.readyState === WebSocket.CONNECTING) {
const msg = 'WebSocket was closed before the connection was established';
return abortHandshake(this, this._req, msg);
}
if (this._socket) {
this.readyState = WebSocket.CLOSING;
this._socket.destroy();
}
}
}
readyStates.forEach((readyState, i) => {
WebSocket[readyState] = i;
});
//
// Add the `onopen`, `onerror`, `onclose`, and `onmessage` attributes.
// See https://html.spec.whatwg.org/multipage/comms.html#the-websocket-interface
//
['open', 'error', 'close', 'message'].forEach((method) => {
Object.defineProperty(WebSocket.prototype, `on${method}`, {
/**
* Return the listener of the event.
*
* @return {(Function|undefined)} The event listener or `undefined`
* @public
*/
get() {
const listeners = this.listeners(method);
for (var i = 0; i < listeners.length; i++) {
if (listeners[i]._listener) return listeners[i]._listener;
}
return undefined;
},
/**
* Add a listener for the event.
*
* @param {Function} listener The listener to add
* @public
*/
set(listener) {
const listeners = this.listeners(method);
for (var i = 0; i < listeners.length; i++) {
//
// Remove only the listeners added via `addEventListener`.
//
if (listeners[i]._listener) this.removeListener(method, listeners[i]);
}
this.addEventListener(method, listener);
}
});
});
WebSocket.prototype.addEventListener = EventTarget.addEventListener;
WebSocket.prototype.removeEventListener = EventTarget.removeEventListener;
module.exports = WebSocket;
/**
* Initialize a WebSocket client.
*
* @param {WebSocket} websocket The client to initialize
* @param {(String|url.Url|url.URL)} address The URL to which to connect
* @param {String} protocols The subprotocols
* @param {Object} options Connection options
* @param {(Boolean|Object)} options.perMessageDeflate Enable/disable
* permessage-deflate
* @param {Number} options.handshakeTimeout Timeout in milliseconds for the
* handshake request
* @param {Number} options.protocolVersion Value of the `Sec-WebSocket-Version`
* header
* @param {String} options.origin Value of the `Origin` or
* `Sec-WebSocket-Origin` header
* @param {Number} options.maxPayload The maximum allowed message size
* @param {Boolean} options.followRedirects Whether or not to follow redirects
* @param {Number} options.maxRedirects The maximum number of redirects allowed
* @private
*/
function initAsClient(websocket, address, protocols, options) {
const opts = Object.assign(
{
protocolVersion: protocolVersions[1],
maxPayload: 100 * 1024 * 1024,
perMessageDeflate: true,
followRedirects: false,
maxRedirects: 10
},
options,
{
createConnection: undefined,
socketPath: undefined,
hostname: undefined,
protocol: undefined,
timeout: undefined,
method: undefined,
auth: undefined,
host: undefined,
path: undefined,
port: undefined
}
);
if (!protocolVersions.includes(opts.protocolVersion)) {
throw new RangeError(
`Unsupported protocol version: ${opts.protocolVersion} ` +
`(supported versions: ${protocolVersions.join(', ')})`
);
}
var parsedUrl;
if (typeof address === 'object' && address.href !== undefined) {
parsedUrl = address;
websocket.url = address.href;
} else {
//
// The WHATWG URL constructor is not available on Node.js < 6.13.0
//
parsedUrl = url.URL ? new url.URL(address) : url.parse(address);
websocket.url = address;
}
const isUnixSocket = parsedUrl.protocol === 'ws+unix:';
if (!parsedUrl.host && (!isUnixSocket || !parsedUrl.pathname)) {
throw new Error(`Invalid URL: ${websocket.url}`);
}
const isSecure =
parsedUrl.protocol === 'wss:' || parsedUrl.protocol === 'https:';
const defaultPort = isSecure ? 443 : 80;
const key = crypto.randomBytes(16).toString('base64');
const get = isSecure ? https.get : http.get;
const path = parsedUrl.search
? `${parsedUrl.pathname || '/'}${parsedUrl.search}`
: parsedUrl.pathname || '/';
var perMessageDeflate;
opts.createConnection = isSecure ? tlsConnect : netConnect;
opts.defaultPort = opts.defaultPort || defaultPort;
opts.port = parsedUrl.port || defaultPort;
opts.host = parsedUrl.hostname.startsWith('[')
? parsedUrl.hostname.slice(1, -1)
: parsedUrl.hostname;
opts.headers = Object.assign(
{
'Sec-WebSocket-Version': opts.protocolVersion,
'Sec-WebSocket-Key': key,
Connection: 'Upgrade',
Upgrade: 'websocket'
},
opts.headers
);
opts.path = path;
opts.timeout = opts.handshakeTimeout;
if (opts.perMessageDeflate) {
perMessageDeflate = new PerMessageDeflate(
opts.perMessageDeflate !== true ? opts.perMessageDeflate : {},
false,
opts.maxPayload
);
opts.headers['Sec-WebSocket-Extensions'] = extension.format({
[PerMessageDeflate.extensionName]: perMessageDeflate.offer()
});
}
if (protocols) {
opts.headers['Sec-WebSocket-Protocol'] = protocols;
}
if (opts.origin) {
if (opts.protocolVersion < 13) {
opts.headers['Sec-WebSocket-Origin'] = opts.origin;
} else {
opts.headers.Origin = opts.origin;
}
}
if (parsedUrl.auth) {
opts.auth = parsedUrl.auth;
} else if (parsedUrl.username || parsedUrl.password) {
opts.auth = `${parsedUrl.username}:${parsedUrl.password}`;
}
if (isUnixSocket) {
const parts = path.split(':');
opts.socketPath = parts[0];
opts.path = parts[1];
}
var req = (websocket._req = get(opts));
if (opts.timeout) {
req.on('timeout', () => {
abortHandshake(websocket, req, 'Opening handshake has timed out');
});
}
req.on('error', (err) => {
if (websocket._req.aborted) return;
req = websocket._req = null;
websocket.readyState = WebSocket.CLOSING;
websocket.emit('error', err);
websocket.emitClose();
});
req.on('response', (res) => {
const location = res.headers.location;
const statusCode = res.statusCode;
if (
location &&
opts.followRedirects &&
statusCode >= 300 &&
statusCode < 400
) {
if (++websocket._redirects > opts.maxRedirects) {
abortHandshake(websocket, req, 'Maximum redirects exceeded');
return;
}
req.abort();
const addr = url.URL
? new url.URL(location, address)
: url.resolve(address, location);
initAsClient(websocket, addr, protocols, options);
} else if (!websocket.emit('unexpected-response', req, res)) {
abortHandshake(
websocket,
req,
`Unexpected server response: ${res.statusCode}`
);
}
});
req.on('upgrade', (res, socket, head) => {
websocket.emit('upgrade', res);
//
// The user may have closed the connection from a listener of the `upgrade`
// event.
//
if (websocket.readyState !== WebSocket.CONNECTING) return;
req = websocket._req = null;
const digest = crypto
.createHash('sha1')
.update(key + GUID)
.digest('base64');
if (res.headers['sec-websocket-accept'] !== digest) {
abortHandshake(websocket, socket, 'Invalid Sec-WebSocket-Accept header');
return;
}
const serverProt = res.headers['sec-websocket-protocol'];
const protList = (protocols || '').split(/, */);
var protError;
if (!protocols && serverProt) {
protError = 'Server sent a subprotocol but none was requested';
} else if (protocols && !serverProt) {
protError = 'Server sent no subprotocol';
} else if (serverProt && !protList.includes(serverProt)) {
protError = 'Server sent an invalid subprotocol';
}
if (protError) {
abortHandshake(websocket, socket, protError);
return;
}
if (serverProt) websocket.protocol = serverProt;
if (perMessageDeflate) {
try {
const extensions = extension.parse(
res.headers['sec-websocket-extensions']
);
if (extensions[PerMessageDeflate.extensionName]) {
perMessageDeflate.accept(extensions[PerMessageDeflate.extensionName]);
websocket._extensions[
PerMessageDeflate.extensionName
] = perMessageDeflate;
}
} catch (err) {
abortHandshake(
websocket,
socket,
'Invalid Sec-WebSocket-Extensions header'
);
return;
}
}
websocket.setSocket(socket, head, opts.maxPayload);
});
}
/**
* Create a `net.Socket` and initiate a connection.
*
* @param {Object} options Connection options
* @return {net.Socket} The newly created socket used to start the connection
* @private
*/
function netConnect(options) {
//
// Override `options.path` only if `options` is a copy of the original options
// object. This is always true on Node.js >= 8 but not on Node.js 6 where
// `options.socketPath` might be `undefined` even if the `socketPath` option
// was originally set.
//
if (options.protocolVersion) options.path = options.socketPath;
return net.connect(options);
}
/**
* Create a `tls.TLSSocket` and initiate a connection.
*
* @param {Object} options Connection options
* @return {tls.TLSSocket} The newly created socket used to start the connection
* @private
*/
function tlsConnect(options) {
options.path = undefined;
options.servername = options.servername || options.host;
return tls.connect(options);
}
/**
* Abort the handshake and emit an error.
*
* @param {WebSocket} websocket The WebSocket instance
* @param {(http.ClientRequest|net.Socket)} stream The request to abort or the
* socket to destroy
* @param {String} message The error message
* @private
*/
function abortHandshake(websocket, stream, message) {
websocket.readyState = WebSocket.CLOSING;
const err = new Error(message);
Error.captureStackTrace(err, abortHandshake);
if (stream.setHeader) {
stream.abort();
stream.once('abort', websocket.emitClose.bind(websocket));
websocket.emit('error', err);
} else {
stream.destroy(err);
stream.once('error', websocket.emit.bind(websocket, 'error'));
stream.once('close', websocket.emitClose.bind(websocket));
}
}
/**
* The listener of the `Receiver` `'conclude'` event.
*
* @param {Number} code The status code
* @param {String} reason The reason for closing
* @private
*/
function receiverOnConclude(code, reason) {
const websocket = this[kWebSocket];
websocket._socket.removeListener('data', socketOnData);
websocket._socket.resume();
websocket._closeFrameReceived = true;
websocket._closeMessage = reason;
websocket._closeCode = code;
if (code === 1005) websocket.close();
else websocket.close(code, reason);
}
/**
* The listener of the `Receiver` `'drain'` event.
*
* @private
*/
function receiverOnDrain() {
this[kWebSocket]._socket.resume();
}
/**
* The listener of the `Receiver` `'error'` event.
*
* @param {(RangeError|Error)} err The emitted error
* @private
*/
function receiverOnError(err) {
const websocket = this[kWebSocket];
websocket._socket.removeListener('data', socketOnData);
websocket.readyState = WebSocket.CLOSING;
websocket._closeCode = err[kStatusCode];
websocket.emit('error', err);
websocket._socket.destroy();
}
/**
* The listener of the `Receiver` `'finish'` event.
*
* @private
*/
function receiverOnFinish() {
this[kWebSocket].emitClose();
}
/**
* The listener of the `Receiver` `'message'` event.
*
* @param {(String|Buffer|ArrayBuffer|Buffer[])} data The message
* @private
*/
function receiverOnMessage(data) {
this[kWebSocket].emit('message', data);
}
/**
* The listener of the `Receiver` `'ping'` event.
*
* @param {Buffer} data The data included in the ping frame
* @private
*/
function receiverOnPing(data) {
const websocket = this[kWebSocket];
websocket.pong(data, !websocket._isServer, NOOP);
websocket.emit('ping', data);
}
/**
* The listener of the `Receiver` `'pong'` event.
*
* @param {Buffer} data The data included in the pong frame
* @private
*/
function receiverOnPong(data) {
this[kWebSocket].emit('pong', data);
}
/**
* The listener of the `net.Socket` `'close'` event.
*
* @private
*/
function socketOnClose() {
const websocket = this[kWebSocket];
this.removeListener('close', socketOnClose);
this.removeListener('end', socketOnEnd);
websocket.readyState = WebSocket.CLOSING;
//
// The close frame might not have been received or the `'end'` event emitted,
// for example, if the socket was destroyed due to an error. Ensure that the
// `receiver` stream is closed after writing any remaining buffered data to
// it. If the readable side of the socket is in flowing mode then there is no
// buffered data as everything has been already written and `readable.read()`
// will return `null`. If instead, the socket is paused, any possible buffered
// data will be read as a single chunk and emitted synchronously in a single
// `'data'` event.
//
websocket._socket.read();
websocket._receiver.end();
this.removeListener('data', socketOnData);
this[kWebSocket] = undefined;
clearTimeout(websocket._closeTimer);
if (
websocket._receiver._writableState.finished ||
websocket._receiver._writableState.errorEmitted
) {
websocket.emitClose();
} else {
websocket._receiver.on('error', receiverOnFinish);
websocket._receiver.on('finish', receiverOnFinish);
}
}
/**
* The listener of the `net.Socket` `'data'` event.
*
* @param {Buffer} chunk A chunk of data
* @private
*/
function socketOnData(chunk) {
if (!this[kWebSocket]._receiver.write(chunk)) {
this.pause();
}
}
/**
* The listener of the `net.Socket` `'end'` event.
*
* @private
*/
function socketOnEnd() {
const websocket = this[kWebSocket];
websocket.readyState = WebSocket.CLOSING;
websocket._receiver.end();
this.end();
}
/**
* The listener of the `net.Socket` `'error'` event.
*
* @private
*/
function socketOnError() {
const websocket = this[kWebSocket];
this.removeListener('error', socketOnError);
this.on('error', NOOP);
websocket.readyState = WebSocket.CLOSING;
this.destroy();
}

45
node_modules/y-websocket/node_modules/ws/package.json generated vendored Normal file
View File

@@ -0,0 +1,45 @@
{
"name": "ws",
"version": "6.2.3",
"description": "Simple to use, blazing fast and thoroughly tested websocket client and server for Node.js",
"keywords": [
"HyBi",
"Push",
"RFC-6455",
"WebSocket",
"WebSockets",
"real-time"
],
"homepage": "https://github.com/websockets/ws",
"bugs": "https://github.com/websockets/ws/issues",
"repository": "websockets/ws",
"author": "Einar Otto Stangvik <einaros@gmail.com> (http://2x.io)",
"license": "MIT",
"main": "index.js",
"browser": "browser.js",
"files": [
"browser.js",
"index.js",
"lib/*.js"
],
"scripts": {
"test": "npm run lint && nyc --reporter=html --reporter=text mocha test/*.test.js",
"integration": "npm run lint && mocha test/*.integration.js",
"lint": "eslint --ignore-path .gitignore . && prettier --check --ignore-path .gitignore \"**/*.{json,md,yml}\""
},
"dependencies": {
"async-limiter": "~1.0.0"
},
"devDependencies": {
"benchmark": "~2.1.4",
"bufferutil": "~4.0.0",
"coveralls": "~3.0.3",
"eslint": "~5.15.0",
"eslint-config-prettier": "~4.1.0",
"eslint-plugin-prettier": "~3.0.0",
"mocha": "~6.0.0",
"nyc": "~13.3.0",
"prettier": "~1.16.1",
"utf-8-validate": "~5.0.0"
}
}

82
node_modules/y-websocket/package.json generated vendored Normal file
View File

@@ -0,0 +1,82 @@
{
"name": "y-websocket",
"version": "1.5.4",
"description": "Websockets provider for Yjs",
"main": "./dist/y-websocket.cjs",
"module": "./src/y-websocket.js",
"types": "./dist/src/y-websocket.d.ts",
"sideEffects": false,
"funding": {
"type": "GitHub Sponsors ❤",
"url": "https://github.com/sponsors/dmonad"
},
"scripts": {
"start": "node ./bin/server.js",
"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"
},
"bin": {
"y-websocket-server": "./bin/server.js",
"y-websocket": "./bin/server.js"
},
"files": [
"dist/*",
"bin/*",
"src/*"
],
"exports": {
"./package.json": "./package.json",
"./bin/utils": "./bin/utils.js",
"./bin/callback": "./bin/callback.js",
".": {
"module": "./src/y-websocket.js",
"import": "./src/y-websocket.js",
"require": "./dist/y-websocket.cjs",
"types": "./dist/src/y-websocket.d.ts"
}
},
"repository": {
"type": "git",
"url": "git+https://github.com/yjs/y-websocket.git"
},
"keywords": [
"Yjs"
],
"author": "Kevin Jahns <kevin.jahns@protonmail.com>",
"license": "MIT",
"bugs": {
"url": "https://github.com/yjs/y-websocket/issues"
},
"homepage": "https://github.com/yjs/y-websocket#readme",
"standard": {
"ignore": [
"/dist",
"/node_modules"
]
},
"dependencies": {
"lib0": "^0.2.52",
"lodash.debounce": "^4.0.8",
"y-protocols": "^1.0.5"
},
"devDependencies": {
"@types/node": "^18.15.0",
"rollup": "^3.19.1",
"standard": "^12.0.1",
"typescript": "^4.9.5",
"yjs": "^13.5.0"
},
"peerDependencies": {
"yjs": "^13.5.6"
},
"optionalDependencies": {
"ws": "^6.2.1",
"y-leveldb": "^0.1.0"
},
"engines": {
"npm": ">=8.0.0",
"node": ">=16.0.0"
}
}

485
node_modules/y-websocket/src/y-websocket.js generated vendored Normal file
View File

@@ -0,0 +1,485 @@
/**
* @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 { Observable } 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<function(encoding.Encoder, decoding.Decoder, WebsocketProvider, boolean, number):void>}
*/
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
}
/**
* @param {WebsocketProvider} provider
*/
const setupWS = (provider) => {
if (provider.shouldConnect && provider.ws === null) {
const websocket = new provider._WS(provider.url)
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) => {
provider.emit('connection-close', [event, provider])
provider.ws = null
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
)
}
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 {Observable<string>}
*/
export class WebsocketProvider extends Observable {
/**
* @param {string} serverUrl
* @param {string} roomname
* @param {Y.Doc} doc
* @param {object} opts
* @param {boolean} [opts.connect]
* @param {awarenessProtocol.Awareness} [opts.awareness]
* @param {Object<string,string>} [opts.params]
* @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 = {},
WebSocketPolyfill = WebSocket,
resyncInterval = -1,
maxBackoffTime = 2500,
disableBc = false
} = {}) {
super()
// ensure that url is always ends with /
while (serverUrl[serverUrl.length - 1] === '/') {
serverUrl = serverUrl.slice(0, serverUrl.length - 1)
}
const encodedParams = url.encodeQueryParams(params)
this.maxBackoffTime = maxBackoffTime
this.bcChannel = serverUrl + '/' + roomname
this.url = serverUrl + '/' + roomname +
(encodedParams.length === 0 ? '' : '?' + encodedParams)
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)
/** @type {WebSocket} */ (this.ws).close()
}
}, messageReconnectTimeout / 10))
if (connect) {
this.connect()
}
}
/**
* @type {boolean}
*/
get synced () {
return this._synced
}
set synced (state) {
if (this._synced !== state) {
this._synced = state
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) {
this.ws.close()
}
}
connect () {
this.shouldConnect = true
if (!this.wsconnected && this.ws === null) {
setupWS(this)
this.connectBc()
}
}
}