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