diff --git a/lib/websocket/driver/hybi/stream_reader.js b/lib/websocket/driver/hybi/stream_reader.js index 525db0e..6c0867e 100644 --- a/lib/websocket/driver/hybi/stream_reader.js +++ b/lib/websocket/driver/hybi/stream_reader.js @@ -3,7 +3,6 @@ var StreamReader = function() { this._queue = []; this._queueSize = 0; - this._cursor = 0; }; StreamReader.prototype.put = function(buffer) { @@ -15,29 +14,49 @@ StreamReader.prototype.put = function(buffer) { StreamReader.prototype.read = function(length) { if (length > this._queueSize) return null; + if (length === 0) return new Buffer(0); + + var queue = this._queue, + first = queue[0], + buffer; + + if (first.length >= length) { + this._queueSize -= length; + if (first.length === length) { + return queue.shift(); + } else { + buffer = first.slice(0, length); + queue[0] = first.slice(length); + return buffer; + } + } + + var remain = length, buffers; + + for (var i=0, n = queue.length; i < n; i++) { + if (remain < queue[i].length) break; + remain -= queue[i].length; + } + buffers = queue.splice(0, i); + + if (remain > 0 && queue.length > 0) { + buffers.push(queue[0].slice(0, remain)); + queue[0] = queue[0].slice(remain); + } + this._queueSize -= length; + return this._concat(buffers, length); +}; + +StreamReader.prototype._concat = function(buffers, length) { + if (Buffer.concat) return Buffer.concat(buffers, length); var buffer = new Buffer(length), - queue = this._queue, - remain = length, - n = queue.length, - i = 0, - chunk, size; + offset = 0; - while (remain > 0 && i < n) { - chunk = queue[i]; - size = Math.min(remain, chunk.length - this._cursor); - - chunk.copy(buffer, length - remain, this._cursor, this._cursor + size); - - remain -= size; - this._queueSize -= size; - this._cursor = (this._cursor + size) % chunk.length; - - i += 1; + for (var i = 0, n = buffers.length; i < n; i++) { + buffers[i].copy(buffer, offset); + offset += buffers[i].length; } - - queue.splice(0, this._cursor === 0 ? i : i - 1); - return buffer; };