device_home/node_modules/engine.io/node_modules/ws/lib/sender.js

   1 'use strict';
   2
   3 const { randomFillSync } = require('crypto');
   4
   5 const PerMessageDeflate = require('./permessage-deflate');
   6 const { EMPTY_BUFFER } = require('./constants');
   7 const { isValidStatusCode } = require('./validation');
   8 const { mask: applyMask, toBuffer } = require('./buffer-util');
   9
  10 const mask = Buffer.alloc(4);
  11
  12 /**
  13  * HyBi Sender implementation.
  14  */
  15 class Sender {
  16   /**
  17    * Creates a Sender instance.
  18    *
  19    * @param {net.Socket} socket The connection socket
  20    * @param {Object} [extensions] An object containing the negotiated extensions
  21    */
  22   constructor(socket, extensions) {
  23     this._extensions = extensions || {};
  24     this._socket = socket;
  25
  26     this._firstFragment = true;
  27     this._compress = false;
  28
  29     this._bufferedBytes = 0;
  30     this._deflating = false;
  31     this._queue = [];
  32   }
  33
  34   /**
  35    * Frames a piece of data according to the HyBi WebSocket protocol.
  36    *
  37    * @param {Buffer} data The data to frame
  38    * @param {Object} options Options object
  39    * @param {Number} options.opcode The opcode
  40    * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be
  41    *     modified
  42    * @param {Boolean} [options.fin=false] Specifies whether or not to set the
  43    *     FIN bit
  44    * @param {Boolean} [options.mask=false] Specifies whether or not to mask
  45    *     `data`
  46    * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the
  47    *     RSV1 bit
  48    * @return {Buffer[]} The framed data as a list of `Buffer` instances
  49    * @public
  50    */
  51   static frame(data, options) {
  52     const merge = options.mask && options.readOnly;
  53     let offset = options.mask ? 6 : 2;
  54     let payloadLength = data.length;
  55
  56     if (data.length >= 65536) {
  57       offset += 8;
  58       payloadLength = 127;
  59     } else if (data.length > 125) {
  60       offset += 2;
  61       payloadLength = 126;
  62     }
  63
  64     const target = Buffer.allocUnsafe(merge ? data.length + offset : offset);
  65
  66     target[0] = options.fin ? options.opcode | 0x80 : options.opcode;
  67     if (options.rsv1) target[0] |= 0x40;
  68
  69     target[1] = payloadLength;
  70
  71     if (payloadLength === 126) {
  72       target.writeUInt16BE(data.length, 2);
  73     } else if (payloadLength === 127) {
  74       target.writeUInt32BE(0, 2);
  75       target.writeUInt32BE(data.length, 6);
  76     }
  77
  78     if (!options.mask) return [target, data];
  79
  80     randomFillSync(mask, 0, 4);
  81
  82     target[1] |= 0x80;
  83     target[offset - 4] = mask[0];
  84     target[offset - 3] = mask[1];
  85     target[offset - 2] = mask[2];
  86     target[offset - 1] = mask[3];
  87
  88     if (merge) {
  89       applyMask(data, mask, target, offset, data.length);
  90       return [target];
  91     }
  92
  93     applyMask(data, mask, data, 0, data.length);
  94     return [target, data];
  95   }
  96
  97   /**
  98    * Sends a close message to the other peer.
  99    *
 100    * @param {Number} [code] The status code component of the body
 101    * @param {String} [data] The message component of the body
 102    * @param {Boolean} [mask=false] Specifies whether or not to mask the message
 103    * @param {Function} [cb] Callback
 104    * @public
 105    */
 106   close(code, data, mask, cb) {
 107     let buf;
 108
 109     if (code === undefined) {
 110       buf = EMPTY_BUFFER;
 111     } else if (typeof code !== 'number' || !isValidStatusCode(code)) {
 112       throw new TypeError('First argument must be a valid error code number');
 113     } else if (data === undefined || data === '') {
 114       buf = Buffer.allocUnsafe(2);
 115       buf.writeUInt16BE(code, 0);
 116     } else {
 117       const length = Buffer.byteLength(data);
 118
 119       if (length > 123) {
 120         throw new RangeError('The message must not be greater than 123 bytes');
 121       }
 122
 123       buf = Buffer.allocUnsafe(2 + length);
 124       buf.writeUInt16BE(code, 0);
 125       buf.write(data, 2);
 126     }
 127
 128     if (this._deflating) {
 129       this.enqueue([this.doClose, buf, mask, cb]);
 130     } else {
 131       this.doClose(buf, mask, cb);
 132     }
 133   }
 134
 135   /**
 136    * Frames and sends a close message.
 137    *
 138    * @param {Buffer} data The message to send
 139    * @param {Boolean} [mask=false] Specifies whether or not to mask `data`
 140    * @param {Function} [cb] Callback
 141    * @private
 142    */
 143   doClose(data, mask, cb) {
 144     this.sendFrame(
 145       Sender.frame(data, {
 146         fin: true,
 147         rsv1: false,
 148         opcode: 0x08,
 149         mask,
 150         readOnly: false
 151       }),
 152       cb
 153     );
 154   }
 155
 156   /**
 157    * Sends a ping message to the other peer.
 158    *
 159    * @param {*} data The message to send
 160    * @param {Boolean} [mask=false] Specifies whether or not to mask `data`
 161    * @param {Function} [cb] Callback
 162    * @public
 163    */
 164   ping(data, mask, cb) {
 165     const buf = toBuffer(data);
 166
 167     if (buf.length > 125) {
 168       throw new RangeError('The data size must not be greater than 125 bytes');
 169     }
 170
 171     if (this._deflating) {
 172       this.enqueue([this.doPing, buf, mask, toBuffer.readOnly, cb]);
 173     } else {
 174       this.doPing(buf, mask, toBuffer.readOnly, cb);
 175     }
 176   }
 177
 178   /**
 179    * Frames and sends a ping message.
 180    *
 181    * @param {Buffer} data The message to send
 182    * @param {Boolean} [mask=false] Specifies whether or not to mask `data`
 183    * @param {Boolean} [readOnly=false] Specifies whether `data` can be modified
 184    * @param {Function} [cb] Callback
 185    * @private
 186    */
 187   doPing(data, mask, readOnly, cb) {
 188     this.sendFrame(
 189       Sender.frame(data, {
 190         fin: true,
 191         rsv1: false,
 192         opcode: 0x09,
 193         mask,
 194         readOnly
 195       }),
 196       cb
 197     );
 198   }
 199
 200   /**
 201    * Sends a pong message to the other peer.
 202    *
 203    * @param {*} data The message to send
 204    * @param {Boolean} [mask=false] Specifies whether or not to mask `data`
 205    * @param {Function} [cb] Callback
 206    * @public
 207    */
 208   pong(data, mask, cb) {
 209     const buf = toBuffer(data);
 210
 211     if (buf.length > 125) {
 212       throw new RangeError('The data size must not be greater than 125 bytes');
 213     }
 214
 215     if (this._deflating) {
 216       this.enqueue([this.doPong, buf, mask, toBuffer.readOnly, cb]);
 217     } else {
 218       this.doPong(buf, mask, toBuffer.readOnly, cb);
 219     }
 220   }
 221
 222   /**
 223    * Frames and sends a pong message.
 224    *
 225    * @param {Buffer} data The message to send
 226    * @param {Boolean} [mask=false] Specifies whether or not to mask `data`
 227    * @param {Boolean} [readOnly=false] Specifies whether `data` can be modified
 228    * @param {Function} [cb] Callback
 229    * @private
 230    */
 231   doPong(data, mask, readOnly, cb) {
 232     this.sendFrame(
 233       Sender.frame(data, {
 234         fin: true,
 235         rsv1: false,
 236         opcode: 0x0a,
 237         mask,
 238         readOnly
 239       }),
 240       cb
 241     );
 242   }
 243
 244   /**
 245    * Sends a data message to the other peer.
 246    *
 247    * @param {*} data The message to send
 248    * @param {Object} options Options object
 249    * @param {Boolean} [options.compress=false] Specifies whether or not to
 250    *     compress `data`
 251    * @param {Boolean} [options.binary=false] Specifies whether `data` is binary
 252    *     or text
 253    * @param {Boolean} [options.fin=false] Specifies whether the fragment is the
 254    *     last one
 255    * @param {Boolean} [options.mask=false] Specifies whether or not to mask
 256    *     `data`
 257    * @param {Function} [cb] Callback
 258    * @public
 259    */
 260   send(data, options, cb) {
 261     const buf = toBuffer(data);
 262     const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];
 263     let opcode = options.binary ? 2 : 1;
 264     let rsv1 = options.compress;
 265
 266     if (this._firstFragment) {
 267       this._firstFragment = false;
 268       if (rsv1 && perMessageDeflate) {
 269         rsv1 = buf.length >= perMessageDeflate._threshold;
 270       }
 271       this._compress = rsv1;
 272     } else {
 273       rsv1 = false;
 274       opcode = 0;
 275     }
 276
 277     if (options.fin) this._firstFragment = true;
 278
 279     if (perMessageDeflate) {
 280       const opts = {
 281         fin: options.fin,
 282         rsv1,
 283         opcode,
 284         mask: options.mask,
 285         readOnly: toBuffer.readOnly
 286       };
 287
 288       if (this._deflating) {
 289         this.enqueue([this.dispatch, buf, this._compress, opts, cb]);
 290       } else {
 291         this.dispatch(buf, this._compress, opts, cb);
 292       }
 293     } else {
 294       this.sendFrame(
 295         Sender.frame(buf, {
 296           fin: options.fin,
 297           rsv1: false,
 298           opcode,
 299           mask: options.mask,
 300           readOnly: toBuffer.readOnly
 301         }),
 302         cb
 303       );
 304     }
 305   }
 306
 307   /**
 308    * Dispatches a data message.
 309    *
 310    * @param {Buffer} data The message to send
 311    * @param {Boolean} [compress=false] Specifies whether or not to compress
 312    *     `data`
 313    * @param {Object} options Options object
 314    * @param {Number} options.opcode The opcode
 315    * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be
 316    *     modified
 317    * @param {Boolean} [options.fin=false] Specifies whether or not to set the
 318    *     FIN bit
 319    * @param {Boolean} [options.mask=false] Specifies whether or not to mask
 320    *     `data`
 321    * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the
 322    *     RSV1 bit
 323    * @param {Function} [cb] Callback
 324    * @private
 325    */
 326   dispatch(data, compress, options, cb) {
 327     if (!compress) {
 328       this.sendFrame(Sender.frame(data, options), cb);
 329       return;
 330     }
 331
 332     const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];
 333
 334     this._bufferedBytes += data.length;
 335     this._deflating = true;
 336     perMessageDeflate.compress(data, options.fin, (_, buf) => {
 337       if (this._socket.destroyed) {
 338         const err = new Error(
 339           'The socket was closed while data was being compressed'
 340         );
 341
 342         if (typeof cb === 'function') cb(err);
 343
 344         for (let i = 0; i < this._queue.length; i++) {
 345           const callback = this._queue[i][4];
 346
 347           if (typeof callback === 'function') callback(err);
 348         }
 349
 350         return;
 351       }
 352
 353       this._bufferedBytes -= data.length;
 354       this._deflating = false;
 355       options.readOnly = false;
 356       this.sendFrame(Sender.frame(buf, options), cb);
 357       this.dequeue();
 358     });
 359   }
 360
 361   /**
 362    * Executes queued send operations.
 363    *
 364    * @private
 365    */
 366   dequeue() {
 367     while (!this._deflating && this._queue.length) {
 368       const params = this._queue.shift();
 369
 370       this._bufferedBytes -= params[1].length;
 371       Reflect.apply(params[0], this, params.slice(1));
 372     }
 373   }
 374
 375   /**
 376    * Enqueues a send operation.
 377    *
 378    * @param {Array} params Send operation parameters.
 379    * @private
 380    */
 381   enqueue(params) {
 382     this._bufferedBytes += params[1].length;
 383     this._queue.push(params);
 384   }
 385
 386   /**
 387    * Sends a frame.
 388    *
 389    * @param {Buffer[]} list The frame to send
 390    * @param {Function} [cb] Callback
 391    * @private
 392    */
 393   sendFrame(list, cb) {
 394     if (list.length === 2) {
 395       this._socket.cork();
 396       this._socket.write(list[0]);
 397       this._socket.write(list[1], cb);
 398       this._socket.uncork();
 399     } else {
 400       this._socket.write(list[0], cb);
 401     }
 402   }
 403 }
 404
 405 module.exports = Sender;