core/websock.js

   1 /*
   2  * Websock: high-performance binary WebSockets
   3  * Copyright (C) 2012 Joel Martin
   4  * Licensed under MPL 2.0 (see LICENSE.txt)
   5  *
   6  * Websock is similar to the standard WebSocket object but with extra
   7  * buffer handling.
   8  *
   9  * Websock has built-in receive queue buffering; the message event
  10  * does not contain actual data but is simply a notification that
  11  * there is new data available. Several rQ* methods are available to
  12  * read binary data off of the receive queue.
  13  */
  14
  15 import * as Log from './util/logging.js';
  16
  17 export default function Websock() {
  18     "use strict";
  19
  20     this._websocket = null;  // WebSocket object
  21
  22     this._rQi = 0;           // Receive queue index
  23     this._rQlen = 0;         // Next write position in the receive queue
  24     this._rQbufferSize = 1024 * 1024 * 4; // Receive queue buffer size (4 MiB)
  25     this._rQmax = this._rQbufferSize / 8;
  26     // called in init: this._rQ = new Uint8Array(this._rQbufferSize);
  27     this._rQ = null; // Receive queue
  28
  29     this._sQbufferSize = 1024 * 10;  // 10 KiB
  30     // called in init: this._sQ = new Uint8Array(this._sQbufferSize);
  31     this._sQlen = 0;
  32     this._sQ = null;  // Send queue
  33
  34     this._eventHandlers = {
  35         'message': function () {},
  36         'open': function () {},
  37         'close': function () {},
  38         'error': function () {}
  39     };
  40 }
  41
  42 // this has performance issues in some versions Chromium, and
  43 // doesn't gain a tremendous amount of performance increase in Firefox
  44 // at the moment.  It may be valuable to turn it on in the future.
  45 const ENABLE_COPYWITHIN = false;
  46
  47 const MAX_RQ_GROW_SIZE = 40 * 1024 * 1024;  // 40 MiB
  48
  49 const typedArrayToString = (function () {
  50     // This is only for PhantomJS, which doesn't like apply-ing
  51     // with Typed Arrays
  52     try {
  53         const arr = new Uint8Array([1, 2, 3]);
  54         String.fromCharCode.apply(null, arr);
  55         return function (a) { return String.fromCharCode.apply(null, a); };
  56     } catch (ex) {
  57         return function (a) {
  58             return String.fromCharCode.apply(
  59                 null, Array.prototype.slice.call(a));
  60         };
  61     }
  62 })();
  63
  64 Websock.prototype = {
  65     // Getters and Setters
  66     get_sQ: function () {
  67         return this._sQ;
  68     },
  69
  70     get_rQ: function () {
  71         return this._rQ;
  72     },
  73
  74     get_rQi: function () {
  75         return this._rQi;
  76     },
  77
  78     set_rQi: function (val) {
  79         this._rQi = val;
  80     },
  81
  82     // Receive Queue
  83     rQlen: function () {
  84         return this._rQlen - this._rQi;
  85     },
  86
  87     rQpeek8: function () {
  88         return this._rQ[this._rQi];
  89     },
  90
  91     rQshift8: function () {
  92         return this._rQ[this._rQi++];
  93     },
  94
  95     rQskip8: function () {
  96         this._rQi++;
  97     },
  98
  99     rQskipBytes: function (num) {
 100         this._rQi += num;
 101     },
 102
 103     // TODO(directxman12): test performance with these vs a DataView
 104     rQshift16: function () {
 105         return (this._rQ[this._rQi++] << 8) +
 106                this._rQ[this._rQi++];
 107     },
 108
 109     rQshift32: function () {
 110         return (this._rQ[this._rQi++] << 24) +
 111                (this._rQ[this._rQi++] << 16) +
 112                (this._rQ[this._rQi++] << 8) +
 113                this._rQ[this._rQi++];
 114     },
 115
 116     rQshiftStr: function (len) {
 117         if (typeof(len) === 'undefined') { len = this.rQlen(); }
 118         const arr = new Uint8Array(this._rQ.buffer, this._rQi, len);
 119         this._rQi += len;
 120         return typedArrayToString(arr);
 121     },
 122
 123     rQshiftBytes: function (len) {
 124         if (typeof(len) === 'undefined') { len = this.rQlen(); }
 125         this._rQi += len;
 126         return new Uint8Array(this._rQ.buffer, this._rQi - len, len);
 127     },
 128
 129     rQshiftTo: function (target, len) {
 130         if (len === undefined) { len = this.rQlen(); }
 131         // TODO: make this just use set with views when using a ArrayBuffer to store the rQ
 132         target.set(new Uint8Array(this._rQ.buffer, this._rQi, len));
 133         this._rQi += len;
 134     },
 135
 136     rQwhole: function () {
 137         return new Uint8Array(this._rQ.buffer, 0, this._rQlen);
 138     },
 139
 140     rQslice: function (start, end) {
 141         if (end) {
 142             return new Uint8Array(this._rQ.buffer, this._rQi + start, end - start);
 143         } else {
 144             return new Uint8Array(this._rQ.buffer, this._rQi + start, this._rQlen - this._rQi - start);
 145         }
 146     },
 147
 148     // Check to see if we must wait for 'num' bytes (default to FBU.bytes)
 149     // to be available in the receive queue. Return true if we need to
 150     // wait (and possibly print a debug message), otherwise false.
 151     rQwait: function (msg, num, goback) {
 152         const rQlen = this._rQlen - this._rQi; // Skip rQlen() function call
 153         if (rQlen < num) {
 154             if (goback) {
 155                 if (this._rQi < goback) {
 156                     throw new Error("rQwait cannot backup " + goback + " bytes");
 157                 }
 158                 this._rQi -= goback;
 159             }
 160             return true; // true means need more data
 161         }
 162         return false;
 163     },
 164
 165     // Send Queue
 166
 167     flush: function () {
 168         if (this._sQlen > 0 && this._websocket.readyState === WebSocket.OPEN) {
 169             this._websocket.send(this._encode_message());
 170             this._sQlen = 0;
 171         }
 172     },
 173
 174     send: function (arr) {
 175         this._sQ.set(arr, this._sQlen);
 176         this._sQlen += arr.length;
 177         this.flush();
 178     },
 179
 180     send_string: function (str) {
 181         this.send(str.split('').map(function (chr) {
 182             return chr.charCodeAt(0);
 183         }));
 184     },
 185
 186     // Event Handlers
 187     off: function (evt) {
 188         this._eventHandlers[evt] = function () {};
 189     },
 190
 191     on: function (evt, handler) {
 192         this._eventHandlers[evt] = handler;
 193     },
 194
 195     _allocate_buffers: function () {
 196         this._rQ = new Uint8Array(this._rQbufferSize);
 197         this._sQ = new Uint8Array(this._sQbufferSize);
 198     },
 199
 200     init: function () {
 201         this._allocate_buffers();
 202         this._rQi = 0;
 203         this._websocket = null;
 204     },
 205
 206     open: function (uri, protocols) {
 207         this.init();
 208
 209         this._websocket = new WebSocket(uri, protocols);
 210         this._websocket.binaryType = 'arraybuffer';
 211
 212         this._websocket.onmessage = this._recv_message.bind(this);
 213         this._websocket.onopen = (function () {
 214             Log.Debug('>> WebSock.onopen');
 215             if (this._websocket.protocol) {
 216                 Log.Info("Server choose sub-protocol: " + this._websocket.protocol);
 217             }
 218
 219             this._eventHandlers.open();
 220             Log.Debug("<< WebSock.onopen");
 221         }).bind(this);
 222         this._websocket.onclose = (function (e) {
 223             Log.Debug(">> WebSock.onclose");
 224             this._eventHandlers.close(e);
 225             Log.Debug("<< WebSock.onclose");
 226         }).bind(this);
 227         this._websocket.onerror = (function (e) {
 228             Log.Debug(">> WebSock.onerror: " + e);
 229             this._eventHandlers.error(e);
 230             Log.Debug("<< WebSock.onerror: " + e);
 231         }).bind(this);
 232     },
 233
 234     close: function () {
 235         if (this._websocket) {
 236             if ((this._websocket.readyState === WebSocket.OPEN) ||
 237                     (this._websocket.readyState === WebSocket.CONNECTING)) {
 238                 Log.Info("Closing WebSocket connection");
 239                 this._websocket.close();
 240             }
 241
 242             this._websocket.onmessage = function (e) { return; };
 243         }
 244     },
 245
 246     // private methods
 247     _encode_message: function () {
 248         // Put in a binary arraybuffer
 249         // according to the spec, you can send ArrayBufferViews with the send method
 250         return new Uint8Array(this._sQ.buffer, 0, this._sQlen);
 251     },
 252
 253     _expand_compact_rQ: function (min_fit) {
 254         const resizeNeeded = min_fit || this._rQlen - this._rQi > this._rQbufferSize / 2;
 255         if (resizeNeeded) {
 256             if (!min_fit) {
 257                 // just double the size if we need to do compaction
 258                 this._rQbufferSize *= 2;
 259             } else {
 260                 // otherwise, make sure we satisy rQlen - rQi + min_fit < rQbufferSize / 8
 261                 this._rQbufferSize = (this._rQlen - this._rQi + min_fit) * 8;
 262             }
 263         }
 264
 265         // we don't want to grow unboundedly
 266         if (this._rQbufferSize > MAX_RQ_GROW_SIZE) {
 267             this._rQbufferSize = MAX_RQ_GROW_SIZE;
 268             if (this._rQbufferSize - this._rQlen - this._rQi < min_fit) {
 269                 throw new Error("Receive Queue buffer exceeded " + MAX_RQ_GROW_SIZE + " bytes, and the new message could not fit");
 270             }
 271         }
 272
 273         if (resizeNeeded) {
 274             const old_rQbuffer = this._rQ.buffer;
 275             this._rQmax = this._rQbufferSize / 8;
 276             this._rQ = new Uint8Array(this._rQbufferSize);
 277             this._rQ.set(new Uint8Array(old_rQbuffer, this._rQi));
 278         } else {
 279             if (ENABLE_COPYWITHIN) {
 280                 this._rQ.copyWithin(0, this._rQi);
 281             } else {
 282                 this._rQ.set(new Uint8Array(this._rQ.buffer, this._rQi));
 283             }
 284         }
 285
 286         this._rQlen = this._rQlen - this._rQi;
 287         this._rQi = 0;
 288     },
 289
 290     _decode_message: function (data) {
 291         // push arraybuffer values onto the end
 292         const u8 = new Uint8Array(data);
 293         if (u8.length > this._rQbufferSize - this._rQlen) {
 294             this._expand_compact_rQ(u8.length);
 295         }
 296         this._rQ.set(u8, this._rQlen);
 297         this._rQlen += u8.length;
 298     },
 299
 300     _recv_message: function (e) {
 301         this._decode_message(e.data);
 302         if (this.rQlen() > 0) {
 303             this._eventHandlers.message();
 304             // Compact the receive queue
 305             if (this._rQlen == this._rQi) {
 306                 this._rQlen = 0;
 307                 this._rQi = 0;
 308             } else if (this._rQlen > this._rQmax) {
 309                 this._expand_compact_rQ();
 310             }
 311         } else {
 312             Log.Debug("Ignoring empty message");
 313         }
 314     }
 315 };