JavaScript 65.5%
Python 17.8%
CSS 13%
HTML 3.7%
1/* eslint no-unused-vars: ["error", { "varsIgnorePattern": "^Duplex" }] */23'use strict';45const { Duplex } = require('stream');6const { randomFillSync } = require('crypto');7const {8 types: { isUint8Array }9} = require('util');1011const PerMessageDeflate = require('./permessage-deflate');12const { EMPTY_BUFFER, kWebSocket, NOOP } = require('./constants');13const { isBlob, isValidStatusCode } = require('./validation');14const { mask: applyMask, toBuffer } = require('./buffer-util');1516const kByteLength = Symbol('kByteLength');17const maskBuffer = Buffer.alloc(4);18const RANDOM_POOL_SIZE = 8 * 1024;19let randomPool;20let randomPoolPointer = RANDOM_POOL_SIZE;2122const DEFAULT = 0;23const DEFLATING = 1;24const GET_BLOB_DATA = 2;2526/**27 * HyBi Sender implementation.28 */29class Sender {30 /**31 * Creates a Sender instance.32 *33 * @param {Duplex} socket The connection socket34 * @param {Object} [extensions] An object containing the negotiated extensions35 * @param {Function} [generateMask] The function used to generate the masking36 * key37 */38 constructor(socket, extensions, generateMask) {39 this._extensions = extensions || {};4041 if (generateMask) {42 this._generateMask = generateMask;43 this._maskBuffer = Buffer.alloc(4);44 }4546 this._socket = socket;4748 this._firstFragment = true;49 this._compress = false;5051 this._bufferedBytes = 0;52 this._queue = [];53 this._state = DEFAULT;54 this.onerror = NOOP;55 this[kWebSocket] = undefined;56 }5758 /**59 * Frames a piece of data according to the HyBi WebSocket protocol.60 *61 * @param {(Buffer|String)} data The data to frame62 * @param {Object} options Options object63 * @param {Boolean} [options.fin=false] Specifies whether or not to set the64 * FIN bit65 * @param {Function} [options.generateMask] The function used to generate the66 * masking key67 * @param {Boolean} [options.mask=false] Specifies whether or not to mask68 * `data`69 * @param {Buffer} [options.maskBuffer] The buffer used to store the masking70 * key71 * @param {Number} options.opcode The opcode72 * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be73 * modified74 * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the75 * RSV1 bit76 * @return {(Buffer|String)[]} The framed data77 * @public78 */79 static frame(data, options) {80 let mask;81 let merge = false;82 let offset = 2;83 let skipMasking = false;8485 if (options.mask) {86 mask = options.maskBuffer || maskBuffer;8788 if (options.generateMask) {89 options.generateMask(mask);90 } else {91 if (randomPoolPointer === RANDOM_POOL_SIZE) {92 /* istanbul ignore else */93 if (randomPool === undefined) {94 //95 // This is lazily initialized because server-sent frames must not96 // be masked so it may never be used.97 //98 randomPool = Buffer.alloc(RANDOM_POOL_SIZE);99 }100101 randomFillSync(randomPool, 0, RANDOM_POOL_SIZE);102 randomPoolPointer = 0;103 }104105 mask[0] = randomPool[randomPoolPointer++];106 mask[1] = randomPool[randomPoolPointer++];107 mask[2] = randomPool[randomPoolPointer++];108 mask[3] = randomPool[randomPoolPointer++];109 }110111 skipMasking = (mask[0] | mask[1] | mask[2] | mask[3]) === 0;112 offset = 6;113 }114115 let dataLength;116117 if (typeof data === 'string') {118 if (119 (!options.mask || skipMasking) &&120 options[kByteLength] !== undefined121 ) {122 dataLength = options[kByteLength];123 } else {124 data = Buffer.from(data);125 dataLength = data.length;126 }127 } else {128 dataLength = data.length;129 merge = options.mask && options.readOnly && !skipMasking;130 }131132 let payloadLength = dataLength;133134 if (dataLength >= 65536) {135 offset += 8;136 payloadLength = 127;137 } else if (dataLength > 125) {138 offset += 2;139 payloadLength = 126;140 }141142 const target = Buffer.allocUnsafe(merge ? dataLength + offset : offset);143144 target[0] = options.fin ? options.opcode | 0x80 : options.opcode;145 if (options.rsv1) target[0] |= 0x40;146147 target[1] = payloadLength;148149 if (payloadLength === 126) {150 target.writeUInt16BE(dataLength, 2);151 } else if (payloadLength === 127) {152 target[2] = target[3] = 0;153 target.writeUIntBE(dataLength, 4, 6);154 }155156 if (!options.mask) return [target, data];157158 target[1] |= 0x80;159 target[offset - 4] = mask[0];160 target[offset - 3] = mask[1];161 target[offset - 2] = mask[2];162 target[offset - 1] = mask[3];163164 if (skipMasking) return [target, data];165166 if (merge) {167 applyMask(data, mask, target, offset, dataLength);168 return [target];169 }170171 applyMask(data, mask, data, 0, dataLength);172 return [target, data];173 }174175 /**176 * Sends a close message to the other peer.177 *178 * @param {Number} [code] The status code component of the body179 * @param {(String|Buffer)} [data] The message component of the body180 * @param {Boolean} [mask=false] Specifies whether or not to mask the message181 * @param {Function} [cb] Callback182 * @public183 */184 close(code, data, mask, cb) {185 let buf;186187 if (code === undefined) {188 buf = EMPTY_BUFFER;189 } else if (typeof code !== 'number' || !isValidStatusCode(code)) {190 throw new TypeError('First argument must be a valid error code number');191 } else if (data === undefined || !data.length) {192 buf = Buffer.allocUnsafe(2);193 buf.writeUInt16BE(code, 0);194 } else {195 const length = Buffer.byteLength(data);196197 if (length > 123) {198 throw new RangeError('The message must not be greater than 123 bytes');199 }200201 buf = Buffer.allocUnsafe(2 + length);202 buf.writeUInt16BE(code, 0);203204 if (typeof data === 'string') {205 buf.write(data, 2);206 } else if (isUint8Array(data)) {207 buf.set(data, 2);208 } else {209 throw new TypeError('Second argument must be a string or a Uint8Array');210 }211 }212213 const options = {214 [kByteLength]: buf.length,215 fin: true,216 generateMask: this._generateMask,217 mask,218 maskBuffer: this._maskBuffer,219 opcode: 0x08,220 readOnly: false,221 rsv1: false222 };223224 if (this._state !== DEFAULT) {225 this.enqueue([this.dispatch, buf, false, options, cb]);226 } else {227 this.sendFrame(Sender.frame(buf, options), cb);228 }229 }230231 /**232 * Sends a ping message to the other peer.233 *234 * @param {*} data The message to send235 * @param {Boolean} [mask=false] Specifies whether or not to mask `data`236 * @param {Function} [cb] Callback237 * @public238 */239 ping(data, mask, cb) {240 let byteLength;241 let readOnly;242243 if (typeof data === 'string') {244 byteLength = Buffer.byteLength(data);245 readOnly = false;246 } else if (isBlob(data)) {247 byteLength = data.size;248 readOnly = false;249 } else {250 data = toBuffer(data);251 byteLength = data.length;252 readOnly = toBuffer.readOnly;253 }254255 if (byteLength > 125) {256 throw new RangeError('The data size must not be greater than 125 bytes');257 }258259 const options = {260 [kByteLength]: byteLength,261 fin: true,262 generateMask: this._generateMask,263 mask,264 maskBuffer: this._maskBuffer,265 opcode: 0x09,266 readOnly,267 rsv1: false268 };269270 if (isBlob(data)) {271 if (this._state !== DEFAULT) {272 this.enqueue([this.getBlobData, data, false, options, cb]);273 } else {274 this.getBlobData(data, false, options, cb);275 }276 } else if (this._state !== DEFAULT) {277 this.enqueue([this.dispatch, data, false, options, cb]);278 } else {279 this.sendFrame(Sender.frame(data, options), cb);280 }281 }282283 /**284 * Sends a pong message to the other peer.285 *286 * @param {*} data The message to send287 * @param {Boolean} [mask=false] Specifies whether or not to mask `data`288 * @param {Function} [cb] Callback289 * @public290 */291 pong(data, mask, cb) {292 let byteLength;293 let readOnly;294295 if (typeof data === 'string') {296 byteLength = Buffer.byteLength(data);297 readOnly = false;298 } else if (isBlob(data)) {299 byteLength = data.size;300 readOnly = false;301 } else {302 data = toBuffer(data);303 byteLength = data.length;304 readOnly = toBuffer.readOnly;305 }306307 if (byteLength > 125) {308 throw new RangeError('The data size must not be greater than 125 bytes');309 }310311 const options = {312 [kByteLength]: byteLength,313 fin: true,314 generateMask: this._generateMask,315 mask,316 maskBuffer: this._maskBuffer,317 opcode: 0x0a,318 readOnly,319 rsv1: false320 };321322 if (isBlob(data)) {323 if (this._state !== DEFAULT) {324 this.enqueue([this.getBlobData, data, false, options, cb]);325 } else {326 this.getBlobData(data, false, options, cb);327 }328 } else if (this._state !== DEFAULT) {329 this.enqueue([this.dispatch, data, false, options, cb]);330 } else {331 this.sendFrame(Sender.frame(data, options), cb);332 }333 }334335 /**336 * Sends a data message to the other peer.337 *338 * @param {*} data The message to send339 * @param {Object} options Options object340 * @param {Boolean} [options.binary=false] Specifies whether `data` is binary341 * or text342 * @param {Boolean} [options.compress=false] Specifies whether or not to343 * compress `data`344 * @param {Boolean} [options.fin=false] Specifies whether the fragment is the345 * last one346 * @param {Boolean} [options.mask=false] Specifies whether or not to mask347 * `data`348 * @param {Function} [cb] Callback349 * @public350 */351 send(data, options, cb) {352 const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];353 let opcode = options.binary ? 2 : 1;354 let rsv1 = options.compress;355356 let byteLength;357 let readOnly;358359 if (typeof data === 'string') {360 byteLength = Buffer.byteLength(data);361 readOnly = false;362 } else if (isBlob(data)) {363 byteLength = data.size;364 readOnly = false;365 } else {366 data = toBuffer(data);367 byteLength = data.length;368 readOnly = toBuffer.readOnly;369 }370371 if (this._firstFragment) {372 this._firstFragment = false;373 if (374 rsv1 &&375 perMessageDeflate &&376 perMessageDeflate.params[377 perMessageDeflate._isServer378 ? 'server_no_context_takeover'379 : 'client_no_context_takeover'380 ]381 ) {382 rsv1 = byteLength >= perMessageDeflate._threshold;383 }384 this._compress = rsv1;385 } else {386 rsv1 = false;387 opcode = 0;388 }389390 if (options.fin) this._firstFragment = true;391392 const opts = {393 [kByteLength]: byteLength,394 fin: options.fin,395 generateMask: this._generateMask,396 mask: options.mask,397 maskBuffer: this._maskBuffer,398 opcode,399 readOnly,400 rsv1401 };402403 if (isBlob(data)) {404 if (this._state !== DEFAULT) {405 this.enqueue([this.getBlobData, data, this._compress, opts, cb]);406 } else {407 this.getBlobData(data, this._compress, opts, cb);408 }409 } else if (this._state !== DEFAULT) {410 this.enqueue([this.dispatch, data, this._compress, opts, cb]);411 } else {412 this.dispatch(data, this._compress, opts, cb);413 }414 }415416 /**417 * Gets the contents of a blob as binary data.418 *419 * @param {Blob} blob The blob420 * @param {Boolean} [compress=false] Specifies whether or not to compress421 * the data422 * @param {Object} options Options object423 * @param {Boolean} [options.fin=false] Specifies whether or not to set the424 * FIN bit425 * @param {Function} [options.generateMask] The function used to generate the426 * masking key427 * @param {Boolean} [options.mask=false] Specifies whether or not to mask428 * `data`429 * @param {Buffer} [options.maskBuffer] The buffer used to store the masking430 * key431 * @param {Number} options.opcode The opcode432 * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be433 * modified434 * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the435 * RSV1 bit436 * @param {Function} [cb] Callback437 * @private438 */439 getBlobData(blob, compress, options, cb) {440 this._bufferedBytes += options[kByteLength];441 this._state = GET_BLOB_DATA;442443 blob444 .arrayBuffer()445 .then((arrayBuffer) => {446 if (this._socket.destroyed) {447 const err = new Error(448 'The socket was closed while the blob was being read'449 );450451 //452 // `callCallbacks` is called in the next tick to ensure that errors453 // that might be thrown in the callbacks behave like errors thrown454 // outside the promise chain.455 //456 process.nextTick(callCallbacks, this, err, cb);457 return;458 }459460 this._bufferedBytes -= options[kByteLength];461 const data = toBuffer(arrayBuffer);462463 if (!compress) {464 this._state = DEFAULT;465 this.sendFrame(Sender.frame(data, options), cb);466 this.dequeue();467 } else {468 this.dispatch(data, compress, options, cb);469 }470 })471 .catch((err) => {472 //473 // `onError` is called in the next tick for the same reason that474 // `callCallbacks` above is.475 //476 process.nextTick(onError, this, err, cb);477 });478 }479480 /**481 * Dispatches a message.482 *483 * @param {(Buffer|String)} data The message to send484 * @param {Boolean} [compress=false] Specifies whether or not to compress485 * `data`486 * @param {Object} options Options object487 * @param {Boolean} [options.fin=false] Specifies whether or not to set the488 * FIN bit489 * @param {Function} [options.generateMask] The function used to generate the490 * masking key491 * @param {Boolean} [options.mask=false] Specifies whether or not to mask492 * `data`493 * @param {Buffer} [options.maskBuffer] The buffer used to store the masking494 * key495 * @param {Number} options.opcode The opcode496 * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be497 * modified498 * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the499 * RSV1 bit500 * @param {Function} [cb] Callback501 * @private502 */503 dispatch(data, compress, options, cb) {504 if (!compress) {505 this.sendFrame(Sender.frame(data, options), cb);506 return;507 }508509 const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];510511 this._bufferedBytes += options[kByteLength];512 this._state = DEFLATING;513 perMessageDeflate.compress(data, options.fin, (_, buf) => {514 if (this._socket.destroyed) {515 const err = new Error(516 'The socket was closed while data was being compressed'517 );518519 callCallbacks(this, err, cb);520 return;521 }522523 this._bufferedBytes -= options[kByteLength];524 this._state = DEFAULT;525 options.readOnly = false;526 this.sendFrame(Sender.frame(buf, options), cb);527 this.dequeue();528 });529 }530531 /**532 * Executes queued send operations.533 *534 * @private535 */536 dequeue() {537 while (this._state === DEFAULT && this._queue.length) {538 const params = this._queue.shift();539540 this._bufferedBytes -= params[3][kByteLength];541 Reflect.apply(params[0], this, params.slice(1));542 }543 }544545 /**546 * Enqueues a send operation.547 *548 * @param {Array} params Send operation parameters.549 * @private550 */551 enqueue(params) {552 this._bufferedBytes += params[3][kByteLength];553 this._queue.push(params);554 }555556 /**557 * Sends a frame.558 *559 * @param {(Buffer | String)[]} list The frame to send560 * @param {Function} [cb] Callback561 * @private562 */563 sendFrame(list, cb) {564 if (list.length === 2) {565 this._socket.cork();566 this._socket.write(list[0]);567 this._socket.write(list[1], cb);568 this._socket.uncork();569 } else {570 this._socket.write(list[0], cb);571 }572 }573}574575module.exports = Sender;576577/**578 * Calls queued callbacks with an error.579 *580 * @param {Sender} sender The `Sender` instance581 * @param {Error} err The error to call the callbacks with582 * @param {Function} [cb] The first callback583 * @private584 */585function callCallbacks(sender, err, cb) {586 if (typeof cb === 'function') cb(err);587588 for (let i = 0; i < sender._queue.length; i++) {589 const params = sender._queue[i];590 const callback = params[params.length - 1];591592 if (typeof callback === 'function') callback(err);593 }594}595596/**597 * Handles a `Sender` error.598 *599 * @param {Sender} sender The `Sender` instance600 * @param {Error} err The error601 * @param {Function} [cb] The first pending callback602 * @private603 */604function onError(sender, err, cb) {605 callCallbacks(sender, err, cb);606 sender.onerror(err);607}608