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