139 lines
3.3 KiB
JavaScript
139 lines
3.3 KiB
JavaScript
var Stream = require('stream').Stream,
|
|
Concat = require('./concat'),
|
|
util = require('util');
|
|
|
|
var defer = (typeof setImmediate === 'function')
|
|
? setImmediate
|
|
: process.nextTick;
|
|
|
|
var StreamReader = function(options, parent) {
|
|
this.readable = !!parent;
|
|
this.writable = !parent;
|
|
this._streams = [];
|
|
this._context = options.context;
|
|
this._parent = parent;
|
|
this._queue = [];
|
|
this._queueSize = 0;
|
|
this._cursor = 0;
|
|
};
|
|
util.inherits(StreamReader, Stream);
|
|
|
|
StreamReader.prototype.write = function(buffer) {
|
|
if (!this.writable) return false;
|
|
if (!buffer || buffer.length === 0) return !this._paused;
|
|
|
|
if (!buffer.copy) buffer = new Buffer(buffer);
|
|
|
|
this._queue.push(buffer);
|
|
this._queueSize += buffer.length;
|
|
this._flush();
|
|
|
|
return !this._paused;
|
|
};
|
|
|
|
StreamReader.prototype.end = function(buffer) {
|
|
if (buffer) this.write(buffer);
|
|
this.writable = false;
|
|
|
|
for (var i = 0, n = this._streams.length; i < n; i++) {
|
|
this._streams[i].emit('end');
|
|
this._streams[i].readable = false;
|
|
}
|
|
|
|
this._context = this._streams = this._queue = [];
|
|
};
|
|
|
|
StreamReader.prototype.pause = function() {
|
|
this._paused = true;
|
|
if (this._parent) this._parent.pause();
|
|
};
|
|
|
|
StreamReader.prototype.resume = function() {
|
|
this._paused = false;
|
|
this.emit('drain');
|
|
if (this._parent) this._parent.resume();
|
|
};
|
|
|
|
StreamReader.prototype.fork = function(length) {
|
|
if (!this.writable) return null;
|
|
|
|
var stream = new StreamReader({context: this._context}, this),
|
|
self = this;
|
|
|
|
stream._remaining = length;
|
|
this._streams.push(stream);
|
|
defer(function() { self._flush() });
|
|
|
|
return stream;
|
|
};
|
|
|
|
StreamReader.prototype.read = function(length, callback) {
|
|
if (!this.writable) return;
|
|
|
|
if (this._queueSize >= length)
|
|
return callback.call(this._context, this._readBytes(length));
|
|
|
|
this.fork(length).pipe(new Concat(callback, this._context));
|
|
};
|
|
|
|
StreamReader.prototype._flush = function() {
|
|
var streams = this._streams, stream, size, buffer;
|
|
|
|
while (streams.length > 0) {
|
|
stream = streams[0];
|
|
size = Math.min(stream._remaining, this._queueSize);
|
|
buffer = this._readBytes(size);
|
|
|
|
if (size > 0) stream.emit('data', buffer);
|
|
stream._remaining -= size;
|
|
|
|
if (stream._remaining > 0) break;
|
|
|
|
stream.readable = false;
|
|
stream.emit('end');
|
|
streams.shift();
|
|
}
|
|
};
|
|
|
|
StreamReader.prototype._readBytes = function(length) {
|
|
var queue = this._queue,
|
|
remain = length,
|
|
n = queue.length,
|
|
first = queue[0],
|
|
i = 0,
|
|
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];
|
|
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;
|
|
}
|
|
|
|
queue.splice(0, this._cursor === 0 ? i : i - 1);
|
|
|
|
return buffer;
|
|
};
|
|
|
|
module.exports = StreamReader;
|