core/websock.js

   1 /*
   2  * Websock: high-performance binary WebSockets
   3  * Copyright (C) 2018 The noVNC Authors
   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 // this has performance issues in some versions Chromium, and
  18 // doesn't gain a tremendous amount of performance increase in Firefox
  19 // at the moment.  It may be valuable to turn it on in the future.
  20 // Also copyWithin() for TypedArrays is not supported in IE 11 or
  21 // Safari 13 (at the moment we want to support Safari 11).
  22 const ENABLE_COPYWITHIN = false;
  23 const MAX_RQ_GROW_SIZE = 40 * 1024 * 1024;  // 40 MiB
  24
  25 export default class Websock {
  26     constructor() {
  27         this._websocket = null;  // WebSocket object
  28
  29         this._rQi = 0;           // Receive queue index
  30         this._rQlen = 0;         // Next write position in the receive queue
  31         this._rQbufferSize = 1024 * 1024 * 4; // Receive queue buffer size (4 MiB)
  32         // called in init: this._rQ = new Uint8Array(this._rQbufferSize);
  33         this._rQ = null; // Receive queue
  34
  35         this._sQbufferSize = 1024 * 10;  // 10 KiB
  36         // called in init: this._sQ = new Uint8Array(this._sQbufferSize);
  37         this._sQlen = 0;
  38         this._sQ = null;  // Send queue
  39
  40         this._eventHandlers = {
  41             message: () => {},
  42             open: () => {},
  43             close: () => {},
  44             error: () => {}
  45         };
  46     }
  47
  48     // Getters and Setters
  49     get sQ() {
  50         return this._sQ;
  51     }
  52
  53     get rQ() {
  54         return this._rQ;
  55     }
  56
  57     get rQi() {
  58         return this._rQi;
  59     }
  60
  61     set rQi(val) {
  62         this._rQi = val;
  63     }
  64
  65     // Receive Queue
  66     get rQlen() {
  67         return this._rQlen - this._rQi;
  68     }
  69
  70     rQpeek8() {
  71         return this._rQ[this._rQi];
  72     }
  73
  74     rQskipBytes(bytes) {
  75         this._rQi += bytes;
  76     }
  77
  78     rQshift8() {
  79         return this._rQshift(1);
  80     }
  81
  82     rQshift16() {
  83         return this._rQshift(2);
  84     }
  85
  86     rQshift32() {
  87         return this._rQshift(4);
  88     }
  89
  90     // TODO(directxman12): test performance with these vs a DataView
  91     _rQshift(bytes) {
  92         let res = 0;
  93         for (let byte = bytes - 1; byte >= 0; byte--) {
  94             res += this._rQ[this._rQi++] << (byte * 8);
  95         }
  96         return res;
  97     }
  98
  99     rQshiftStr(len) {
 100         if (typeof(len) === 'undefined') { len = this.rQlen; }
 101         let str = "";
 102         // Handle large arrays in steps to avoid long strings on the stack
 103         for (let i = 0; i < len; i += 4096) {
 104             let part = this.rQshiftBytes(Math.min(4096, len - i));
 105             str += String.fromCharCode.apply(null, part);
 106         }
 107         return str;
 108     }
 109
 110     rQshiftBytes(len) {
 111         if (typeof(len) === 'undefined') { len = this.rQlen; }
 112         this._rQi += len;
 113         return new Uint8Array(this._rQ.buffer, this._rQi - len, len);
 114     }
 115
 116     rQshiftTo(target, len) {
 117         if (len === undefined) { len = this.rQlen; }
 118         // TODO: make this just use set with views when using a ArrayBuffer to store the rQ
 119         target.set(new Uint8Array(this._rQ.buffer, this._rQi, len));
 120         this._rQi += len;
 121     }
 122
 123     rQslice(start, end = this.rQlen) {
 124         return new Uint8Array(this._rQ.buffer, this._rQi + start, end - start);
 125     }
 126
 127     // Check to see if we must wait for 'num' bytes (default to FBU.bytes)
 128     // to be available in the receive queue. Return true if we need to
 129     // wait (and possibly print a debug message), otherwise false.
 130     rQwait(msg, num, goback) {
 131         if (this.rQlen < num) {
 132             if (goback) {
 133                 if (this._rQi < goback) {
 134                     throw new Error("rQwait cannot backup " + goback + " bytes");
 135                 }
 136                 this._rQi -= goback;
 137             }
 138             return true; // true means need more data
 139         }
 140         return false;
 141     }
 142
 143     // Send Queue
 144
 145     flush() {
 146         if (this._sQlen > 0 && this._websocket.readyState === WebSocket.OPEN) {
 147             this._websocket.send(this._encode_message());
 148             this._sQlen = 0;
 149         }
 150     }
 151
 152     send(arr) {
 153         this._sQ.set(arr, this._sQlen);
 154         this._sQlen += arr.length;
 155         this.flush();
 156     }
 157
 158     send_string(str) {
 159         this.send(str.split('').map(chr => chr.charCodeAt(0)));
 160     }
 161
 162     // Event Handlers
 163     off(evt) {
 164         this._eventHandlers[evt] = () => {};
 165     }
 166
 167     on(evt, handler) {
 168         this._eventHandlers[evt] = handler;
 169     }
 170
 171     _allocate_buffers() {
 172         this._rQ = new Uint8Array(this._rQbufferSize);
 173         this._sQ = new Uint8Array(this._sQbufferSize);
 174     }
 175
 176     init() {
 177         this._allocate_buffers();
 178         this._rQi = 0;
 179         this._websocket = null;
 180     }
 181
 182     open(uri, protocols) {
 183         this.init();
 184
 185         this._websocket = new WebSocket(uri, protocols);
 186         this._websocket.binaryType = 'arraybuffer';
 187
 188         this._websocket.onmessage = this._recv_message.bind(this);
 189         this._websocket.onopen = () => {
 190             Log.Debug('>> WebSock.onopen');
 191             if (this._websocket.protocol) {
 192                 Log.Info("Server choose sub-protocol: " + this._websocket.protocol);
 193             }
 194
 195             this._eventHandlers.open();
 196             Log.Debug("<< WebSock.onopen");
 197         };
 198         this._websocket.onclose = (e) => {
 199             Log.Debug(">> WebSock.onclose");
 200             this._eventHandlers.close(e);
 201             Log.Debug("<< WebSock.onclose");
 202         };
 203         this._websocket.onerror = (e) => {
 204             Log.Debug(">> WebSock.onerror: " + e);
 205             this._eventHandlers.error(e);
 206             Log.Debug("<< WebSock.onerror: " + e);
 207         };
 208     }
 209
 210     close() {
 211         if (this._websocket) {
 212             if ((this._websocket.readyState === WebSocket.OPEN) ||
 213                     (this._websocket.readyState === WebSocket.CONNECTING)) {
 214                 Log.Info("Closing WebSocket connection");
 215                 this._websocket.close();
 216             }
 217
 218             this._websocket.onmessage = () => {};
 219         }
 220     }
 221
 222     // private methods
 223     _encode_message() {
 224         // Put in a binary arraybuffer
 225         // according to the spec, you can send ArrayBufferViews with the send method
 226         return new Uint8Array(this._sQ.buffer, 0, this._sQlen);
 227     }
 228
 229     // We want to move all the unread data to the start of the queue,
 230     // e.g. compacting.
 231     // The function also expands the receive que if needed, and for
 232     // performance reasons we combine these two actions to avoid
 233     // unneccessary copying.
 234     _expand_compact_rQ(min_fit) {
 235         // if we're using less than 1/8th of the buffer even with the incoming bytes, compact in place
 236         // instead of resizing
 237         const required_buffer_size =  (this._rQlen - this._rQi + min_fit) * 8;
 238         const resizeNeeded = this._rQbufferSize < required_buffer_size;
 239
 240         if (resizeNeeded) {
 241             // Make sure we always *at least* double the buffer size, and have at least space for 8x
 242             // the current amount of data
 243             this._rQbufferSize = Math.max(this._rQbufferSize * 2, required_buffer_size);
 244         }
 245
 246         // we don't want to grow unboundedly
 247         if (this._rQbufferSize > MAX_RQ_GROW_SIZE) {
 248             this._rQbufferSize = MAX_RQ_GROW_SIZE;
 249             if (this._rQbufferSize - this.rQlen < min_fit) {
 250                 throw new Error("Receive Queue buffer exceeded " + MAX_RQ_GROW_SIZE + " bytes, and the new message could not fit");
 251             }
 252         }
 253
 254         if (resizeNeeded) {
 255             const old_rQbuffer = this._rQ.buffer;
 256             this._rQ = new Uint8Array(this._rQbufferSize);
 257             this._rQ.set(new Uint8Array(old_rQbuffer, this._rQi, this._rQlen - this._rQi));
 258         } else {
 259             if (ENABLE_COPYWITHIN) {
 260                 this._rQ.copyWithin(0, this._rQi, this._rQlen);
 261             } else {
 262                 this._rQ.set(new Uint8Array(this._rQ.buffer, this._rQi, this._rQlen - this._rQi));
 263             }
 264         }
 265
 266         this._rQlen = this._rQlen - this._rQi;
 267         this._rQi = 0;
 268     }
 269
 270     // push arraybuffer values onto the end of the receive que
 271     _decode_message(data) {
 272         const u8 = new Uint8Array(data);
 273         if (u8.length > this._rQbufferSize - this._rQlen) {
 274             this._expand_compact_rQ(u8.length);
 275         }
 276         this._rQ.set(u8, this._rQlen);
 277         this._rQlen += u8.length;
 278     }
 279
 280     _recv_message(e) {
 281         this._decode_message(e.data);
 282         if (this.rQlen > 0) {
 283             this._eventHandlers.message();
 284             if (this._rQlen == this._rQi) {
 285                 // All data has now been processed, this means we
 286                 // can reset the receive queue.
 287                 this._rQlen = 0;
 288                 this._rQi = 0;
 289             }
 290         } else {
 291             Log.Debug("Ignoring empty message");
 292         }
 293     }
 294 }