socketioMEANseed/server/node_modules/ws/lib/Receiver.js
server/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 , PerMessageDeflate = require('./PerMessageDeflate');
13
14/**
15 * HyBi Receiver implementation
16 */
17
18function Receiver (extensions,maxPayload) {
19 if (this instanceof Receiver === false) {
20 throw new TypeError("Classes can't be function-called");
21 }
22 if(typeof extensions==='number'){
23 maxPayload=extensions;
24 extensions={};
25 }
26
27
28 // memory pool for fragmented messages
29 var fragmentedPoolPrevUsed = -1;
30 this.fragmentedBufferPool = new BufferPool(1024, function(db, length) {
31 return db.used + length;
32 }, function(db) {
33 return fragmentedPoolPrevUsed = fragmentedPoolPrevUsed >= 0 ?
34 Math.ceil((fragmentedPoolPrevUsed + db.used) / 2) :
35 db.used;
36 });
37
38 // memory pool for unfragmented messages
39 var unfragmentedPoolPrevUsed = -1;
40 this.unfragmentedBufferPool = new BufferPool(1024, function(db, length) {
41 return db.used + length;
42 }, function(db) {
43 return unfragmentedPoolPrevUsed = unfragmentedPoolPrevUsed >= 0 ?
44 Math.ceil((unfragmentedPoolPrevUsed + db.used) / 2) :
45 db.used;
46 });
47 this.extensions = extensions || {};
48 this.maxPayload = maxPayload || 0;
49 this.currentPayloadLength = 0;
50 this.state = {
51 activeFragmentedOperation: null,
52 lastFragment: false,
53 masked: false,
54 opcode: 0,
55 fragmentedOperation: false
56 };
57 this.overflow = [];
58 this.headerBuffer = new Buffer(10);
59 this.expectOffset = 0;
60 this.expectBuffer = null;
61 this.expectHandler = null;
62 this.currentMessage = [];
63 this.currentMessageLength = 0;
64 this.messageHandlers = [];
65 this.expectHeader(2, this.processPacket);
66 this.dead = false;
67 this.processing = false;
68
69 this.onerror = function() {};
70 this.ontext = function() {};
71 this.onbinary = function() {};
72 this.onclose = function() {};
73 this.onping = function() {};
74 this.onpong = function() {};
75}
76
77module.exports = Receiver;
78
79/**
80 * Add new data to the parser.
81 *
82 * @api public
83 */
84
85Receiver.prototype.add = function(data) {
86 if (this.dead) return;
87 var dataLength = data.length;
88 if (dataLength == 0) return;
89 if (this.expectBuffer == null) {
90 this.overflow.push(data);
91 return;
92 }
93 var toRead = Math.min(dataLength, this.expectBuffer.length - this.expectOffset);
94 fastCopy(toRead, data, this.expectBuffer, this.expectOffset);
95 this.expectOffset += toRead;
96 if (toRead < dataLength) {
97 this.overflow.push(data.slice(toRead));
98 }
99 while (this.expectBuffer && this.expectOffset == this.expectBuffer.length) {
100 var bufferForHandler = this.expectBuffer;
101 this.expectBuffer = null;
102 this.expectOffset = 0;
103 this.expectHandler.call(this, bufferForHandler);
104 }
105};
106
107/**
108 * Releases all resources used by the receiver.
109 *
110 * @api public
111 */
112
113Receiver.prototype.cleanup = function() {
114 this.dead = true;
115 this.overflow = null;
116 this.headerBuffer = null;
117 this.expectBuffer = null;
118 this.expectHandler = null;
119 this.unfragmentedBufferPool = null;
120 this.fragmentedBufferPool = null;
121 this.state = null;
122 this.currentMessage = null;
123 this.onerror = null;
124 this.ontext = null;
125 this.onbinary = null;
126 this.onclose = null;
127 this.onping = null;
128 this.onpong = null;
129};
130
131/**
132 * Waits for a certain amount of header bytes to be available, then fires a callback.
133 *
134 * @api private
135 */
136
137Receiver.prototype.expectHeader = function(length, handler) {
138 if (length == 0) {
139 handler(null);
140 return;
141 }
142 this.expectBuffer = this.headerBuffer.slice(this.expectOffset, this.expectOffset + length);
143 this.expectHandler = handler;
144 var toRead = length;
145 while (toRead > 0 && this.overflow.length > 0) {
146 var fromOverflow = this.overflow.pop();
147 if (toRead < fromOverflow.length) this.overflow.push(fromOverflow.slice(toRead));
148 var read = Math.min(fromOverflow.length, toRead);
149 fastCopy(read, fromOverflow, this.expectBuffer, this.expectOffset);
150 this.expectOffset += read;
151 toRead -= read;
152 }
153};
154
155/**
156 * Waits for a certain amount of data bytes to be available, then fires a callback.
157 *
158 * @api private
159 */
160
161Receiver.prototype.expectData = function(length, handler) {
162 if (length == 0) {
163 handler(null);
164 return;
165 }
166 this.expectBuffer = this.allocateFromPool(length, this.state.fragmentedOperation);
167 this.expectHandler = handler;
168 var toRead = length;
169 while (toRead > 0 && this.overflow.length > 0) {
170 var fromOverflow = this.overflow.pop();
171 if (toRead < fromOverflow.length) this.overflow.push(fromOverflow.slice(toRead));
172 var read = Math.min(fromOverflow.length, toRead);
173 fastCopy(read, fromOverflow, this.expectBuffer, this.expectOffset);
174 this.expectOffset += read;
175 toRead -= read;
176 }
177};
178
179/**
180 * Allocates memory from the buffer pool.
181 *
182 * @api private
183 */
184
185Receiver.prototype.allocateFromPool = function(length, isFragmented) {
186 return (isFragmented ? this.fragmentedBufferPool : this.unfragmentedBufferPool).get(length);
187};
188
189/**
190 * Start processing a new packet.
191 *
192 * @api private
193 */
194
195Receiver.prototype.processPacket = function (data) {
196 if (this.extensions[PerMessageDeflate.extensionName]) {
197 if ((data[0] & 0x30) != 0) {
198 this.error('reserved fields (2, 3) must be empty', 1002);
199 return;
200 }
201 } else {
202 if ((data[0] & 0x70) != 0) {
203 this.error('reserved fields must be empty', 1002);
204 return;
205 }
206 }
207 this.state.lastFragment = (data[0] & 0x80) == 0x80;
208 this.state.masked = (data[1] & 0x80) == 0x80;
209 var compressed = (data[0] & 0x40) == 0x40;
210 var opcode = data[0] & 0xf;
211 if (opcode === 0) {
212 if (compressed) {
213 this.error('continuation frame cannot have the Per-message Compressed bits', 1002);
214 return;
215 }
216 // continuation frame
217 this.state.fragmentedOperation = true;
218 this.state.opcode = this.state.activeFragmentedOperation;
219 if (!(this.state.opcode == 1 || this.state.opcode == 2)) {
220 this.error('continuation frame cannot follow current opcode', 1002);
221 return;
222 }
223 }
224 else {
225 if (opcode < 3 && this.state.activeFragmentedOperation != null) {
226 this.error('data frames after the initial data frame must have opcode 0', 1002);
227 return;
228 }
229 if (opcode >= 8 && compressed) {
230 this.error('control frames cannot have the Per-message Compressed bits', 1002);
231 return;
232 }
233 this.state.compressed = compressed;
234 this.state.opcode = opcode;
235 if (this.state.lastFragment === false) {
236 this.state.fragmentedOperation = true;
237 this.state.activeFragmentedOperation = opcode;
238 }
239 else this.state.fragmentedOperation = false;
240 }
241 var handler = opcodes[this.state.opcode];
242 if (typeof handler == 'undefined') this.error('no handler for opcode ' + this.state.opcode, 1002);
243 else {
244 handler.start.call(this, data);
245 }
246};
247
248/**
249 * Endprocessing a packet.
250 *
251 * @api private
252 */
253
254Receiver.prototype.endPacket = function() {
255 if (this.dead) return;
256 if (!this.state.fragmentedOperation) this.unfragmentedBufferPool.reset(true);
257 else if (this.state.lastFragment) this.fragmentedBufferPool.reset(true);
258 this.expectOffset = 0;
259 this.expectBuffer = null;
260 this.expectHandler = null;
261 if (this.state.lastFragment && this.state.opcode === this.state.activeFragmentedOperation) {
262 // end current fragmented operation
263 this.state.activeFragmentedOperation = null;
264 }
265 this.currentPayloadLength = 0;
266 this.state.lastFragment = false;
267 this.state.opcode = this.state.activeFragmentedOperation != null ? this.state.activeFragmentedOperation : 0;
268 this.state.masked = false;
269 this.expectHeader(2, this.processPacket);
270};
271
272/**
273 * Reset the parser state.
274 *
275 * @api private
276 */
277
278Receiver.prototype.reset = function() {
279 if (this.dead) return;
280 this.state = {
281 activeFragmentedOperation: null,
282 lastFragment: false,
283 masked: false,
284 opcode: 0,
285 fragmentedOperation: false
286 };
287 this.fragmentedBufferPool.reset(true);
288 this.unfragmentedBufferPool.reset(true);
289 this.expectOffset = 0;
290 this.expectBuffer = null;
291 this.expectHandler = null;
292 this.overflow = [];
293 this.currentMessage = [];
294 this.currentMessageLength = 0;
295 this.messageHandlers = [];
296 this.currentPayloadLength = 0;
297};
298
299/**
300 * Unmask received data.
301 *
302 * @api private
303 */
304
305Receiver.prototype.unmask = function (mask, buf, binary) {
306 if (mask != null && buf != null) bufferUtil.unmask(buf, mask);
307 if (binary) return buf;
308 return buf != null ? buf.toString('utf8') : '';
309};
310
311/**
312 * Handles an error
313 *
314 * @api private
315 */
316
317Receiver.prototype.error = function (reason, protocolErrorCode) {
318 if (this.dead) return;
319 this.reset();
320 if(typeof reason == 'string'){
321 this.onerror(new Error(reason), protocolErrorCode);
322 }
323 else if(reason.constructor == Error){
324 this.onerror(reason, protocolErrorCode);
325 }
326 else{
327 this.onerror(new Error("An error occured"),protocolErrorCode);
328 }
329 return this;
330};
331
332/**
333 * Execute message handler buffers
334 *
335 * @api private
336 */
337
338Receiver.prototype.flush = function() {
339 if (this.processing || this.dead) return;
340
341 var handler = this.messageHandlers.shift();
342 if (!handler) return;
343
344 this.processing = true;
345 var self = this;
346
347 handler(function() {
348 self.processing = false;
349 self.flush();
350 });
351};
352
353/**
354 * Apply extensions to message
355 *
356 * @api private
357 */
358
359Receiver.prototype.applyExtensions = function(messageBuffer, fin, compressed, callback) {
360 var self = this;
361 if (compressed) {
362 this.extensions[PerMessageDeflate.extensionName].decompress(messageBuffer, fin, function(err, buffer) {
363 if (self.dead) return;
364 if (err) {
365 callback(new Error('invalid compressed data'));
366 return;
367 }
368 callback(null, buffer);
369 });
370 } else {
371 callback(null, messageBuffer);
372 }
373};
374
375/**
376* Checks payload size, disconnects socket when it exceeds maxPayload
377*
378* @api private
379*/
380Receiver.prototype.maxPayloadExceeded = function(length) {
381 if (this.maxPayload=== undefined || this.maxPayload === null || this.maxPayload < 1) {
382 return false;
383 }
384 var fullLength = this.currentPayloadLength + length;
385 if (fullLength < this.maxPayload) {
386 this.currentPayloadLength = fullLength;
387 return false;
388 }
389 this.error('payload cannot exceed ' + this.maxPayload + ' bytes', 1009);
390 this.messageBuffer=[];
391 this.cleanup();
392
393 return true;
394};
395
396/**
397 * Buffer utilities
398 */
399
400function readUInt16BE(start) {
401 return (this[start]<<8) +
402 this[start+1];
403}
404
405function readUInt32BE(start) {
406 return (this[start]<<24) +
407 (this[start+1]<<16) +
408 (this[start+2]<<8) +
409 this[start+3];
410}
411
412function fastCopy(length, srcBuffer, dstBuffer, dstOffset) {
413 switch (length) {
414 default: srcBuffer.copy(dstBuffer, dstOffset, 0, length); break;
415 case 16: dstBuffer[dstOffset+15] = srcBuffer[15];
416 case 15: dstBuffer[dstOffset+14] = srcBuffer[14];
417 case 14: dstBuffer[dstOffset+13] = srcBuffer[13];
418 case 13: dstBuffer[dstOffset+12] = srcBuffer[12];
419 case 12: dstBuffer[dstOffset+11] = srcBuffer[11];
420 case 11: dstBuffer[dstOffset+10] = srcBuffer[10];
421 case 10: dstBuffer[dstOffset+9] = srcBuffer[9];
422 case 9: dstBuffer[dstOffset+8] = srcBuffer[8];
423 case 8: dstBuffer[dstOffset+7] = srcBuffer[7];
424 case 7: dstBuffer[dstOffset+6] = srcBuffer[6];
425 case 6: dstBuffer[dstOffset+5] = srcBuffer[5];
426 case 5: dstBuffer[dstOffset+4] = srcBuffer[4];
427 case 4: dstBuffer[dstOffset+3] = srcBuffer[3];
428 case 3: dstBuffer[dstOffset+2] = srcBuffer[2];
429 case 2: dstBuffer[dstOffset+1] = srcBuffer[1];
430 case 1: dstBuffer[dstOffset] = srcBuffer[0];
431 }
432}
433
434function clone(obj) {
435 var cloned = {};
436 for (var k in obj) {
437 if (obj.hasOwnProperty(k)) {
438 cloned[k] = obj[k];
439 }
440 }
441 return cloned;
442}
443
444/**
445 * Opcode handlers
446 */
447
448var opcodes = {
449 // text
450 '1': {
451 start: function(data) {
452 var self = this;
453 // decode length
454 var firstLength = data[1] & 0x7f;
455 if (firstLength < 126) {
456 if (self.maxPayloadExceeded(firstLength)){
457 self.error('Maximumpayload exceeded in compressed text message. Aborting...', 1009);
458 return;
459 }
460 opcodes['1'].getData.call(self, firstLength);
461 }
462 else if (firstLength == 126) {
463 self.expectHeader(2, function(data) {
464 var length = readUInt16BE.call(data, 0);
465 if (self.maxPayloadExceeded(length)){
466 self.error('Maximumpayload exceeded in compressed text message. Aborting...', 1009);
467 return;
468 }
469 opcodes['1'].getData.call(self, length);
470 });
471 }
472 else if (firstLength == 127) {
473 self.expectHeader(8, function(data) {
474 if (readUInt32BE.call(data, 0) != 0) {
475 self.error('packets with length spanning more than 32 bit is currently not supported', 1008);
476 return;
477 }
478 var length = readUInt32BE.call(data, 4);
479 if (self.maxPayloadExceeded(length)){
480 self.error('Maximumpayload exceeded in compressed text message. Aborting...', 1009);
481 return;
482 }
483 opcodes['1'].getData.call(self, readUInt32BE.call(data, 4));
484 });
485 }
486 },
487 getData: function(length) {
488 var self = this;
489 if (self.state.masked) {
490 self.expectHeader(4, function(data) {
491 var mask = data;
492 self.expectData(length, function(data) {
493 opcodes['1'].finish.call(self, mask, data);
494 });
495 });
496 }
497 else {
498 self.expectData(length, function(data) {
499 opcodes['1'].finish.call(self, null, data);
500 });
501 }
502 },
503 finish: function(mask, data) {
504 var self = this;
505 var packet = this.unmask(mask, data, true) || new Buffer(0);
506 var state = clone(this.state);
507 this.messageHandlers.push(function(callback) {
508 self.applyExtensions(packet, state.lastFragment, state.compressed, function(err, buffer) {
509 if (err) {
510 if(err.type===1009){
511 return self.error('Maximumpayload exceeded in compressed text message. Aborting...', 1009);
512 }
513 return self.error(err.message, 1007);
514 }
515 if (buffer != null) {
516 if( self.maxPayload==0 || (self.maxPayload > 0 && (self.currentMessageLength + buffer.length) < self.maxPayload) ){
517 self.currentMessage.push(buffer);
518 }
519 else{
520 self.currentMessage=null;
521 self.currentMessage = [];
522 self.currentMessageLength = 0;
523 self.error(new Error('Maximum payload exceeded. maxPayload: '+self.maxPayload), 1009);
524 return;
525 }
526 self.currentMessageLength += buffer.length;
527 }
528 if (state.lastFragment) {
529 var messageBuffer = Buffer.concat(self.currentMessage);
530 self.currentMessage = [];
531 self.currentMessageLength = 0;
532 if (!Validation.isValidUTF8(messageBuffer)) {
533 self.error('invalid utf8 sequence', 1007);
534 return;
535 }
536 self.ontext(messageBuffer.toString('utf8'), {masked: state.masked, buffer: messageBuffer});
537 }
538 callback();
539 });
540 });
541 this.flush();
542 this.endPacket();
543 }
544 },
545 // binary
546 '2': {
547 start: function(data) {
548 var self = this;
549 // decode length
550 var firstLength = data[1] & 0x7f;
551 if (firstLength < 126) {
552 if (self.maxPayloadExceeded(firstLength)){
553 self.error('Max payload exceeded in compressed text message. Aborting...', 1009);
554 return;
555 }
556 opcodes['2'].getData.call(self, firstLength);
557 }
558 else if (firstLength == 126) {
559 self.expectHeader(2, function(data) {
560 var length = readUInt16BE.call(data, 0);
561 if (self.maxPayloadExceeded(length)){
562 self.error('Max payload exceeded in compressed text message. Aborting...', 1009);
563 return;
564 }
565 opcodes['2'].getData.call(self, length);
566 });
567 }
568 else if (firstLength == 127) {
569 self.expectHeader(8, function(data) {
570 if (readUInt32BE.call(data, 0) != 0) {
571 self.error('packets with length spanning more than 32 bit is currently not supported', 1008);
572 return;
573 }
574 var length = readUInt32BE.call(data, 4, true);
575 if (self.maxPayloadExceeded(length)){
576 self.error('Max payload exceeded in compressed text message. Aborting...', 1009);
577 return;
578 }
579 opcodes['2'].getData.call(self, length);
580 });
581 }
582 },
583 getData: function(length) {
584 var self = this;
585 if (self.state.masked) {
586 self.expectHeader(4, function(data) {
587 var mask = data;
588 self.expectData(length, function(data) {
589 opcodes['2'].finish.call(self, mask, data);
590 });
591 });
592 }
593 else {
594 self.expectData(length, function(data) {
595 opcodes['2'].finish.call(self, null, data);
596 });
597 }
598 },
599 finish: function(mask, data) {
600 var self = this;
601 var packet = this.unmask(mask, data, true) || new Buffer(0);
602 var state = clone(this.state);
603 this.messageHandlers.push(function(callback) {
604 self.applyExtensions(packet, state.lastFragment, state.compressed, function(err, buffer) {
605 if (err) {
606 if(err.type===1009){
607 return self.error('Max payload exceeded in compressed binary message. Aborting...', 1009);
608 }
609 return self.error(err.message, 1007);
610 }
611 if (buffer != null) {
612 if( self.maxPayload==0 || (self.maxPayload > 0 && (self.currentMessageLength + buffer.length) < self.maxPayload) ){
613 self.currentMessage.push(buffer);
614 }
615 else{
616 self.currentMessage=null;
617 self.currentMessage = [];
618 self.currentMessageLength = 0;
619 self.error(new Error('Maximum payload exceeded'), 1009);
620 return;
621 }
622 self.currentMessageLength += buffer.length;
623 }
624 if (state.lastFragment) {
625 var messageBuffer = Buffer.concat(self.currentMessage);
626 self.currentMessage = [];
627 self.currentMessageLength = 0;
628 self.onbinary(messageBuffer, {masked: state.masked, buffer: messageBuffer});
629 }
630 callback();
631 });
632 });
633 this.flush();
634 this.endPacket();
635 }
636 },
637 // close
638 '8': {
639 start: function(data) {
640 var self = this;
641 if (self.state.lastFragment == false) {
642 self.error('fragmented close is not supported', 1002);
643 return;
644 }
645
646 // decode length
647 var firstLength = data[1] & 0x7f;
648 if (firstLength < 126) {
649 opcodes['8'].getData.call(self, firstLength);
650 }
651 else {
652 self.error('control frames cannot have more than 125 bytes of data', 1002);
653 }
654 },
655 getData: function(length) {
656 var self = this;
657 if (self.state.masked) {
658 self.expectHeader(4, function(data) {
659 var mask = data;
660 self.expectData(length, function(data) {
661 opcodes['8'].finish.call(self, mask, data);
662 });
663 });
664 }
665 else {
666 self.expectData(length, function(data) {
667 opcodes['8'].finish.call(self, null, data);
668 });
669 }
670 },
671 finish: function(mask, data) {
672 var self = this;
673 data = self.unmask(mask, data, true);
674
675 var state = clone(this.state);
676 this.messageHandlers.push(function() {
677 if (data && data.length == 1) {
678 self.error('close packets with data must be at least two bytes long', 1002);
679 return;
680 }
681 var code = data && data.length > 1 ? readUInt16BE.call(data, 0) : 1000;
682 if (!ErrorCodes.isValidErrorCode(code)) {
683 self.error('invalid error code', 1002);
684 return;
685 }
686 var message = '';
687 if (data && data.length > 2) {
688 var messageBuffer = data.slice(2);
689 if (!Validation.isValidUTF8(messageBuffer)) {
690 self.error('invalid utf8 sequence', 1007);
691 return;
692 }
693 message = messageBuffer.toString('utf8');
694 }
695 self.onclose(code, message, {masked: state.masked});
696 self.reset();
697 });
698 this.flush();
699 },
700 },
701 // ping
702 '9': {
703 start: function(data) {
704 var self = this;
705 if (self.state.lastFragment == false) {
706 self.error('fragmented ping is not supported', 1002);
707 return;
708 }
709
710 // decode length
711 var firstLength = data[1] & 0x7f;
712 if (firstLength < 126) {
713 opcodes['9'].getData.call(self, firstLength);
714 }
715 else {
716 self.error('control frames cannot have more than 125 bytes of data', 1002);
717 }
718 },
719 getData: function(length) {
720 var self = this;
721 if (self.state.masked) {
722 self.expectHeader(4, function(data) {
723 var mask = data;
724 self.expectData(length, function(data) {
725 opcodes['9'].finish.call(self, mask, data);
726 });
727 });
728 }
729 else {
730 self.expectData(length, function(data) {
731 opcodes['9'].finish.call(self, null, data);
732 });
733 }
734 },
735 finish: function(mask, data) {
736 var self = this;
737 data = this.unmask(mask, data, true);
738 var state = clone(this.state);
739 this.messageHandlers.push(function(callback) {
740 self.onping(data, {masked: state.masked, binary: true});
741 callback();
742 });
743 this.flush();
744 this.endPacket();
745 }
746 },
747 // pong
748 '10': {
749 start: function(data) {
750 var self = this;
751 if (self.state.lastFragment == false) {
752 self.error('fragmented pong is not supported', 1002);
753 return;
754 }
755
756 // decode length
757 var firstLength = data[1] & 0x7f;
758 if (firstLength < 126) {
759 opcodes['10'].getData.call(self, firstLength);
760 }
761 else {
762 self.error('control frames cannot have more than 125 bytes of data', 1002);
763 }
764 },
765 getData: function(length) {
766 var self = this;
767 if (this.state.masked) {
768 this.expectHeader(4, function(data) {
769 var mask = data;
770 self.expectData(length, function(data) {
771 opcodes['10'].finish.call(self, mask, data);
772 });
773 });
774 }
775 else {
776 this.expectData(length, function(data) {
777 opcodes['10'].finish.call(self, null, data);
778 });
779 }
780 },
781 finish: function(mask, data) {
782 var self = this;
783 data = self.unmask(mask, data, true);
784 var state = clone(this.state);
785 this.messageHandlers.push(function(callback) {
786 self.onpong(data, {masked: state.masked, binary: true});
787 callback();
788 });
789 this.flush();
790 this.endPacket();
791 }
792 }
793}