websock.js 11 KB

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