diff --git a/lib/websocket/driver/hybi/stream_reader.js b/lib/websocket/driver/hybi/stream_reader.js index f3783a1..84b5165 100644 --- a/lib/websocket/driver/hybi/stream_reader.js +++ b/lib/websocket/driver/hybi/stream_reader.js @@ -57,6 +57,9 @@ StreamReader.prototype.fork = function(length) { StreamReader.prototype.read = function(length, callback) { if (!this.writable) return; + if (this._queueSize >= length) + return callback.call(this._context, this._readBytes(length)); + var self = this; this.fork(length).pipe(new Concat(function(buffer) { @@ -67,7 +70,7 @@ StreamReader.prototype.read = function(length, callback) { StreamReader.prototype._flush = function() { var streams = this._streams, stream, size, buffer; - while (streams.length > 0 && this._queueSize >= 0) { + while (streams.length > 0 && this._queueSize > 0) { stream = streams[0]; size = Math.min(stream._remaining, this._queueSize); buffer = this._readBytes(size); @@ -84,12 +87,26 @@ StreamReader.prototype._flush = function() { }; StreamReader.prototype._readBytes = function(length) { - var buffer = new Buffer(length), - queue = this._queue, + var queue = this._queue, remain = length, n = queue.length, + first = queue[0], i = 0, - chunk, size; + buffer, chunk, size; + + if (length === 0) return new Buffer(0); + + if (remain <= first.length - this._cursor) { + buffer = first.slice(this._cursor, this._cursor + remain); + + this._queueSize -= remain; + this._cursor = (this._cursor + remain) % first.length; + if (this._cursor === 0) this._queue.shift(); + + return buffer; + } + + buffer = new Buffer(length); while (remain > 0 && i < n) { chunk = queue[i];