comunicationLeap/node_modules/ws/lib/Receiver.js
node_modules/ws/lib/Receiver.jsRaw
1/*!
2 * ws: a node.js websocket client
3 * Copyright(c) 2011 Einar Otto Stangvik <einaros@gmail.com>
4 * MIT Licensed
5 */
6
7var util = require('util')
8 , Validation = require('./Validation').Validation
9 , ErrorCodes = require('./ErrorCodes')
10 , BufferPool = require('./BufferPool')
11 , bufferUtil = require('./BufferUtil').BufferUtil;
12
13/**
14 * Node version 0.4 and 0.6 compatibility
15 */
16
17var isNodeV4 = /^v0\.4/.test(process.version);
18
19/**
20 * HyBi Receiver implementation
21 */
22
23function Receiver () {
24 // memory pool for fragmented messages
25 var fragmentedPoolPrevUsed = -1;
26 this.fragmentedBufferPool = new BufferPool(1024, function(db, length) {
27 return db.used + length;
28 }, function(db) {
29 return fragmentedPoolPrevUsed = fragmentedPoolPrevUsed >= 0 ?
30 (fragmentedPoolPrevUsed + db.used) / 2 :
31 db.used;
32 });
33
34 // memory pool for unfragmented messages
35 var unfragmentedPoolPrevUsed = -1;
36 this.unfragmentedBufferPool = new BufferPool(1024, function(db, length) {
37 return db.used + length;
38 }, function(db) {
39 return unfragmentedPoolPrevUsed = unfragmentedPoolPrevUsed >= 0 ?
40 (unfragmentedPoolPrevUsed + db.used) / 2 :
41 db.used;
42 });
43
44 this.state = {
45 activeFragmentedOperation: null,
46 lastFragment: false,
47 masked: false,
48 opcode: 0,
49 fragmentedOperation: false
50 };
51 this.overflow = [];
52 this.headerBuffer = new Buffer(10);
53 this.expectOffset = 0;
54 this.expectBuffer = null;
55 this.expectHandler = null;
56 this.currentMessage = [];
57 this.expectHeader(2, this.processPacket);
58 this.dead = false;
59
60 this.onerror = function() {};
61 this.ontext = function() {};
62 this.onbinary = function() {};
63 this.onclose = function() {};
64 this.onping = function() {};
65 this.onpong = function() {};
66};
67
68module.exports = Receiver;
69
70/**
71 * Add new data to the parser.
72 *
73 * @api public
74 */
75
76Receiver.prototype.add = function(data) {
77 var dataLength = data.length;
78 if (dataLength == 0) return;
79 if (this.expectBuffer == null) {
80 this.overflow.push(data);
81 return;
82 }
83 var toRead = Math.min(dataLength, this.expectBuffer.length - this.expectOffset);
84 fastCopy(toRead, data, this.expectBuffer, this.expectOffset);
85 this.expectOffset += toRead;
86 if (toRead < dataLength) {
87 this.overflow.push(data.slice(toRead));
88 }
89 while (this.expectBuffer && this.expectOffset == this.expectBuffer.length) {
90 var bufferForHandler = this.expectBuffer;
91 this.expectBuffer = null;
92 this.expectOffset = 0;
93 this.expectHandler.call(this, bufferForHandler);
94 }
95}
96
97/**
98 * Releases all resources used by the receiver.
99 *
100 * @api public
101 */
102
103Receiver.prototype.cleanup = function() {
104 this.dead = true;
105 this.overflow = null;
106 this.headerBuffer = null;
107 this.expectBuffer = null;
108 this.expectHandler = null;
109 this.unfragmentedBufferPool = null;
110 this.fragmentedBufferPool = null;
111 this.state = null;
112 this.currentMessage = null;
113 this.onerror = null;
114 this.ontext = null;
115 this.onbinary = null;
116 this.onclose = null;
117 this.onping = null;
118 this.onpong = null;
119}
120
121/**
122 * Waits for a certain amount of header bytes to be available, then fires a callback.
123 *
124 * @api private
125 */
126
127Receiver.prototype.expectHeader = function(length, handler) {
128 if (length == 0) {
129 handler(null);
130 return;
131 }
132 this.expectBuffer = this.headerBuffer.slice(this.expectOffset, this.expectOffset + length);
133 this.expectHandler = handler;
134 var toRead = length;
135 while (toRead > 0 && this.overflow.length > 0) {
136 var fromOverflow = this.overflow.pop();
137 if (toRead < fromOverflow.length) this.overflow.push(fromOverflow.slice(toRead));
138 var read = Math.min(fromOverflow.length, toRead);
139 fastCopy(read, fromOverflow, this.expectBuffer, this.expectOffset);
140 this.expectOffset += read;
141 toRead -= read;
142 }
143}
144
145/**
146 * Waits for a certain amount of data bytes to be available, then fires a callback.
147 *
148 * @api private
149 */
150
151Receiver.prototype.expectData = function(length, handler) {
152 if (length == 0) {
153 handler(null);
154 return;
155 }
156 this.expectBuffer = this.allocateFromPool(length, this.state.fragmentedOperation);
157 this.expectHandler = handler;
158 var toRead = length;
159 while (toRead > 0 && this.overflow.length > 0) {
160 var fromOverflow = this.overflow.pop();
161 if (toRead < fromOverflow.length) this.overflow.push(fromOverflow.slice(toRead));
162 var read = Math.min(fromOverflow.length, toRead);
163 fastCopy(read, fromOverflow, this.expectBuffer, this.expectOffset);
164 this.expectOffset += read;
165 toRead -= read;
166 }
167}
168
169/**
170 * Allocates memory from the buffer pool.
171 *
172 * @api private
173 */
174
175Receiver.prototype.allocateFromPool = !isNodeV4
176 ? function(length, isFragmented) { return (isFragmented ? this.fragmentedBufferPool : this.unfragmentedBufferPool).get(length); }
177 : function(length) { return new Buffer(length); };
178
179/**
180 * Start processing a new packet.
181 *
182 * @api private
183 */
184
185Receiver.prototype.processPacket = function (data) {
186 if ((data[0] & 0x70) != 0) {
187 this.error('reserved fields must be empty', 1002);
188 return;
189 }
190 this.state.lastFragment = (data[0] & 0x80) == 0x80;
191 this.state.masked = (data[1] & 0x80) == 0x80;
192 var opcode = data[0] & 0xf;
193 if (opcode === 0) {
194 // continuation frame
195 this.state.fragmentedOperation = true;
196 this.state.opcode = this.state.activeFragmentedOperation;
197 if (!(this.state.opcode == 1 || this.state.opcode == 2)) {
198 this.error('continuation frame cannot follow current opcode', 1002);
199 return;
200 }
201 }
202 else {
203 if (opcode < 3 && this.state.activeFragmentedOperation != null) {
204 this.error('data frames after the initial data frame must have opcode 0', 1002);
205 return;
206 }
207 this.state.opcode = opcode;
208 if (this.state.lastFragment === false) {
209 this.state.fragmentedOperation = true;
210 this.state.activeFragmentedOperation = opcode;
211 }
212 else this.state.fragmentedOperation = false;
213 }
214 var handler = opcodes[this.state.opcode];
215 if (typeof handler == 'undefined') this.error('no handler for opcode ' + this.state.opcode, 1002);
216 else {
217 handler.start.call(this, data);
218 }
219}
220
221/**
222 * Endprocessing a packet.
223 *
224 * @api private
225 */
226
227Receiver.prototype.endPacket = function() {
228 if (!this.state.fragmentedOperation) this.unfragmentedBufferPool.reset(true);
229 else if (this.state.lastFragment) this.fragmentedBufferPool.reset(false);
230 this.expectOffset = 0;
231 this.expectBuffer = null;
232 this.expectHandler = null;
233 if (this.state.lastFragment && this.state.opcode === this.state.activeFragmentedOperation) {
234 // end current fragmented operation
235 this.state.activeFragmentedOperation = null;
236 }
237 this.state.lastFragment = false;
238 this.state.opcode = this.state.activeFragmentedOperation != null ? this.state.activeFragmentedOperation : 0;
239 this.state.masked = false;
240 this.expectHeader(2, this.processPacket);
241}
242
243/**
244 * Reset the parser state.
245 *
246 * @api private
247 */
248
249Receiver.prototype.reset = function() {
250 if (this.dead) return;
251 this.state = {
252 activeFragmentedOperation: null,
253 lastFragment: false,
254 masked: false,
255 opcode: 0,
256 fragmentedOperation: false
257 };
258 this.fragmentedBufferPool.reset(true);
259 this.unfragmentedBufferPool.reset(true);
260 this.expectOffset = 0;
261 this.expectBuffer = null;
262 this.expectHandler = null;
263 this.overflow = [];
264 this.currentMessage = [];
265}
266
267/**
268 * Unmask received data.
269 *
270 * @api private
271 */
272
273Receiver.prototype.unmask = function (mask, buf, binary) {
274 if (mask != null && buf != null) bufferUtil.unmask(buf, mask);
275 if (binary) return buf;
276 return buf != null ? buf.toString('utf8') : '';
277}
278
279/**
280 * Concatenates a list of buffers.
281 *
282 * @api private
283 */
284
285Receiver.prototype.concatBuffers = function(buffers) {
286 var length = 0;
287 for (var i = 0, l = buffers.length; i < l; ++i) length += buffers[i].length;
288 var mergedBuffer = new Buffer(length);
289 bufferUtil.merge(mergedBuffer, buffers);
290 return mergedBuffer;
291}
292
293/**
294 * Handles an error
295 *
296 * @api private
297 */
298
299Receiver.prototype.error = function (reason, protocolErrorCode) {
300 this.reset();
301 this.onerror(reason, protocolErrorCode);
302 return this;
303}
304
305/**
306 * Buffer utilities
307 */
308
309function readUInt16BE(start) {
310 return (this[start]<<8) +
311 this[start+1];
312}
313
314function readUInt32BE(start) {
315 return (this[start]<<24) +
316 (this[start+1]<<16) +
317 (this[start+2]<<8) +
318 this[start+3];
319}
320
321function fastCopy(length, srcBuffer, dstBuffer, dstOffset) {
322 switch (length) {
323 default: srcBuffer.copy(dstBuffer, dstOffset, 0, length); break;
324 case 16: dstBuffer[dstOffset+15] = srcBuffer[15];
325 case 15: dstBuffer[dstOffset+14] = srcBuffer[14];
326 case 14: dstBuffer[dstOffset+13] = srcBuffer[13];
327 case 13: dstBuffer[dstOffset+12] = srcBuffer[12];
328 case 12: dstBuffer[dstOffset+11] = srcBuffer[11];
329 case 11: dstBuffer[dstOffset+10] = srcBuffer[10];
330 case 10: dstBuffer[dstOffset+9] = srcBuffer[9];
331 case 9: dstBuffer[dstOffset+8] = srcBuffer[8];
332 case 8: dstBuffer[dstOffset+7] = srcBuffer[7];
333 case 7: dstBuffer[dstOffset+6] = srcBuffer[6];
334 case 6: dstBuffer[dstOffset+5] = srcBuffer[5];
335 case 5: dstBuffer[dstOffset+4] = srcBuffer[4];
336 case 4: dstBuffer[dstOffset+3] = srcBuffer[3];
337 case 3: dstBuffer[dstOffset+2] = srcBuffer[2];
338 case 2: dstBuffer[dstOffset+1] = srcBuffer[1];
339 case 1: dstBuffer[dstOffset] = srcBuffer[0];
340 }
341}
342
343/**
344 * Opcode handlers
345 */
346
347var opcodes = {
348 // text
349 '1': {
350 start: function(data) {
351 var self = this;
352 // decode length
353 var firstLength = data[1] & 0x7f;
354 if (firstLength < 126) {
355 opcodes['1'].getData.call(self, firstLength);
356 }
357 else if (firstLength == 126) {
358 self.expectHeader(2, function(data) {
359 opcodes['1'].getData.call(self, readUInt16BE.call(data, 0));
360 });
361 }
362 else if (firstLength == 127) {
363 self.expectHeader(8, function(data) {
364 if (readUInt32BE.call(data, 0) != 0) {
365 self.error('packets with length spanning more than 32 bit is currently not supported', 1008);
366 return;
367 }
368 opcodes['1'].getData.call(self, readUInt32BE.call(data, 4));
369 });
370 }
371 },
372 getData: function(length) {
373 var self = this;
374 if (self.state.masked) {
375 self.expectHeader(4, function(data) {
376 var mask = data;
377 self.expectData(length, function(data) {
378 opcodes['1'].finish.call(self, mask, data);
379 });
380 });
381 }
382 else {
383 self.expectData(length, function(data) {
384 opcodes['1'].finish.call(self, null, data);
385 });
386 }
387 },
388 finish: function(mask, data) {
389 var packet = this.unmask(mask, data, true);
390 if (packet != null) this.currentMessage.push(packet);
391 if (this.state.lastFragment) {
392 var messageBuffer = this.concatBuffers(this.currentMessage);
393 if (!Validation.isValidUTF8(messageBuffer)) {
394 this.error('invalid utf8 sequence', 1007);
395 return;
396 }
397 this.ontext(messageBuffer.toString('utf8'), {masked: this.state.masked, buffer: messageBuffer});
398 this.currentMessage = [];
399 }
400 this.endPacket();
401 }
402 },
403 // binary
404 '2': {
405 start: function(data) {
406 var self = this;
407 // decode length
408 var firstLength = data[1] & 0x7f;
409 if (firstLength < 126) {
410 opcodes['2'].getData.call(self, firstLength);
411 }
412 else if (firstLength == 126) {
413 self.expectHeader(2, function(data) {
414 opcodes['2'].getData.call(self, readUInt16BE.call(data, 0));
415 });
416 }
417 else if (firstLength == 127) {
418 self.expectHeader(8, function(data) {
419 if (readUInt32BE.call(data, 0) != 0) {
420 self.error('packets with length spanning more than 32 bit is currently not supported', 1008);
421 return;
422 }
423 opcodes['2'].getData.call(self, readUInt32BE.call(data, 4, true));
424 });
425 }
426 },
427 getData: function(length) {
428 var self = this;
429 if (self.state.masked) {
430 self.expectHeader(4, function(data) {
431 var mask = data;
432 self.expectData(length, function(data) {
433 opcodes['2'].finish.call(self, mask, data);
434 });
435 });
436 }
437 else {
438 self.expectData(length, function(data) {
439 opcodes['2'].finish.call(self, null, data);
440 });
441 }
442 },
443 finish: function(mask, data) {
444 var packet = this.unmask(mask, data, true);
445 if (packet != null) this.currentMessage.push(packet);
446 if (this.state.lastFragment) {
447 var messageBuffer = this.concatBuffers(this.currentMessage);
448 this.onbinary(messageBuffer, {masked: this.state.masked, buffer: messageBuffer});
449 this.currentMessage = [];
450 }
451 this.endPacket();
452 }
453 },
454 // close
455 '8': {
456 start: function(data) {
457 var self = this;
458 if (self.state.lastFragment == false) {
459 self.error('fragmented close is not supported', 1002);
460 return;
461 }
462
463 // decode length
464 var firstLength = data[1] & 0x7f;
465 if (firstLength < 126) {
466 opcodes['8'].getData.call(self, firstLength);
467 }
468 else {
469 self.error('control frames cannot have more than 125 bytes of data', 1002);
470 }
471 },
472 getData: function(length) {
473 var self = this;
474 if (self.state.masked) {
475 self.expectHeader(4, function(data) {
476 var mask = data;
477 self.expectData(length, function(data) {
478 opcodes['8'].finish.call(self, mask, data);
479 });
480 });
481 }
482 else {
483 self.expectData(length, function(data) {
484 opcodes['8'].finish.call(self, null, data);
485 });
486 }
487 },
488 finish: function(mask, data) {
489 var self = this;
490 data = self.unmask(mask, data, true);
491 if (data && data.length == 1) {
492 self.error('close packets with data must be at least two bytes long', 1002);
493 return;
494 }
495 var code = data && data.length > 1 ? readUInt16BE.call(data, 0) : 1000;
496 if (!ErrorCodes.isValidErrorCode(code)) {
497 self.error('invalid error code', 1002);
498 return;
499 }
500 var message = '';
501 if (data && data.length > 2) {
502 var messageBuffer = data.slice(2);
503 if (!Validation.isValidUTF8(messageBuffer)) {
504 self.error('invalid utf8 sequence', 1007);
505 return;
506 }
507 message = messageBuffer.toString('utf8');
508 }
509 this.onclose(code, message, {masked: self.state.masked});
510 this.reset();
511 },
512 },
513 // ping
514 '9': {
515 start: function(data) {
516 var self = this;
517 if (self.state.lastFragment == false) {
518 self.error('fragmented ping is not supported', 1002);
519 return;
520 }
521
522 // decode length
523 var firstLength = data[1] & 0x7f;
524 if (firstLength < 126) {
525 opcodes['9'].getData.call(self, firstLength);
526 }
527 else {
528 self.error('control frames cannot have more than 125 bytes of data', 1002);
529 }
530 },
531 getData: function(length) {
532 var self = this;
533 if (self.state.masked) {
534 self.expectHeader(4, function(data) {
535 var mask = data;
536 self.expectData(length, function(data) {
537 opcodes['9'].finish.call(self, mask, data);
538 });
539 });
540 }
541 else {
542 self.expectData(length, function(data) {
543 opcodes['9'].finish.call(self, null, data);
544 });
545 }
546 },
547 finish: function(mask, data) {
548 this.onping(this.unmask(mask, data, true), {masked: this.state.masked, binary: true});
549 this.endPacket();
550 }
551 },
552 // pong
553 '10': {
554 start: function(data) {
555 var self = this;
556 if (self.state.lastFragment == false) {
557 self.error('fragmented pong is not supported', 1002);
558 return;
559 }
560
561 // decode length
562 var firstLength = data[1] & 0x7f;
563 if (firstLength < 126) {
564 opcodes['10'].getData.call(self, firstLength);
565 }
566 else {
567 self.error('control frames cannot have more than 125 bytes of data', 1002);
568 }
569 },
570 getData: function(length) {
571 var self = this;
572 if (this.state.masked) {
573 this.expectHeader(4, function(data) {
574 var mask = data;
575 self.expectData(length, function(data) {
576 opcodes['10'].finish.call(self, mask, data);
577 });
578 });
579 }
580 else {
581 this.expectData(length, function(data) {
582 opcodes['10'].finish.call(self, null, data);
583 });
584 }
585 },
586 finish: function(mask, data) {
587 this.onpong(this.unmask(mask, data, true), {masked: this.state.masked, binary: true});
588 this.endPacket();
589 }
590 }
591}