| 1 | /* |
| 2 | * Websock: high-performance buffering wrapper |
| 3 | * Copyright (C) 2019 The noVNC Authors |
| 4 | * Licensed under MPL 2.0 (see LICENSE.txt) |
| 5 | * |
| 6 | * Websock is similar to the standard WebSocket / RTCDataChannel object |
| 7 | * but with extra 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 MAX_RQ_GROW_SIZE = 40 * 1024 * 1024; // 40 MiB |
| 21 | |
| 22 | // Constants pulled from RTCDataChannelState enum |
| 23 | // https://developer.mozilla.org/en-US/docs/Web/API/RTCDataChannel/readyState#RTCDataChannelState_enum |
| 24 | const DataChannel = { |
| 25 | CONNECTING: "connecting", |
| 26 | OPEN: "open", |
| 27 | CLOSING: "closing", |
| 28 | CLOSED: "closed" |
| 29 | }; |
| 30 | |
| 31 | const ReadyStates = { |
| 32 | CONNECTING: [WebSocket.CONNECTING, DataChannel.CONNECTING], |
| 33 | OPEN: [WebSocket.OPEN, DataChannel.OPEN], |
| 34 | CLOSING: [WebSocket.CLOSING, DataChannel.CLOSING], |
| 35 | CLOSED: [WebSocket.CLOSED, DataChannel.CLOSED], |
| 36 | }; |
| 37 | |
| 38 | // Properties a raw channel must have, WebSocket and RTCDataChannel are two examples |
| 39 | const rawChannelProps = [ |
| 40 | "send", |
| 41 | "close", |
| 42 | "binaryType", |
| 43 | "onerror", |
| 44 | "onmessage", |
| 45 | "onopen", |
| 46 | "protocol", |
| 47 | "readyState", |
| 48 | ]; |
| 49 | |
| 50 | export default class Websock { |
| 51 | constructor() { |
| 52 | this._websocket = null; // WebSocket or RTCDataChannel object |
| 53 | |
| 54 | this._rQi = 0; // Receive queue index |
| 55 | this._rQlen = 0; // Next write position in the receive queue |
| 56 | this._rQbufferSize = 1024 * 1024 * 4; // Receive queue buffer size (4 MiB) |
| 57 | // called in init: this._rQ = new Uint8Array(this._rQbufferSize); |
| 58 | this._rQ = null; // Receive queue |
| 59 | |
| 60 | this._sQbufferSize = 1024 * 10; // 10 KiB |
| 61 | // called in init: this._sQ = new Uint8Array(this._sQbufferSize); |
| 62 | this._sQlen = 0; |
| 63 | this._sQ = null; // Send queue |
| 64 | |
| 65 | this._eventHandlers = { |
| 66 | message: () => {}, |
| 67 | open: () => {}, |
| 68 | close: () => {}, |
| 69 | error: () => {} |
| 70 | }; |
| 71 | } |
| 72 | |
| 73 | // Getters and Setters |
| 74 | |
| 75 | get readyState() { |
| 76 | let subState; |
| 77 | |
| 78 | if (this._websocket === null) { |
| 79 | return "unused"; |
| 80 | } |
| 81 | |
| 82 | subState = this._websocket.readyState; |
| 83 | |
| 84 | if (ReadyStates.CONNECTING.includes(subState)) { |
| 85 | return "connecting"; |
| 86 | } else if (ReadyStates.OPEN.includes(subState)) { |
| 87 | return "open"; |
| 88 | } else if (ReadyStates.CLOSING.includes(subState)) { |
| 89 | return "closing"; |
| 90 | } else if (ReadyStates.CLOSED.includes(subState)) { |
| 91 | return "closed"; |
| 92 | } |
| 93 | |
| 94 | return "unknown"; |
| 95 | } |
| 96 | |
| 97 | // Receive Queue |
| 98 | rQpeek8() { |
| 99 | return this._rQ[this._rQi]; |
| 100 | } |
| 101 | |
| 102 | rQskipBytes(bytes) { |
| 103 | this._rQi += bytes; |
| 104 | } |
| 105 | |
| 106 | rQshift8() { |
| 107 | return this._rQshift(1); |
| 108 | } |
| 109 | |
| 110 | rQshift16() { |
| 111 | return this._rQshift(2); |
| 112 | } |
| 113 | |
| 114 | rQshift32() { |
| 115 | return this._rQshift(4); |
| 116 | } |
| 117 | |
| 118 | // TODO(directxman12): test performance with these vs a DataView |
| 119 | _rQshift(bytes) { |
| 120 | let res = 0; |
| 121 | for (let byte = bytes - 1; byte >= 0; byte--) { |
| 122 | res += this._rQ[this._rQi++] << (byte * 8); |
| 123 | } |
| 124 | return res >>> 0; |
| 125 | } |
| 126 | |
| 127 | rQshiftStr(len) { |
| 128 | let str = ""; |
| 129 | // Handle large arrays in steps to avoid long strings on the stack |
| 130 | for (let i = 0; i < len; i += 4096) { |
| 131 | let part = this.rQshiftBytes(Math.min(4096, len - i), false); |
| 132 | str += String.fromCharCode.apply(null, part); |
| 133 | } |
| 134 | return str; |
| 135 | } |
| 136 | |
| 137 | rQshiftBytes(len, copy=true) { |
| 138 | this._rQi += len; |
| 139 | if (copy) { |
| 140 | return this._rQ.slice(this._rQi - len, this._rQi); |
| 141 | } else { |
| 142 | return this._rQ.subarray(this._rQi - len, this._rQi); |
| 143 | } |
| 144 | } |
| 145 | |
| 146 | rQshiftTo(target, len) { |
| 147 | // TODO: make this just use set with views when using a ArrayBuffer to store the rQ |
| 148 | target.set(new Uint8Array(this._rQ.buffer, this._rQi, len)); |
| 149 | this._rQi += len; |
| 150 | } |
| 151 | |
| 152 | rQpeekBytes(len, copy=true) { |
| 153 | if (copy) { |
| 154 | return this._rQ.slice(this._rQi, this._rQi + len); |
| 155 | } else { |
| 156 | return this._rQ.subarray(this._rQi, this._rQi + len); |
| 157 | } |
| 158 | } |
| 159 | |
| 160 | // Check to see if we must wait for 'num' bytes (default to FBU.bytes) |
| 161 | // to be available in the receive queue. Return true if we need to |
| 162 | // wait (and possibly print a debug message), otherwise false. |
| 163 | rQwait(msg, num, goback) { |
| 164 | if (this._rQlen - this._rQi < num) { |
| 165 | if (goback) { |
| 166 | if (this._rQi < goback) { |
| 167 | throw new Error("rQwait cannot backup " + goback + " bytes"); |
| 168 | } |
| 169 | this._rQi -= goback; |
| 170 | } |
| 171 | return true; // true means need more data |
| 172 | } |
| 173 | return false; |
| 174 | } |
| 175 | |
| 176 | // Send Queue |
| 177 | |
| 178 | sQpush8(num) { |
| 179 | this._sQensureSpace(1); |
| 180 | this._sQ[this._sQlen++] = num; |
| 181 | } |
| 182 | |
| 183 | sQpush16(num) { |
| 184 | this._sQensureSpace(2); |
| 185 | this._sQ[this._sQlen++] = (num >> 8) & 0xff; |
| 186 | this._sQ[this._sQlen++] = (num >> 0) & 0xff; |
| 187 | } |
| 188 | |
| 189 | sQpush32(num) { |
| 190 | this._sQensureSpace(4); |
| 191 | this._sQ[this._sQlen++] = (num >> 24) & 0xff; |
| 192 | this._sQ[this._sQlen++] = (num >> 16) & 0xff; |
| 193 | this._sQ[this._sQlen++] = (num >> 8) & 0xff; |
| 194 | this._sQ[this._sQlen++] = (num >> 0) & 0xff; |
| 195 | } |
| 196 | |
| 197 | sQpushString(str) { |
| 198 | let bytes = str.split('').map(chr => chr.charCodeAt(0)); |
| 199 | this.sQpushBytes(new Uint8Array(bytes)); |
| 200 | } |
| 201 | |
| 202 | sQpushBytes(bytes) { |
| 203 | for (let offset = 0;offset < bytes.length;) { |
| 204 | this._sQensureSpace(1); |
| 205 | |
| 206 | let chunkSize = this._sQbufferSize - this._sQlen; |
| 207 | if (chunkSize > bytes.length - offset) { |
| 208 | chunkSize = bytes.length - offset; |
| 209 | } |
| 210 | |
| 211 | this._sQ.set(bytes.subarray(offset, offset + chunkSize), this._sQlen); |
| 212 | this._sQlen += chunkSize; |
| 213 | offset += chunkSize; |
| 214 | } |
| 215 | } |
| 216 | |
| 217 | flush() { |
| 218 | if (this._sQlen > 0 && this.readyState === 'open') { |
| 219 | this._websocket.send(new Uint8Array(this._sQ.buffer, 0, this._sQlen)); |
| 220 | this._sQlen = 0; |
| 221 | } |
| 222 | } |
| 223 | |
| 224 | _sQensureSpace(bytes) { |
| 225 | if (this._sQbufferSize - this._sQlen < bytes) { |
| 226 | this.flush(); |
| 227 | } |
| 228 | } |
| 229 | |
| 230 | // Event Handlers |
| 231 | off(evt) { |
| 232 | this._eventHandlers[evt] = () => {}; |
| 233 | } |
| 234 | |
| 235 | on(evt, handler) { |
| 236 | this._eventHandlers[evt] = handler; |
| 237 | } |
| 238 | |
| 239 | _allocateBuffers() { |
| 240 | this._rQ = new Uint8Array(this._rQbufferSize); |
| 241 | this._sQ = new Uint8Array(this._sQbufferSize); |
| 242 | } |
| 243 | |
| 244 | init() { |
| 245 | this._allocateBuffers(); |
| 246 | this._rQi = 0; |
| 247 | this._websocket = null; |
| 248 | } |
| 249 | |
| 250 | open(uri, protocols) { |
| 251 | this.attach(new WebSocket(uri, protocols)); |
| 252 | } |
| 253 | |
| 254 | attach(rawChannel) { |
| 255 | this.init(); |
| 256 | |
| 257 | // Must get object and class methods to be compatible with the tests. |
| 258 | const channelProps = [...Object.keys(rawChannel), ...Object.getOwnPropertyNames(Object.getPrototypeOf(rawChannel))]; |
| 259 | for (let i = 0; i < rawChannelProps.length; i++) { |
| 260 | const prop = rawChannelProps[i]; |
| 261 | if (channelProps.indexOf(prop) < 0) { |
| 262 | throw new Error('Raw channel missing property: ' + prop); |
| 263 | } |
| 264 | } |
| 265 | |
| 266 | this._websocket = rawChannel; |
| 267 | this._websocket.binaryType = "arraybuffer"; |
| 268 | this._websocket.onmessage = this._recvMessage.bind(this); |
| 269 | |
| 270 | this._websocket.onopen = () => { |
| 271 | Log.Debug('>> WebSock.onopen'); |
| 272 | if (this._websocket.protocol) { |
| 273 | Log.Info("Server choose sub-protocol: " + this._websocket.protocol); |
| 274 | } |
| 275 | |
| 276 | this._eventHandlers.open(); |
| 277 | Log.Debug("<< WebSock.onopen"); |
| 278 | }; |
| 279 | |
| 280 | this._websocket.onclose = (e) => { |
| 281 | Log.Debug(">> WebSock.onclose"); |
| 282 | this._eventHandlers.close(e); |
| 283 | Log.Debug("<< WebSock.onclose"); |
| 284 | }; |
| 285 | |
| 286 | this._websocket.onerror = (e) => { |
| 287 | Log.Debug(">> WebSock.onerror: " + e); |
| 288 | this._eventHandlers.error(e); |
| 289 | Log.Debug("<< WebSock.onerror: " + e); |
| 290 | }; |
| 291 | } |
| 292 | |
| 293 | close() { |
| 294 | if (this._websocket) { |
| 295 | if (this.readyState === 'connecting' || |
| 296 | this.readyState === 'open') { |
| 297 | Log.Info("Closing WebSocket connection"); |
| 298 | this._websocket.close(); |
| 299 | } |
| 300 | |
| 301 | this._websocket.onmessage = () => {}; |
| 302 | } |
| 303 | } |
| 304 | |
| 305 | // private methods |
| 306 | |
| 307 | // We want to move all the unread data to the start of the queue, |
| 308 | // e.g. compacting. |
| 309 | // The function also expands the receive que if needed, and for |
| 310 | // performance reasons we combine these two actions to avoid |
| 311 | // unnecessary copying. |
| 312 | _expandCompactRQ(minFit) { |
| 313 | // if we're using less than 1/8th of the buffer even with the incoming bytes, compact in place |
| 314 | // instead of resizing |
| 315 | const requiredBufferSize = (this._rQlen - this._rQi + minFit) * 8; |
| 316 | const resizeNeeded = this._rQbufferSize < requiredBufferSize; |
| 317 | |
| 318 | if (resizeNeeded) { |
| 319 | // Make sure we always *at least* double the buffer size, and have at least space for 8x |
| 320 | // the current amount of data |
| 321 | this._rQbufferSize = Math.max(this._rQbufferSize * 2, requiredBufferSize); |
| 322 | } |
| 323 | |
| 324 | // we don't want to grow unboundedly |
| 325 | if (this._rQbufferSize > MAX_RQ_GROW_SIZE) { |
| 326 | this._rQbufferSize = MAX_RQ_GROW_SIZE; |
| 327 | if (this._rQbufferSize - (this._rQlen - this._rQi) < minFit) { |
| 328 | throw new Error("Receive Queue buffer exceeded " + MAX_RQ_GROW_SIZE + " bytes, and the new message could not fit"); |
| 329 | } |
| 330 | } |
| 331 | |
| 332 | if (resizeNeeded) { |
| 333 | const oldRQbuffer = this._rQ.buffer; |
| 334 | this._rQ = new Uint8Array(this._rQbufferSize); |
| 335 | this._rQ.set(new Uint8Array(oldRQbuffer, this._rQi, this._rQlen - this._rQi)); |
| 336 | } else { |
| 337 | this._rQ.copyWithin(0, this._rQi, this._rQlen); |
| 338 | } |
| 339 | |
| 340 | this._rQlen = this._rQlen - this._rQi; |
| 341 | this._rQi = 0; |
| 342 | } |
| 343 | |
| 344 | // push arraybuffer values onto the end of the receive que |
| 345 | _recvMessage(e) { |
| 346 | if (this._rQlen == this._rQi) { |
| 347 | // All data has now been processed, this means we |
| 348 | // can reset the receive queue. |
| 349 | this._rQlen = 0; |
| 350 | this._rQi = 0; |
| 351 | } |
| 352 | const u8 = new Uint8Array(e.data); |
| 353 | if (u8.length > this._rQbufferSize - this._rQlen) { |
| 354 | this._expandCompactRQ(u8.length); |
| 355 | } |
| 356 | this._rQ.set(u8, this._rQlen); |
| 357 | this._rQlen += u8.length; |
| 358 | |
| 359 | if (this._rQlen - this._rQi > 0) { |
| 360 | this._eventHandlers.message(); |
| 361 | } else { |
| 362 | Log.Debug("Ignoring empty message"); |
| 363 | } |
| 364 | } |
| 365 | } |