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 Websock.prototype = {
  50     // Getters and Setters
  51     get_sQ: function () {
  52         return this._sQ;
  53     },
  54
  55     get_rQ: function () {
  56         return this._rQ;
  57     },
  58
  59     get_rQi: function () {
  60         return this._rQi;
  61     },
  62
  63     set_rQi: function (val) {
  64         this._rQi = val;
  65     },
  66
  67     // Receive Queue
  68     rQlen: function () {
  69         return this._rQlen - this._rQi;
  70     },
  71
  72     rQpeek8: function () {
  73         return this._rQ[this._rQi];
  74     },
  75
  76     rQshift8: function () {
  77         return this._rQ[this._rQi++];
  78     },
  79
  80     rQskip8: function () {
  81         this._rQi++;
  82     },
  83
  84     rQskipBytes: function (num) {
  85         this._rQi += num;
  86     },
  87
  88     // TODO(directxman12): test performance with these vs a DataView
  89     rQshift16: function () {
  90         return (this._rQ[this._rQi++] << 8) +
  91                this._rQ[this._rQi++];
  92     },
  93
  94     rQshift32: function () {
  95         return (this._rQ[this._rQi++] << 24) +
  96                (this._rQ[this._rQi++] << 16) +
  97                (this._rQ[this._rQi++] << 8) +
  98                this._rQ[this._rQi++];
  99     },
 100
 101     rQshiftStr: function (len) {
 102         if (typeof(len) === 'undefined') { len = this.rQlen(); }
 103         const arr = this.rQshiftBytes(len);
 104         return String.fromCharCode.apply(null, arr);
 105     },
 106
 107     rQshiftBytes: function (len) {
 108         if (typeof(len) === 'undefined') { len = this.rQlen(); }
 109         this._rQi += len;
 110         return new Uint8Array(this._rQ.buffer, this._rQi - len, len);
 111     },
 112
 113     rQshiftTo: function (target, len) {
 114         if (len === undefined) { len = this.rQlen(); }
 115         // TODO: make this just use set with views when using a ArrayBuffer to store the rQ
 116         target.set(new Uint8Array(this._rQ.buffer, this._rQi, len));
 117         this._rQi += len;
 118     },
 119
 120     rQwhole: function () {
 121         return new Uint8Array(this._rQ.buffer, 0, this._rQlen);
 122     },
 123
 124     rQslice: function (start, end) {
 125         if (end) {
 126             return new Uint8Array(this._rQ.buffer, this._rQi + start, end - start);
 127         } else {
 128             return new Uint8Array(this._rQ.buffer, this._rQi + start, this._rQlen - this._rQi - start);
 129         }
 130     },
 131
 132     // Check to see if we must wait for 'num' bytes (default to FBU.bytes)
 133     // to be available in the receive queue. Return true if we need to
 134     // wait (and possibly print a debug message), otherwise false.
 135     rQwait: function (msg, num, goback) {
 136         const rQlen = this._rQlen - this._rQi; // Skip rQlen() function call
 137         if (rQlen < num) {
 138             if (goback) {
 139                 if (this._rQi < goback) {
 140                     throw new Error("rQwait cannot backup " + goback + " bytes");
 141                 }
 142                 this._rQi -= goback;
 143             }
 144             return true; // true means need more data
 145         }
 146         return false;
 147     },
 148
 149     // Send Queue
 150
 151     flush: function () {
 152         if (this._sQlen > 0 && this._websocket.readyState === WebSocket.OPEN) {
 153             this._websocket.send(this._encode_message());
 154             this._sQlen = 0;
 155         }
 156     },
 157
 158     send: function (arr) {
 159         this._sQ.set(arr, this._sQlen);
 160         this._sQlen += arr.length;
 161         this.flush();
 162     },
 163
 164     send_string: function (str) {
 165         this.send(str.split('').map(function (chr) {
 166             return chr.charCodeAt(0);
 167         }));
 168     },
 169
 170     // Event Handlers
 171     off: function (evt) {
 172         this._eventHandlers[evt] = function () {};
 173     },
 174
 175     on: function (evt, handler) {
 176         this._eventHandlers[evt] = handler;
 177     },
 178
 179     _allocate_buffers: function () {
 180         this._rQ = new Uint8Array(this._rQbufferSize);
 181         this._sQ = new Uint8Array(this._sQbufferSize);
 182     },
 183
 184     init: function () {
 185         this._allocate_buffers();
 186         this._rQi = 0;
 187         this._websocket = null;
 188     },
 189
 190     open: function (uri, protocols) {
 191         this.init();
 192
 193         this._websocket = new WebSocket(uri, protocols);
 194         this._websocket.binaryType = 'arraybuffer';
 195
 196         this._websocket.onmessage = this._recv_message.bind(this);
 197         this._websocket.onopen = (function () {
 198             Log.Debug('>> WebSock.onopen');
 199             if (this._websocket.protocol) {
 200                 Log.Info("Server choose sub-protocol: " + this._websocket.protocol);
 201             }
 202
 203             this._eventHandlers.open();
 204             Log.Debug("<< WebSock.onopen");
 205         }).bind(this);
 206         this._websocket.onclose = (function (e) {
 207             Log.Debug(">> WebSock.onclose");
 208             this._eventHandlers.close(e);
 209             Log.Debug("<< WebSock.onclose");
 210         }).bind(this);
 211         this._websocket.onerror = (function (e) {
 212             Log.Debug(">> WebSock.onerror: " + e);
 213             this._eventHandlers.error(e);
 214             Log.Debug("<< WebSock.onerror: " + e);
 215         }).bind(this);
 216     },
 217
 218     close: function () {
 219         if (this._websocket) {
 220             if ((this._websocket.readyState === WebSocket.OPEN) ||
 221                     (this._websocket.readyState === WebSocket.CONNECTING)) {
 222                 Log.Info("Closing WebSocket connection");
 223                 this._websocket.close();
 224             }
 225
 226             this._websocket.onmessage = function (e) { return; };
 227         }
 228     },
 229
 230     // private methods
 231     _encode_message: function () {
 232         // Put in a binary arraybuffer
 233         // according to the spec, you can send ArrayBufferViews with the send method
 234         return new Uint8Array(this._sQ.buffer, 0, this._sQlen);
 235     },
 236
 237     _expand_compact_rQ: function (min_fit) {
 238         const resizeNeeded = min_fit || this._rQlen - this._rQi > this._rQbufferSize / 2;
 239         if (resizeNeeded) {
 240             if (!min_fit) {
 241                 // just double the size if we need to do compaction
 242                 this._rQbufferSize *= 2;
 243             } else {
 244                 // otherwise, make sure we satisy rQlen - rQi + min_fit < rQbufferSize / 8
 245                 this._rQbufferSize = (this._rQlen - this._rQi + min_fit) * 8;
 246             }
 247         }
 248
 249         // we don't want to grow unboundedly
 250         if (this._rQbufferSize > MAX_RQ_GROW_SIZE) {
 251             this._rQbufferSize = MAX_RQ_GROW_SIZE;
 252             if (this._rQbufferSize - this._rQlen - this._rQi < min_fit) {
 253                 throw new Error("Receive Queue buffer exceeded " + MAX_RQ_GROW_SIZE + " bytes, and the new message could not fit");
 254             }
 255         }
 256
 257         if (resizeNeeded) {
 258             const old_rQbuffer = this._rQ.buffer;
 259             this._rQmax = this._rQbufferSize / 8;
 260             this._rQ = new Uint8Array(this._rQbufferSize);
 261             this._rQ.set(new Uint8Array(old_rQbuffer, this._rQi));
 262         } else {
 263             if (ENABLE_COPYWITHIN) {
 264                 this._rQ.copyWithin(0, this._rQi);
 265             } else {
 266                 this._rQ.set(new Uint8Array(this._rQ.buffer, this._rQi));
 267             }
 268         }
 269
 270         this._rQlen = this._rQlen - this._rQi;
 271         this._rQi = 0;
 272     },
 273
 274     _decode_message: function (data) {
 275         // push arraybuffer values onto the end
 276         const u8 = new Uint8Array(data);
 277         if (u8.length > this._rQbufferSize - this._rQlen) {
 278             this._expand_compact_rQ(u8.length);
 279         }
 280         this._rQ.set(u8, this._rQlen);
 281         this._rQlen += u8.length;
 282     },
 283
 284     _recv_message: function (e) {
 285         this._decode_message(e.data);
 286         if (this.rQlen() > 0) {
 287             this._eventHandlers.message();
 288             // Compact the receive queue
 289             if (this._rQlen == this._rQi) {
 290                 this._rQlen = 0;
 291                 this._rQi = 0;
 292             } else if (this._rQlen > this._rQmax) {
 293                 this._expand_compact_rQ();
 294             }
 295         } else {
 296             Log.Debug("Ignoring empty message");
 297         }
 298     }
 299 };