strong-tie/inbound-calls
0
1/* eslint no-unused-vars: ["error", { "varsIgnorePattern": "^Duplex" }] */2 3'use strict';4 5const { Duplex } = require('stream');6const { randomFillSync } = require('crypto');7 8const PerMessageDeflate = require('./permessage-deflate');9const { EMPTY_BUFFER, kWebSocket, NOOP } = require('./constants');10const { isBlob, isValidStatusCode } = require('./validation');11const { mask: applyMask, toBuffer } = require('./buffer-util');12 13const kByteLength = Symbol('kByteLength');14const maskBuffer = Buffer.alloc(4);15const RANDOM_POOL_SIZE = 8 * 1024;16let randomPool;17let randomPoolPointer = RANDOM_POOL_SIZE;18 19const DEFAULT = 0;20const DEFLATING = 1;21const GET_BLOB_DATA = 2;22 23/**24 * HyBi Sender implementation.25 */26class Sender {27 /**28 * Creates a Sender instance.29 *30 * @param {Duplex} socket The connection socket31 * @param {Object} [extensions] An object containing the negotiated extensions32 * @param {Function} [generateMask] The function used to generate the masking33 * key34 */35 constructor(socket, extensions, generateMask) {36 this._extensions = extensions || {};37 38 if (generateMask) {39 this._generateMask = generateMask;40 this._maskBuffer = Buffer.alloc(4);41 }42 43 this._socket = socket;44 45 this._firstFragment = true;46 this._compress = false;47 48 this._bufferedBytes = 0;49 this._queue = [];50 this._state = DEFAULT;51 this.onerror = NOOP;52 this[kWebSocket] = undefined;53 }54 55 /**56 * Frames a piece of data according to the HyBi WebSocket protocol.57 *58 * @param {(Buffer|String)} data The data to frame59 * @param {Object} options Options object60 * @param {Boolean} [options.fin=false] Specifies whether or not to set the61 * FIN bit62 * @param {Function} [options.generateMask] The function used to generate the63 * masking key64 * @param {Boolean} [options.mask=false] Specifies whether or not to mask65 * `data`66 * @param {Buffer} [options.maskBuffer] The buffer used to store the masking67 * key68 * @param {Number} options.opcode The opcode69 * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be70 * modified71 * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the72 * RSV1 bit73 * @return {(Buffer|String)[]} The framed data74 * @public75 */76 static frame(data, options) {77 let mask;78 let merge = false;79 let offset = 2;80 let skipMasking = false;81 82 if (options.mask) {83 mask = options.maskBuffer || maskBuffer;84 85 if (options.generateMask) {86 options.generateMask(mask);87 } else {88 if (randomPoolPointer === RANDOM_POOL_SIZE) {89 /* istanbul ignore else */90 if (randomPool === undefined) {91 //92 // This is lazily initialized because server-sent frames must not93 // be masked so it may never be used.94 //95 randomPool = Buffer.alloc(RANDOM_POOL_SIZE);96 }97 98 randomFillSync(randomPool, 0, RANDOM_POOL_SIZE);99 randomPoolPointer = 0;100 }101 102 mask[0] = randomPool[randomPoolPointer++];103 mask[1] = randomPool[randomPoolPointer++];104 mask[2] = randomPool[randomPoolPointer++];105 mask[3] = randomPool[randomPoolPointer++];106 }107 108 skipMasking = (mask[0] | mask[1] | mask[2] | mask[3]) === 0;109 offset = 6;110 }111 112 let dataLength;113 114 if (typeof data === 'string') {115 if (116 (!options.mask || skipMasking) &&117 options[kByteLength] !== undefined118 ) {119 dataLength = options[kByteLength];120 } else {121 data = Buffer.from(data);122 dataLength = data.length;123 }124 } else {125 dataLength = data.length;126 merge = options.mask && options.readOnly && !skipMasking;127 }128 129 let payloadLength = dataLength;130 131 if (dataLength >= 65536) {132 offset += 8;133 payloadLength = 127;134 } else if (dataLength > 125) {135 offset += 2;136 payloadLength = 126;137 }138 139 const target = Buffer.allocUnsafe(merge ? dataLength + offset : offset);140 141 target[0] = options.fin ? options.opcode | 0x80 : options.opcode;142 if (options.rsv1) target[0] |= 0x40;143 144 target[1] = payloadLength;145 146 if (payloadLength === 126) {147 target.writeUInt16BE(dataLength, 2);148 } else if (payloadLength === 127) {149 target[2] = target[3] = 0;150 target.writeUIntBE(dataLength, 4, 6);151 }152 153 if (!options.mask) return [target, data];154 155 target[1] |= 0x80;156 target[offset - 4] = mask[0];157 target[offset - 3] = mask[1];158 target[offset - 2] = mask[2];159 target[offset - 1] = mask[3];160 161 if (skipMasking) return [target, data];162 163 if (merge) {164 applyMask(data, mask, target, offset, dataLength);165 return [target];166 }167 168 applyMask(data, mask, data, 0, dataLength);169 return [target, data];170 }171 172 /**173 * Sends a close message to the other peer.174 *175 * @param {Number} [code] The status code component of the body176 * @param {(String|Buffer)} [data] The message component of the body177 * @param {Boolean} [mask=false] Specifies whether or not to mask the message178 * @param {Function} [cb] Callback179 * @public180 */181 close(code, data, mask, cb) {182 let buf;183 184 if (code === undefined) {185 buf = EMPTY_BUFFER;186 } else if (typeof code !== 'number' || !isValidStatusCode(code)) {187 throw new TypeError('First argument must be a valid error code number');188 } else if (data === undefined || !data.length) {189 buf = Buffer.allocUnsafe(2);190 buf.writeUInt16BE(code, 0);191 } else {192 const length = Buffer.byteLength(data);193 194 if (length > 123) {195 throw new RangeError('The message must not be greater than 123 bytes');196 }197 198 buf = Buffer.allocUnsafe(2 + length);199 buf.writeUInt16BE(code, 0);200 201 if (typeof data === 'string') {202 buf.write(data, 2);203 } else {204 buf.set(data, 2);205 }206 }207 208 const options = {209 [kByteLength]: buf.length,210 fin: true,211 generateMask: this._generateMask,212 mask,213 maskBuffer: this._maskBuffer,214 opcode: 0x08,215 readOnly: false,216 rsv1: false217 };218 219 if (this._state !== DEFAULT) {220 this.enqueue([this.dispatch, buf, false, options, cb]);221 } else {222 this.sendFrame(Sender.frame(buf, options), cb);223 }224 }225 226 /**227 * Sends a ping message to the other peer.228 *229 * @param {*} data The message to send230 * @param {Boolean} [mask=false] Specifies whether or not to mask `data`231 * @param {Function} [cb] Callback232 * @public233 */234 ping(data, mask, cb) {235 let byteLength;236 let readOnly;237 238 if (typeof data === 'string') {239 byteLength = Buffer.byteLength(data);240 readOnly = false;241 } else if (isBlob(data)) {242 byteLength = data.size;243 readOnly = false;244 } else {245 data = toBuffer(data);246 byteLength = data.length;247 readOnly = toBuffer.readOnly;248 }249 250 if (byteLength > 125) {251 throw new RangeError('The data size must not be greater than 125 bytes');252 }253 254 const options = {255 [kByteLength]: byteLength,256 fin: true,257 generateMask: this._generateMask,258 mask,259 maskBuffer: this._maskBuffer,260 opcode: 0x09,261 readOnly,262 rsv1: false263 };264 265 if (isBlob(data)) {266 if (this._state !== DEFAULT) {267 this.enqueue([this.getBlobData, data, false, options, cb]);268 } else {269 this.getBlobData(data, false, options, cb);270 }271 } else if (this._state !== DEFAULT) {272 this.enqueue([this.dispatch, data, false, options, cb]);273 } else {274 this.sendFrame(Sender.frame(data, options), cb);275 }276 }277 278 /**279 * Sends a pong message to the other peer.280 *281 * @param {*} data The message to send282 * @param {Boolean} [mask=false] Specifies whether or not to mask `data`283 * @param {Function} [cb] Callback284 * @public285 */286 pong(data, mask, cb) {287 let byteLength;288 let readOnly;289 290 if (typeof data === 'string') {291 byteLength = Buffer.byteLength(data);292 readOnly = false;293 } else if (isBlob(data)) {294 byteLength = data.size;295 readOnly = false;296 } else {297 data = toBuffer(data);298 byteLength = data.length;299 readOnly = toBuffer.readOnly;300 }301 302 if (byteLength > 125) {303 throw new RangeError('The data size must not be greater than 125 bytes');304 }305 306 const options = {307 [kByteLength]: byteLength,308 fin: true,309 generateMask: this._generateMask,310 mask,311 maskBuffer: this._maskBuffer,312 opcode: 0x0a,313 readOnly,314 rsv1: false315 };316 317 if (isBlob(data)) {318 if (this._state !== DEFAULT) {319 this.enqueue([this.getBlobData, data, false, options, cb]);320 } else {321 this.getBlobData(data, false, options, cb);322 }323 } else if (this._state !== DEFAULT) {324 this.enqueue([this.dispatch, data, false, options, cb]);325 } else {326 this.sendFrame(Sender.frame(data, options), cb);327 }328 }329 330 /**331 * Sends a data message to the other peer.332 *333 * @param {*} data The message to send334 * @param {Object} options Options object335 * @param {Boolean} [options.binary=false] Specifies whether `data` is binary336 * or text337 * @param {Boolean} [options.compress=false] Specifies whether or not to338 * compress `data`339 * @param {Boolean} [options.fin=false] Specifies whether the fragment is the340 * last one341 * @param {Boolean} [options.mask=false] Specifies whether or not to mask342 * `data`343 * @param {Function} [cb] Callback344 * @public345 */346 send(data, options, cb) {347 const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];348 let opcode = options.binary ? 2 : 1;349 let rsv1 = options.compress;350 351 let byteLength;352 let readOnly;353 354 if (typeof data === 'string') {355 byteLength = Buffer.byteLength(data);356 readOnly = false;357 } else if (isBlob(data)) {358 byteLength = data.size;359 readOnly = false;360 } else {361 data = toBuffer(data);362 byteLength = data.length;363 readOnly = toBuffer.readOnly;364 }365 366 if (this._firstFragment) {367 this._firstFragment = false;368 if (369 rsv1 &&370 perMessageDeflate &&371 perMessageDeflate.params[372 perMessageDeflate._isServer373 ? 'server_no_context_takeover'374 : 'client_no_context_takeover'375 ]376 ) {377 rsv1 = byteLength >= perMessageDeflate._threshold;378 }379 this._compress = rsv1;380 } else {381 rsv1 = false;382 opcode = 0;383 }384 385 if (options.fin) this._firstFragment = true;386 387 const opts = {388 [kByteLength]: byteLength,389 fin: options.fin,390 generateMask: this._generateMask,391 mask: options.mask,392 maskBuffer: this._maskBuffer,393 opcode,394 readOnly,395 rsv1396 };397 398 if (isBlob(data)) {399 if (this._state !== DEFAULT) {400 this.enqueue([this.getBlobData, data, this._compress, opts, cb]);401 } else {402 this.getBlobData(data, this._compress, opts, cb);403 }404 } else if (this._state !== DEFAULT) {405 this.enqueue([this.dispatch, data, this._compress, opts, cb]);406 } else {407 this.dispatch(data, this._compress, opts, cb);408 }409 }410 411 /**412 * Gets the contents of a blob as binary data.413 *414 * @param {Blob} blob The blob415 * @param {Boolean} [compress=false] Specifies whether or not to compress416 * the data417 * @param {Object} options Options object418 * @param {Boolean} [options.fin=false] Specifies whether or not to set the419 * FIN bit420 * @param {Function} [options.generateMask] The function used to generate the421 * masking key422 * @param {Boolean} [options.mask=false] Specifies whether or not to mask423 * `data`424 * @param {Buffer} [options.maskBuffer] The buffer used to store the masking425 * key426 * @param {Number} options.opcode The opcode427 * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be428 * modified429 * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the430 * RSV1 bit431 * @param {Function} [cb] Callback432 * @private433 */434 getBlobData(blob, compress, options, cb) {435 this._bufferedBytes += options[kByteLength];436 this._state = GET_BLOB_DATA;437 438 blob439 .arrayBuffer()440 .then((arrayBuffer) => {441 if (this._socket.destroyed) {442 const err = new Error(443 'The socket was closed while the blob was being read'444 );445 446 //447 // `callCallbacks` is called in the next tick to ensure that errors448 // that might be thrown in the callbacks behave like errors thrown449 // outside the promise chain.450 //451 process.nextTick(callCallbacks, this, err, cb);452 return;453 }454 455 this._bufferedBytes -= options[kByteLength];456 const data = toBuffer(arrayBuffer);457 458 if (!compress) {459 this._state = DEFAULT;460 this.sendFrame(Sender.frame(data, options), cb);461 this.dequeue();462 } else {463 this.dispatch(data, compress, options, cb);464 }465 })466 .catch((err) => {467 //468 // `onError` is called in the next tick for the same reason that469 // `callCallbacks` above is.470 //471 process.nextTick(onError, this, err, cb);472 });473 }474 475 /**476 * Dispatches a message.477 *478 * @param {(Buffer|String)} data The message to send479 * @param {Boolean} [compress=false] Specifies whether or not to compress480 * `data`481 * @param {Object} options Options object482 * @param {Boolean} [options.fin=false] Specifies whether or not to set the483 * FIN bit484 * @param {Function} [options.generateMask] The function used to generate the485 * masking key486 * @param {Boolean} [options.mask=false] Specifies whether or not to mask487 * `data`488 * @param {Buffer} [options.maskBuffer] The buffer used to store the masking489 * key490 * @param {Number} options.opcode The opcode491 * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be492 * modified493 * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the494 * RSV1 bit495 * @param {Function} [cb] Callback496 * @private497 */498 dispatch(data, compress, options, cb) {499 if (!compress) {500 this.sendFrame(Sender.frame(data, options), cb);501 return;502 }503 504 const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];505 506 this._bufferedBytes += options[kByteLength];507 this._state = DEFLATING;508 perMessageDeflate.compress(data, options.fin, (_, buf) => {509 if (this._socket.destroyed) {510 const err = new Error(511 'The socket was closed while data was being compressed'512 );513 514 callCallbacks(this, err, cb);515 return;516 }517 518 this._bufferedBytes -= options[kByteLength];519 this._state = DEFAULT;520 options.readOnly = false;521 this.sendFrame(Sender.frame(buf, options), cb);522 this.dequeue();523 });524 }525 526 /**527 * Executes queued send operations.528 *529 * @private530 */531 dequeue() {532 while (this._state === DEFAULT && this._queue.length) {533 const params = this._queue.shift();534 535 this._bufferedBytes -= params[3][kByteLength];536 Reflect.apply(params[0], this, params.slice(1));537 }538 }539 540 /**541 * Enqueues a send operation.542 *543 * @param {Array} params Send operation parameters.544 * @private545 */546 enqueue(params) {547 this._bufferedBytes += params[3][kByteLength];548 this._queue.push(params);549 }550 551 /**552 * Sends a frame.553 *554 * @param {Buffer[]} list The frame to send555 * @param {Function} [cb] Callback556 * @private557 */558 sendFrame(list, cb) {559 if (list.length === 2) {560 this._socket.cork();561 this._socket.write(list[0]);562 this._socket.write(list[1], cb);563 this._socket.uncork();564 } else {565 this._socket.write(list[0], cb);566 }567 }568}569 570module.exports = Sender;571 572/**573 * Calls queued callbacks with an error.574 *575 * @param {Sender} sender The `Sender` instance576 * @param {Error} err The error to call the callbacks with577 * @param {Function} [cb] The first callback578 * @private579 */580function callCallbacks(sender, err, cb) {581 if (typeof cb === 'function') cb(err);582 583 for (let i = 0; i < sender._queue.length; i++) {584 const params = sender._queue[i];585 const callback = params[params.length - 1];586 587 if (typeof callback === 'function') callback(err);588 }589}590 591/**592 * Handles a `Sender` error.593 *594 * @param {Sender} sender The `Sender` instance595 * @param {Error} err The error596 * @param {Function} [cb] The first pending callback597 * @private598 */599function onError(sender, err, cb) {600 callCallbacks(sender, err, cb);601 sender.onerror(err);602}603