[Pkg-javascript-commits] [node-stream-browserify] 03/44: add all the other stream files
Bastien Roucariès
rouca at moszumanska.debian.org
Sun Aug 13 14:18:41 UTC 2017
This is an automated email from the git hooks/post-receive script.
rouca pushed a commit to branch master
in repository node-stream-browserify.
commit 053b1476184ddcbbc54f84164fbc7df39c8e2d67
Author: James Halliday <mail at substack.net>
Date: Sat Dec 7 22:28:28 2013 -0800
add all the other stream files
---
duplex.js | 73 +++++
index.js | 10 +-
package.json | 3 +-
passthrough.js | 41 +++
readable.js | 932 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
transform.js | 204 +++++++++++++
writable.js | 368 +++++++++++++++++++++++
7 files changed, 1625 insertions(+), 6 deletions(-)
diff --git a/duplex.js b/duplex.js
new file mode 100644
index 0000000..6925059
--- /dev/null
+++ b/duplex.js
@@ -0,0 +1,73 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+// a duplex stream is just a stream that is both readable and writable.
+// Since JS doesn't have multiple prototypal inheritance, this class
+// prototypally inherits from Readable, and then parasitically from
+// Writable.
+
+module.exports = Duplex;
+var inherits = require('inherits');
+var setImmediate = require('setimmediate');
+var Readable = require('./readable.js');
+var Writable = require('./writable.js');
+
+inherits(Duplex, Readable);
+
+Duplex.prototype.pipe = Writable.prototype.pipe;
+Duplex.prototype.write = Writable.prototype.write;
+Duplex.prototype.end = Writable.prototype.end;
+Duplex.prototype._write = Writable.prototype._write;
+
+function Duplex(options) {
+ if (!(this instanceof Duplex))
+ return new Duplex(options);
+
+ Readable.call(this, options);
+ Writable.call(this, options);
+
+ if (options && options.readable === false)
+ this.readable = false;
+
+ if (options && options.writable === false)
+ this.writable = false;
+
+ this.allowHalfOpen = true;
+ if (options && options.allowHalfOpen === false)
+ this.allowHalfOpen = false;
+
+ this.once('end', onend);
+}
+
+// the no-half-open enforcer
+function onend() {
+ // if we allow half-open state, or if the writable side ended,
+ // then we're ok.
+ if (this.allowHalfOpen || this._writableState.ended)
+ return;
+
+ // no more data can be written.
+ // But allow more writes to happen in this tick.
+ var self = this;
+ setImmediate(function () {
+ self.end();
+ });
+}
diff --git a/index.js b/index.js
index 8e05c62..5462325 100644
--- a/index.js
+++ b/index.js
@@ -25,11 +25,11 @@ var EE = require('events').EventEmitter;
var inherits = require('inherits');
inherits(Stream, EE);
-Stream.Readable = require('_stream_readable');
-Stream.Writable = require('_stream_writable');
-Stream.Duplex = require('_stream_duplex');
-Stream.Transform = require('_stream_transform');
-Stream.PassThrough = require('_stream_passthrough');
+Stream.Readable = require('./readable.js');
+Stream.Writable = require('./writable.js');
+Stream.Duplex = require('./duplex.js');
+Stream.Transform = require('./transform.js');
+Stream.PassThrough = require('./passthrough.js');
// Backwards-compat with node 0.4.x
Stream.Stream = Stream;
diff --git a/package.json b/package.json
index a157900..96d47ea 100644
--- a/package.json
+++ b/package.json
@@ -4,7 +4,8 @@
"description": "the stream module from node core for browsers",
"main": "index.js",
"dependencies": {
- "inherits": "~2.0.1"
+ "inherits": "~2.0.1",
+ "setimmediate": "~1.0.1"
},
"repository": {
"type": "git",
diff --git a/passthrough.js b/passthrough.js
new file mode 100644
index 0000000..754901d
--- /dev/null
+++ b/passthrough.js
@@ -0,0 +1,41 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+// a passthrough stream.
+// basically just the most minimal sort of Transform stream.
+// Every written chunk gets output as-is.
+
+module.exports = PassThrough;
+
+var Transform = require('./transform.js');
+var inherits = require('inherits');
+inherits(PassThrough, Transform);
+
+function PassThrough(options) {
+ if (!(this instanceof PassThrough))
+ return new PassThrough(options);
+
+ Transform.call(this, options);
+}
+
+PassThrough.prototype._transform = function(chunk, encoding, cb) {
+ cb(null, chunk);
+};
diff --git a/readable.js b/readable.js
new file mode 100644
index 0000000..3c4c7d9
--- /dev/null
+++ b/readable.js
@@ -0,0 +1,932 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+module.exports = Readable;
+Readable.ReadableState = ReadableState;
+
+var EE = require('events').EventEmitter;
+var Stream = require('./index.js');
+var Buffer = require('buffer').Buffer;
+var setImmediate = require('setimmediate');
+var StringDecoder;
+
+inherits(Readable, Stream);
+
+function ReadableState(options, stream) {
+ options = options || {};
+
+ // the point at which it stops calling _read() to fill the buffer
+ // Note: 0 is a valid value, means "don't call _read preemptively ever"
+ var hwm = options.highWaterMark;
+ this.highWaterMark = (hwm || hwm === 0) ? hwm : 16 * 1024;
+
+ // cast to ints.
+ this.highWaterMark = ~~this.highWaterMark;
+
+ this.buffer = [];
+ this.length = 0;
+ this.pipes = null;
+ this.pipesCount = 0;
+ this.flowing = false;
+ this.ended = false;
+ this.endEmitted = false;
+ this.reading = false;
+
+ // In streams that never have any data, and do push(null) right away,
+ // the consumer can miss the 'end' event if they do some I/O before
+ // consuming the stream. So, we don't emit('end') until some reading
+ // happens.
+ this.calledRead = false;
+
+ // a flag to be able to tell if the onwrite cb is called immediately,
+ // or on a later tick. We set this to true at first, becuase any
+ // actions that shouldn't happen until "later" should generally also
+ // not happen before the first write call.
+ this.sync = true;
+
+ // whenever we return null, then we set a flag to say
+ // that we're awaiting a 'readable' event emission.
+ this.needReadable = false;
+ this.emittedReadable = false;
+ this.readableListening = false;
+
+
+ // object stream flag. Used to make read(n) ignore n and to
+ // make all the buffer merging and length checks go away
+ this.objectMode = !!options.objectMode;
+
+ // Crypto is kind of old and crusty. Historically, its default string
+ // encoding is 'binary' so we have to make this configurable.
+ // Everything else in the universe uses 'utf8', though.
+ this.defaultEncoding = options.defaultEncoding || 'utf8';
+
+ // when piping, we only care about 'readable' events that happen
+ // after read()ing all the bytes and not getting any pushback.
+ this.ranOut = false;
+
+ // the number of writers that are awaiting a drain event in .pipe()s
+ this.awaitDrain = 0;
+
+ // if true, a maybeReadMore has been scheduled
+ this.readingMore = false;
+
+ this.decoder = null;
+ this.encoding = null;
+ if (options.encoding) {
+ if (!StringDecoder)
+ StringDecoder = require('string_decoder').StringDecoder;
+ this.decoder = new StringDecoder(options.encoding);
+ this.encoding = options.encoding;
+ }
+}
+
+function Readable(options) {
+ if (!(this instanceof Readable))
+ return new Readable(options);
+
+ this._readableState = new ReadableState(options, this);
+
+ // legacy
+ this.readable = true;
+
+ Stream.call(this);
+}
+
+// Manually shove something into the read() buffer.
+// This returns true if the highWaterMark has not been hit yet,
+// similar to how Writable.write() returns true if you should
+// write() some more.
+Readable.prototype.push = function(chunk, encoding) {
+ var state = this._readableState;
+
+ if (typeof chunk === 'string' && !state.objectMode) {
+ encoding = encoding || state.defaultEncoding;
+ if (encoding !== state.encoding) {
+ chunk = new Buffer(chunk, encoding);
+ encoding = '';
+ }
+ }
+
+ return readableAddChunk(this, state, chunk, encoding, false);
+};
+
+// Unshift should *always* be something directly out of read()
+Readable.prototype.unshift = function(chunk) {
+ var state = this._readableState;
+ return readableAddChunk(this, state, chunk, '', true);
+};
+
+function readableAddChunk(stream, state, chunk, encoding, addToFront) {
+ var er = chunkInvalid(state, chunk);
+ if (er) {
+ stream.emit('error', er);
+ } else if (chunk === null || chunk === undefined) {
+ state.reading = false;
+ if (!state.ended)
+ onEofChunk(stream, state);
+ } else if (state.objectMode || chunk && chunk.length > 0) {
+ if (state.ended && !addToFront) {
+ var e = new Error('stream.push() after EOF');
+ stream.emit('error', e);
+ } else if (state.endEmitted && addToFront) {
+ var e = new Error('stream.unshift() after end event');
+ stream.emit('error', e);
+ } else {
+ if (state.decoder && !addToFront && !encoding)
+ chunk = state.decoder.write(chunk);
+
+ // update the buffer info.
+ state.length += state.objectMode ? 1 : chunk.length;
+ if (addToFront) {
+ state.buffer.unshift(chunk);
+ } else {
+ state.reading = false;
+ state.buffer.push(chunk);
+ }
+
+ if (state.needReadable)
+ emitReadable(stream);
+
+ maybeReadMore(stream, state);
+ }
+ } else if (!addToFront) {
+ state.reading = false;
+ }
+
+ return needMoreData(state);
+}
+
+
+
+// if it's past the high water mark, we can push in some more.
+// Also, if we have no data yet, we can stand some
+// more bytes. This is to work around cases where hwm=0,
+// such as the repl. Also, if the push() triggered a
+// readable event, and the user called read(largeNumber) such that
+// needReadable was set, then we ought to push more, so that another
+// 'readable' event will be triggered.
+function needMoreData(state) {
+ return !state.ended &&
+ (state.needReadable ||
+ state.length < state.highWaterMark ||
+ state.length === 0);
+}
+
+// backwards compatibility.
+Readable.prototype.setEncoding = function(enc) {
+ if (!StringDecoder)
+ StringDecoder = require('string_decoder').StringDecoder;
+ this._readableState.decoder = new StringDecoder(enc);
+ this._readableState.encoding = enc;
+};
+
+// Don't raise the hwm > 128MB
+var MAX_HWM = 0x800000;
+function roundUpToNextPowerOf2(n) {
+ if (n >= MAX_HWM) {
+ n = MAX_HWM;
+ } else {
+ // Get the next highest power of 2
+ n--;
+ for (var p = 1; p < 32; p <<= 1) n |= n >> p;
+ n++;
+ }
+ return n;
+}
+
+function howMuchToRead(n, state) {
+ if (state.length === 0 && state.ended)
+ return 0;
+
+ if (state.objectMode)
+ return n === 0 ? 0 : 1;
+
+ if (isNaN(n) || n === null) {
+ // only flow one buffer at a time
+ if (state.flowing && state.buffer.length)
+ return state.buffer[0].length;
+ else
+ return state.length;
+ }
+
+ if (n <= 0)
+ return 0;
+
+ // If we're asking for more than the target buffer level,
+ // then raise the water mark. Bump up to the next highest
+ // power of 2, to prevent increasing it excessively in tiny
+ // amounts.
+ if (n > state.highWaterMark)
+ state.highWaterMark = roundUpToNextPowerOf2(n);
+
+ // don't have that much. return null, unless we've ended.
+ if (n > state.length) {
+ if (!state.ended) {
+ state.needReadable = true;
+ return 0;
+ } else
+ return state.length;
+ }
+
+ return n;
+}
+
+// you can override either this method, or the async _read(n) below.
+Readable.prototype.read = function(n) {
+ var state = this._readableState;
+ state.calledRead = true;
+ var nOrig = n;
+
+ if (typeof n !== 'number' || n > 0)
+ state.emittedReadable = false;
+
+ // if we're doing read(0) to trigger a readable event, but we
+ // already have a bunch of data in the buffer, then just trigger
+ // the 'readable' event and move on.
+ if (n === 0 &&
+ state.needReadable &&
+ (state.length >= state.highWaterMark || state.ended)) {
+ emitReadable(this);
+ return null;
+ }
+
+ n = howMuchToRead(n, state);
+
+ // if we've ended, and we're now clear, then finish it up.
+ if (n === 0 && state.ended) {
+ if (state.length === 0)
+ endReadable(this);
+ return null;
+ }
+
+ // All the actual chunk generation logic needs to be
+ // *below* the call to _read. The reason is that in certain
+ // synthetic stream cases, such as passthrough streams, _read
+ // may be a completely synchronous operation which may change
+ // the state of the read buffer, providing enough data when
+ // before there was *not* enough.
+ //
+ // So, the steps are:
+ // 1. Figure out what the state of things will be after we do
+ // a read from the buffer.
+ //
+ // 2. If that resulting state will trigger a _read, then call _read.
+ // Note that this may be asynchronous, or synchronous. Yes, it is
+ // deeply ugly to write APIs this way, but that still doesn't mean
+ // that the Readable class should behave improperly, as streams are
+ // designed to be sync/async agnostic.
+ // Take note if the _read call is sync or async (ie, if the read call
+ // has returned yet), so that we know whether or not it's safe to emit
+ // 'readable' etc.
+ //
+ // 3. Actually pull the requested chunks out of the buffer and return.
+
+ // if we need a readable event, then we need to do some reading.
+ var doRead = state.needReadable;
+
+ // if we currently have less than the highWaterMark, then also read some
+ if (state.length - n <= state.highWaterMark)
+ doRead = true;
+
+ // however, if we've ended, then there's no point, and if we're already
+ // reading, then it's unnecessary.
+ if (state.ended || state.reading)
+ doRead = false;
+
+ if (doRead) {
+ state.reading = true;
+ state.sync = true;
+ // if the length is currently zero, then we *need* a readable event.
+ if (state.length === 0)
+ state.needReadable = true;
+ // call internal read method
+ this._read(state.highWaterMark);
+ state.sync = false;
+ }
+
+ // If _read called its callback synchronously, then `reading`
+ // will be false, and we need to re-evaluate how much data we
+ // can return to the user.
+ if (doRead && !state.reading)
+ n = howMuchToRead(nOrig, state);
+
+ var ret;
+ if (n > 0)
+ ret = fromList(n, state);
+ else
+ ret = null;
+
+ if (ret === null) {
+ state.needReadable = true;
+ n = 0;
+ }
+
+ state.length -= n;
+
+ // If we have nothing in the buffer, then we want to know
+ // as soon as we *do* get something into the buffer.
+ if (state.length === 0 && !state.ended)
+ state.needReadable = true;
+
+ // If we happened to read() exactly the remaining amount in the
+ // buffer, and the EOF has been seen at this point, then make sure
+ // that we emit 'end' on the very next tick.
+ if (state.ended && !state.endEmitted && state.length === 0)
+ endReadable(this);
+
+ return ret;
+};
+
+function chunkInvalid(state, chunk) {
+ var er = null;
+ if (!Buffer.isBuffer(chunk) &&
+ 'string' !== typeof chunk &&
+ chunk !== null &&
+ chunk !== undefined &&
+ !state.objectMode &&
+ !er) {
+ er = new TypeError('Invalid non-string/buffer chunk');
+ }
+ return er;
+}
+
+
+function onEofChunk(stream, state) {
+ if (state.decoder && !state.ended) {
+ var chunk = state.decoder.end();
+ if (chunk && chunk.length) {
+ state.buffer.push(chunk);
+ state.length += state.objectMode ? 1 : chunk.length;
+ }
+ }
+ state.ended = true;
+
+ // if we've ended and we have some data left, then emit
+ // 'readable' now to make sure it gets picked up.
+ if (state.length > 0)
+ emitReadable(stream);
+ else
+ endReadable(stream);
+}
+
+// Don't emit readable right away in sync mode, because this can trigger
+// another read() call => stack overflow. This way, it might trigger
+// a nextTick recursion warning, but that's not so bad.
+function emitReadable(stream) {
+ var state = stream._readableState;
+ state.needReadable = false;
+ if (state.emittedReadable)
+ return;
+
+ state.emittedReadable = true;
+ if (state.sync)
+ setImmediate(function() {
+ emitReadable_(stream);
+ });
+ else
+ emitReadable_(stream);
+}
+
+function emitReadable_(stream) {
+ stream.emit('readable');
+}
+
+
+// at this point, the user has presumably seen the 'readable' event,
+// and called read() to consume some data. that may have triggered
+// in turn another _read(n) call, in which case reading = true if
+// it's in progress.
+// However, if we're not ended, or reading, and the length < hwm,
+// then go ahead and try to read some more preemptively.
+function maybeReadMore(stream, state) {
+ if (!state.readingMore) {
+ state.readingMore = true;
+ setImmediate(function() {
+ maybeReadMore_(stream, state);
+ });
+ }
+}
+
+function maybeReadMore_(stream, state) {
+ var len = state.length;
+ while (!state.reading && !state.flowing && !state.ended &&
+ state.length < state.highWaterMark) {
+ stream.read(0);
+ if (len === state.length)
+ // didn't get any data, stop spinning.
+ break;
+ else
+ len = state.length;
+ }
+ state.readingMore = false;
+}
+
+// abstract method. to be overridden in specific implementation classes.
+// call cb(er, data) where data is <= n in length.
+// for virtual (non-string, non-buffer) streams, "length" is somewhat
+// arbitrary, and perhaps not very meaningful.
+Readable.prototype._read = function(n) {
+ this.emit('error', new Error('not implemented'));
+};
+
+Readable.prototype.pipe = function(dest, pipeOpts) {
+ var src = this;
+ var state = this._readableState;
+
+ switch (state.pipesCount) {
+ case 0:
+ state.pipes = dest;
+ break;
+ case 1:
+ state.pipes = [state.pipes, dest];
+ break;
+ default:
+ state.pipes.push(dest);
+ break;
+ }
+ state.pipesCount += 1;
+
+ var doEnd = (!pipeOpts || pipeOpts.end !== false) &&
+ dest !== process.stdout &&
+ dest !== process.stderr;
+
+ var endFn = doEnd ? onend : cleanup;
+ if (state.endEmitted)
+ setImmediate(endFn);
+ else
+ src.once('end', endFn);
+
+ dest.on('unpipe', onunpipe);
+ function onunpipe(readable) {
+ if (readable !== src) return;
+ cleanup();
+ }
+
+ function onend() {
+ dest.end();
+ }
+
+ // when the dest drains, it reduces the awaitDrain counter
+ // on the source. This would be more elegant with a .once()
+ // handler in flow(), but adding and removing repeatedly is
+ // too slow.
+ var ondrain = pipeOnDrain(src);
+ dest.on('drain', ondrain);
+
+ function cleanup() {
+ // cleanup event handlers once the pipe is broken
+ dest.removeListener('close', onclose);
+ dest.removeListener('finish', onfinish);
+ dest.removeListener('drain', ondrain);
+ dest.removeListener('error', onerror);
+ dest.removeListener('unpipe', onunpipe);
+ src.removeListener('end', onend);
+ src.removeListener('end', cleanup);
+
+ // if the reader is waiting for a drain event from this
+ // specific writer, then it would cause it to never start
+ // flowing again.
+ // So, if this is awaiting a drain, then we just call it now.
+ // If we don't know, then assume that we are waiting for one.
+ if (!dest._writableState || dest._writableState.needDrain)
+ ondrain();
+ }
+
+ // if the dest has an error, then stop piping into it.
+ // however, don't suppress the throwing behavior for this.
+ // check for listeners before emit removes one-time listeners.
+ var errListeners = EE.listenerCount(dest, 'error');
+ function onerror(er) {
+ unpipe();
+ if (errListeners === 0 && EE.listenerCount(dest, 'error') === 0)
+ dest.emit('error', er);
+ }
+ dest.once('error', onerror);
+
+ // Both close and finish should trigger unpipe, but only once.
+ function onclose() {
+ dest.removeListener('finish', onfinish);
+ unpipe();
+ }
+ dest.once('close', onclose);
+ function onfinish() {
+ dest.removeListener('close', onclose);
+ unpipe();
+ }
+ dest.once('finish', onfinish);
+
+ function unpipe() {
+ src.unpipe(dest);
+ }
+
+ // tell the dest that it's being piped to
+ dest.emit('pipe', src);
+
+ // start the flow if it hasn't been started already.
+ if (!state.flowing) {
+ // the handler that waits for readable events after all
+ // the data gets sucked out in flow.
+ // This would be easier to follow with a .once() handler
+ // in flow(), but that is too slow.
+ this.on('readable', pipeOnReadable);
+
+ state.flowing = true;
+ setImmediate(function() {
+ flow(src);
+ });
+ }
+
+ return dest;
+};
+
+function pipeOnDrain(src) {
+ return function() {
+ var dest = this;
+ var state = src._readableState;
+ state.awaitDrain--;
+ if (state.awaitDrain === 0)
+ flow(src);
+ };
+}
+
+function flow(src) {
+ var state = src._readableState;
+ var chunk;
+ state.awaitDrain = 0;
+
+ function write(dest, i, list) {
+ var written = dest.write(chunk);
+ if (false === written) {
+ state.awaitDrain++;
+ }
+ }
+
+ while (state.pipesCount && null !== (chunk = src.read())) {
+
+ if (state.pipesCount === 1)
+ write(state.pipes, 0, null);
+ else
+ forEach(state.pipes, write);
+
+ src.emit('data', chunk);
+
+ // if anyone needs a drain, then we have to wait for that.
+ if (state.awaitDrain > 0)
+ return;
+ }
+
+ // if every destination was unpiped, either before entering this
+ // function, or in the while loop, then stop flowing.
+ //
+ // NB: This is a pretty rare edge case.
+ if (state.pipesCount === 0) {
+ state.flowing = false;
+
+ // if there were data event listeners added, then switch to old mode.
+ if (EE.listenerCount(src, 'data') > 0)
+ emitDataEvents(src);
+ return;
+ }
+
+ // at this point, no one needed a drain, so we just ran out of data
+ // on the next readable event, start it over again.
+ state.ranOut = true;
+}
+
+function pipeOnReadable() {
+ if (this._readableState.ranOut) {
+ this._readableState.ranOut = false;
+ flow(this);
+ }
+}
+
+
+Readable.prototype.unpipe = function(dest) {
+ var state = this._readableState;
+
+ // if we're not piping anywhere, then do nothing.
+ if (state.pipesCount === 0)
+ return this;
+
+ // just one destination. most common case.
+ if (state.pipesCount === 1) {
+ // passed in one, but it's not the right one.
+ if (dest && dest !== state.pipes)
+ return this;
+
+ if (!dest)
+ dest = state.pipes;
+
+ // got a match.
+ state.pipes = null;
+ state.pipesCount = 0;
+ this.removeListener('readable', pipeOnReadable);
+ state.flowing = false;
+ if (dest)
+ dest.emit('unpipe', this);
+ return this;
+ }
+
+ // slow case. multiple pipe destinations.
+
+ if (!dest) {
+ // remove all.
+ var dests = state.pipes;
+ var len = state.pipesCount;
+ state.pipes = null;
+ state.pipesCount = 0;
+ this.removeListener('readable', pipeOnReadable);
+ state.flowing = false;
+
+ for (var i = 0; i < len; i++)
+ dests[i].emit('unpipe', this);
+ return this;
+ }
+
+ // try to find the right one.
+ var i = indexOf(state.pipes, dest);
+ if (i === -1)
+ return this;
+
+ state.pipes.splice(i, 1);
+ state.pipesCount -= 1;
+ if (state.pipesCount === 1)
+ state.pipes = state.pipes[0];
+
+ dest.emit('unpipe', this);
+
+ return this;
+};
+
+// set up data events if they are asked for
+// Ensure readable listeners eventually get something
+Readable.prototype.on = function(ev, fn) {
+ var res = Stream.prototype.on.call(this, ev, fn);
+
+ if (ev === 'data' && !this._readableState.flowing)
+ emitDataEvents(this);
+
+ if (ev === 'readable' && this.readable) {
+ var state = this._readableState;
+ if (!state.readableListening) {
+ state.readableListening = true;
+ state.emittedReadable = false;
+ state.needReadable = true;
+ if (!state.reading) {
+ this.read(0);
+ } else if (state.length) {
+ emitReadable(this, state);
+ }
+ }
+ }
+
+ return res;
+};
+Readable.prototype.addListener = Readable.prototype.on;
+
+// pause() and resume() are remnants of the legacy readable stream API
+// If the user uses them, then switch into old mode.
+Readable.prototype.resume = function() {
+ emitDataEvents(this);
+ this.read(0);
+ this.emit('resume');
+};
+
+Readable.prototype.pause = function() {
+ emitDataEvents(this, true);
+ this.emit('pause');
+};
+
+function emitDataEvents(stream, startPaused) {
+ var state = stream._readableState;
+
+ if (state.flowing) {
+ // https://github.com/isaacs/readable-stream/issues/16
+ throw new Error('Cannot switch to old mode now.');
+ }
+
+ var paused = startPaused || false;
+ var readable = false;
+
+ // convert to an old-style stream.
+ stream.readable = true;
+ stream.pipe = Stream.prototype.pipe;
+ stream.on = stream.addListener = Stream.prototype.on;
+
+ stream.on('readable', function() {
+ readable = true;
+
+ var c;
+ while (!paused && (null !== (c = stream.read())))
+ stream.emit('data', c);
+
+ if (c === null) {
+ readable = false;
+ stream._readableState.needReadable = true;
+ }
+ });
+
+ stream.pause = function() {
+ paused = true;
+ this.emit('pause');
+ };
+
+ stream.resume = function() {
+ paused = false;
+ if (readable)
+ setImmediate(function() {
+ stream.emit('readable');
+ });
+ else
+ this.read(0);
+ this.emit('resume');
+ };
+
+ // now make it start, just in case it hadn't already.
+ stream.emit('readable');
+}
+
+// wrap an old-style stream as the async data source.
+// This is *not* part of the readable stream interface.
+// It is an ugly unfortunate mess of history.
+Readable.prototype.wrap = function(stream) {
+ var state = this._readableState;
+ var paused = false;
+
+ var self = this;
+ stream.on('end', function() {
+ if (state.decoder && !state.ended) {
+ var chunk = state.decoder.end();
+ if (chunk && chunk.length)
+ self.push(chunk);
+ }
+
+ self.push(null);
+ });
+
+ stream.on('data', function(chunk) {
+ if (state.decoder)
+ chunk = state.decoder.write(chunk);
+ if (!chunk || !state.objectMode && !chunk.length)
+ return;
+
+ var ret = self.push(chunk);
+ if (!ret) {
+ paused = true;
+ stream.pause();
+ }
+ });
+
+ // proxy all the other methods.
+ // important when wrapping filters and duplexes.
+ for (var i in stream) {
+ if (typeof stream[i] === 'function' &&
+ typeof this[i] === 'undefined') {
+ this[i] = function(method) { return function() {
+ return stream[method].apply(stream, arguments);
+ }}(i);
+ }
+ }
+
+ // proxy certain important events.
+ var events = ['error', 'close', 'destroy', 'pause', 'resume'];
+ forEach(events, function(ev) {
+ stream.on(ev, function () {
+ return self.emit.apply(self, [ev].concat([].slice.call(arguments)));
+ });
+ });
+
+ // when we try to consume some more bytes, simply unpause the
+ // underlying stream.
+ self._read = function(n) {
+ if (paused) {
+ paused = false;
+ stream.resume();
+ }
+ };
+
+ return self;
+};
+
+
+
+// exposed for testing purposes only.
+Readable._fromList = fromList;
+
+// Pluck off n bytes from an array of buffers.
+// Length is the combined lengths of all the buffers in the list.
+function fromList(n, state) {
+ var list = state.buffer;
+ var length = state.length;
+ var stringMode = !!state.decoder;
+ var objectMode = !!state.objectMode;
+ var ret;
+
+ // nothing in the list, definitely empty.
+ if (list.length === 0)
+ return null;
+
+ if (length === 0)
+ ret = null;
+ else if (objectMode)
+ ret = list.shift();
+ else if (!n || n >= length) {
+ // read it all, truncate the array.
+ if (stringMode)
+ ret = list.join('');
+ else
+ ret = Buffer.concat(list, length);
+ list.length = 0;
+ } else {
+ // read just some of it.
+ if (n < list[0].length) {
+ // just take a part of the first list item.
+ // slice is the same for buffers and strings.
+ var buf = list[0];
+ ret = buf.slice(0, n);
+ list[0] = buf.slice(n);
+ } else if (n === list[0].length) {
+ // first list is a perfect match
+ ret = list.shift();
+ } else {
+ // complex case.
+ // we have enough to cover it, but it spans past the first buffer.
+ if (stringMode)
+ ret = '';
+ else
+ ret = new Buffer(n);
+
+ var c = 0;
+ for (var i = 0, l = list.length; i < l && c < n; i++) {
+ var buf = list[0];
+ var cpy = Math.min(n - c, buf.length);
+
+ if (stringMode)
+ ret += buf.slice(0, cpy);
+ else
+ buf.copy(ret, c, 0, cpy);
+
+ if (cpy < buf.length)
+ list[0] = buf.slice(cpy);
+ else
+ list.shift();
+
+ c += cpy;
+ }
+ }
+ }
+
+ return ret;
+}
+
+function endReadable(stream) {
+ var state = stream._readableState;
+
+ // If we get here before consuming all the bytes, then that is a
+ // bug in node. Should never happen.
+ if (state.length > 0)
+ throw new Error('endReadable called on non-empty stream');
+
+ if (!state.endEmitted && state.calledRead) {
+ state.ended = true;
+ setImmediate(function() {
+ // Check that we didn't get one last unshift.
+ if (!state.endEmitted && state.length === 0) {
+ state.endEmitted = true;
+ stream.readable = false;
+ stream.emit('end');
+ }
+ });
+ }
+}
+
+function forEach (xs, f) {
+ for (var i = 0, l = xs.length; i < l; i++) {
+ f(xs[i], i);
+ }
+}
+
+function indexOf (xs, x) {
+ for (var i = 0, l = xs.length; i < l; i++) {
+ if (xs[i] === x) return i;
+ }
+ return -1;
+}
diff --git a/transform.js b/transform.js
new file mode 100644
index 0000000..502becd
--- /dev/null
+++ b/transform.js
@@ -0,0 +1,204 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+// a transform stream is a readable/writable stream where you do
+// something with the data. Sometimes it's called a "filter",
+// but that's not a great name for it, since that implies a thing where
+// some bits pass through, and others are simply ignored. (That would
+// be a valid example of a transform, of course.)
+//
+// While the output is causally related to the input, it's not a
+// necessarily symmetric or synchronous transformation. For example,
+// a zlib stream might take multiple plain-text writes(), and then
+// emit a single compressed chunk some time in the future.
+//
+// Here's how this works:
+//
+// The Transform stream has all the aspects of the readable and writable
+// stream classes. When you write(chunk), that calls _write(chunk,cb)
+// internally, and returns false if there's a lot of pending writes
+// buffered up. When you call read(), that calls _read(n) until
+// there's enough pending readable data buffered up.
+//
+// In a transform stream, the written data is placed in a buffer. When
+// _read(n) is called, it transforms the queued up data, calling the
+// buffered _write cb's as it consumes chunks. If consuming a single
+// written chunk would result in multiple output chunks, then the first
+// outputted bit calls the readcb, and subsequent chunks just go into
+// the read buffer, and will cause it to emit 'readable' if necessary.
+//
+// This way, back-pressure is actually determined by the reading side,
+// since _read has to be called to start processing a new chunk. However,
+// a pathological inflate type of transform can cause excessive buffering
+// here. For example, imagine a stream where every byte of input is
+// interpreted as an integer from 0-255, and then results in that many
+// bytes of output. Writing the 4 bytes {ff,ff,ff,ff} would result in
+// 1kb of data being output. In this case, you could write a very small
+// amount of input, and end up with a very large amount of output. In
+// such a pathological inflating mechanism, there'd be no way to tell
+// the system to stop doing the transform. A single 4MB write could
+// cause the system to run out of memory.
+//
+// However, even in such a pathological case, only a single written chunk
+// would be consumed, and then the rest would wait (un-transformed) until
+// the results of the previous transformed chunk were consumed.
+
+module.exports = Transform;
+
+var Duplex = require('./duplex.js');
+var inherits = require('inherits');
+inherits(Transform, Duplex);
+
+
+function TransformState(options, stream) {
+ this.afterTransform = function(er, data) {
+ return afterTransform(stream, er, data);
+ };
+
+ this.needTransform = false;
+ this.transforming = false;
+ this.writecb = null;
+ this.writechunk = null;
+}
+
+function afterTransform(stream, er, data) {
+ var ts = stream._transformState;
+ ts.transforming = false;
+
+ var cb = ts.writecb;
+
+ if (!cb)
+ return stream.emit('error', new Error('no writecb in Transform class'));
+
+ ts.writechunk = null;
+ ts.writecb = null;
+
+ if (data !== null && data !== undefined)
+ stream.push(data);
+
+ if (cb)
+ cb(er);
+
+ var rs = stream._readableState;
+ rs.reading = false;
+ if (rs.needReadable || rs.length < rs.highWaterMark) {
+ stream._read(rs.highWaterMark);
+ }
+}
+
+
+function Transform(options) {
+ if (!(this instanceof Transform))
+ return new Transform(options);
+
+ Duplex.call(this, options);
+
+ var ts = this._transformState = new TransformState(options, this);
+
+ // when the writable side finishes, then flush out anything remaining.
+ var stream = this;
+
+ // start out asking for a readable event once data is transformed.
+ this._readableState.needReadable = true;
+
+ // we have implemented the _read method, and done the other things
+ // that Readable wants before the first _read call, so unset the
+ // sync guard flag.
+ this._readableState.sync = false;
+
+ this.once('finish', function() {
+ if ('function' === typeof this._flush)
+ this._flush(function(er) {
+ done(stream, er);
+ });
+ else
+ done(stream);
+ });
+}
+
+Transform.prototype.push = function(chunk, encoding) {
+ this._transformState.needTransform = false;
+ return Duplex.prototype.push.call(this, chunk, encoding);
+};
+
+// This is the part where you do stuff!
+// override this function in implementation classes.
+// 'chunk' is an input chunk.
+//
+// Call `push(newChunk)` to pass along transformed output
+// to the readable side. You may call 'push' zero or more times.
+//
+// Call `cb(err)` when you are done with this chunk. If you pass
+// an error, then that'll put the hurt on the whole operation. If you
+// never call cb(), then you'll never get another chunk.
+Transform.prototype._transform = function(chunk, encoding, cb) {
+ throw new Error('not implemented');
+};
+
+Transform.prototype._write = function(chunk, encoding, cb) {
+ var ts = this._transformState;
+ ts.writecb = cb;
+ ts.writechunk = chunk;
+ ts.writeencoding = encoding;
+ if (!ts.transforming) {
+ var rs = this._readableState;
+ if (ts.needTransform ||
+ rs.needReadable ||
+ rs.length < rs.highWaterMark)
+ this._read(rs.highWaterMark);
+ }
+};
+
+// Doesn't matter what the args are here.
+// _transform does all the work.
+// That we got here means that the readable side wants more data.
+Transform.prototype._read = function(n) {
+ var ts = this._transformState;
+
+ if (ts.writechunk && ts.writecb && !ts.transforming) {
+ ts.transforming = true;
+ this._transform(ts.writechunk, ts.writeencoding, ts.afterTransform);
+ } else {
+ // mark that we need a transform, so that any data that comes in
+ // will get processed, now that we've asked for it.
+ ts.needTransform = true;
+ }
+};
+
+
+function done(stream, er) {
+ if (er)
+ return stream.emit('error', er);
+
+ // if there's nothing in the write buffer, then that means
+ // that nothing more will ever be provided
+ var ws = stream._writableState;
+ var rs = stream._readableState;
+ var ts = stream._transformState;
+
+ if (ws.length)
+ throw new Error('calling transform done when ws.length != 0');
+
+ if (ts.transforming)
+ throw new Error('calling transform done when still transforming');
+
+ return stream.push(null);
+}
diff --git a/writable.js b/writable.js
new file mode 100644
index 0000000..2164530
--- /dev/null
+++ b/writable.js
@@ -0,0 +1,368 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+// A bit simpler than readable streams.
+// Implement an async ._write(chunk, cb), and it'll handle all
+// the drain event emission and buffering.
+
+module.exports = Writable;
+Writable.WritableState = WritableState;
+
+var inherits = require('inherits');
+var Stream = require('./index.js');
+var setImmediate = require('setimmediate');
+var Buffer = require('buffer').Buffer;
+
+inherits(Writable, Stream);
+
+function WriteReq(chunk, encoding, cb) {
+ this.chunk = chunk;
+ this.encoding = encoding;
+ this.callback = cb;
+}
+
+function WritableState(options, stream) {
+ options = options || {};
+
+ // the point at which write() starts returning false
+ // Note: 0 is a valid value, means that we always return false if
+ // the entire buffer is not flushed immediately on write()
+ var hwm = options.highWaterMark;
+ this.highWaterMark = (hwm || hwm === 0) ? hwm : 16 * 1024;
+
+ // object stream flag to indicate whether or not this stream
+ // contains buffers or objects.
+ this.objectMode = !!options.objectMode;
+
+ // cast to ints.
+ this.highWaterMark = ~~this.highWaterMark;
+
+ this.needDrain = false;
+ // at the start of calling end()
+ this.ending = false;
+ // when end() has been called, and returned
+ this.ended = false;
+ // when 'finish' is emitted
+ this.finished = false;
+
+ // should we decode strings into buffers before passing to _write?
+ // this is here so that some node-core streams can optimize string
+ // handling at a lower level.
+ var noDecode = options.decodeStrings === false;
+ this.decodeStrings = !noDecode;
+
+ // Crypto is kind of old and crusty. Historically, its default string
+ // encoding is 'binary' so we have to make this configurable.
+ // Everything else in the universe uses 'utf8', though.
+ this.defaultEncoding = options.defaultEncoding || 'utf8';
+
+ // not an actual buffer we keep track of, but a measurement
+ // of how much we're waiting to get pushed to some underlying
+ // socket or file.
+ this.length = 0;
+
+ // a flag to see when we're in the middle of a write.
+ this.writing = false;
+
+ // a flag to be able to tell if the onwrite cb is called immediately,
+ // or on a later tick. We set this to true at first, becuase any
+ // actions that shouldn't happen until "later" should generally also
+ // not happen before the first write call.
+ this.sync = true;
+
+ // a flag to know if we're processing previously buffered items, which
+ // may call the _write() callback in the same tick, so that we don't
+ // end up in an overlapped onwrite situation.
+ this.bufferProcessing = false;
+
+ // the callback that's passed to _write(chunk,cb)
+ this.onwrite = function(er) {
+ onwrite(stream, er);
+ };
+
+ // the callback that the user supplies to write(chunk,encoding,cb)
+ this.writecb = null;
+
+ // the amount that is being written when _write is called.
+ this.writelen = 0;
+
+ this.buffer = [];
+}
+
+function Writable(options) {
+ // Writable ctor is applied to Duplexes, though they're not
+ // instanceof Writable, they're instanceof Readable.
+ if (!(this instanceof Writable) && !(this instanceof Stream.Duplex))
+ return new Writable(options);
+
+ this._writableState = new WritableState(options, this);
+
+ // legacy.
+ this.writable = true;
+
+ Stream.call(this);
+}
+
+// Otherwise people can pipe Writable streams, which is just wrong.
+Writable.prototype.pipe = function() {
+ this.emit('error', new Error('Cannot pipe. Not readable.'));
+};
+
+
+function writeAfterEnd(stream, state, cb) {
+ var er = new Error('write after end');
+ // TODO: defer error events consistently everywhere, not just the cb
+ stream.emit('error', er);
+ setImmediate(function() {
+ cb(er);
+ });
+}
+
+// If we get something that is not a buffer, string, null, or undefined,
+// and we're not in objectMode, then that's an error.
+// Otherwise stream chunks are all considered to be of length=1, and the
+// watermarks determine how many objects to keep in the buffer, rather than
+// how many bytes or characters.
+function validChunk(stream, state, chunk, cb) {
+ var valid = true;
+ if (!Buffer.isBuffer(chunk) &&
+ 'string' !== typeof chunk &&
+ chunk !== null &&
+ chunk !== undefined &&
+ !state.objectMode) {
+ var er = new TypeError('Invalid non-string/buffer chunk');
+ stream.emit('error', er);
+ setImmediate(function() {
+ cb(er);
+ });
+ valid = false;
+ }
+ return valid;
+}
+
+Writable.prototype.write = function(chunk, encoding, cb) {
+ var state = this._writableState;
+ var ret = false;
+
+ if (typeof encoding === 'function') {
+ cb = encoding;
+ encoding = null;
+ }
+
+ if (Buffer.isBuffer(chunk))
+ encoding = 'buffer';
+ else if (!encoding)
+ encoding = state.defaultEncoding;
+
+ if (typeof cb !== 'function')
+ cb = function() {};
+
+ if (state.ended)
+ writeAfterEnd(this, state, cb);
+ else if (validChunk(this, state, chunk, cb))
+ ret = writeOrBuffer(this, state, chunk, encoding, cb);
+
+ return ret;
+};
+
+function decodeChunk(state, chunk, encoding) {
+ if (!state.objectMode &&
+ state.decodeStrings !== false &&
+ typeof chunk === 'string') {
+ chunk = new Buffer(chunk, encoding);
+ }
+ return chunk;
+}
+
+// if we're already writing something, then just put this
+// in the queue, and wait our turn. Otherwise, call _write
+// If we return false, then we need a drain event, so set that flag.
+function writeOrBuffer(stream, state, chunk, encoding, cb) {
+ chunk = decodeChunk(state, chunk, encoding);
+ var len = state.objectMode ? 1 : chunk.length;
+
+ state.length += len;
+
+ var ret = state.length < state.highWaterMark;
+ state.needDrain = !ret;
+
+ if (state.writing)
+ state.buffer.push(new WriteReq(chunk, encoding, cb));
+ else
+ doWrite(stream, state, len, chunk, encoding, cb);
+
+ return ret;
+}
+
+function doWrite(stream, state, len, chunk, encoding, cb) {
+ state.writelen = len;
+ state.writecb = cb;
+ state.writing = true;
+ state.sync = true;
+ stream._write(chunk, encoding, state.onwrite);
+ state.sync = false;
+}
+
+function onwriteError(stream, state, sync, er, cb) {
+ if (sync)
+ setImmediate(function() {
+ cb(er);
+ });
+ else
+ cb(er);
+
+ stream.emit('error', er);
+}
+
+function onwriteStateUpdate(state) {
+ state.writing = false;
+ state.writecb = null;
+ state.length -= state.writelen;
+ state.writelen = 0;
+}
+
+function onwrite(stream, er) {
+ var state = stream._writableState;
+ var sync = state.sync;
+ var cb = state.writecb;
+
+ onwriteStateUpdate(state);
+
+ if (er)
+ onwriteError(stream, state, sync, er, cb);
+ else {
+ // Check if we're actually ready to finish, but don't emit yet
+ var finished = needFinish(stream, state);
+
+ if (!finished && !state.bufferProcessing && state.buffer.length)
+ clearBuffer(stream, state);
+
+ if (sync) {
+ setImmediate(function() {
+ afterWrite(stream, state, finished, cb);
+ });
+ } else {
+ afterWrite(stream, state, finished, cb);
+ }
+ }
+}
+
+function afterWrite(stream, state, finished, cb) {
+ if (!finished)
+ onwriteDrain(stream, state);
+ cb();
+ if (finished)
+ finishMaybe(stream, state);
+}
+
+// Must force callback to be called on nextTick, so that we don't
+// emit 'drain' before the write() consumer gets the 'false' return
+// value, and has a chance to attach a 'drain' listener.
+function onwriteDrain(stream, state) {
+ if (state.length === 0 && state.needDrain) {
+ state.needDrain = false;
+ stream.emit('drain');
+ }
+}
+
+
+// if there's something in the buffer waiting, then process it
+function clearBuffer(stream, state) {
+ state.bufferProcessing = true;
+
+ for (var c = 0; c < state.buffer.length; c++) {
+ var entry = state.buffer[c];
+ var chunk = entry.chunk;
+ var encoding = entry.encoding;
+ var cb = entry.callback;
+ var len = state.objectMode ? 1 : chunk.length;
+
+ doWrite(stream, state, len, chunk, encoding, cb);
+
+ // if we didn't call the onwrite immediately, then
+ // it means that we need to wait until it does.
+ // also, that means that the chunk and cb are currently
+ // being processed, so move the buffer counter past them.
+ if (state.writing) {
+ c++;
+ break;
+ }
+ }
+
+ state.bufferProcessing = false;
+ if (c < state.buffer.length)
+ state.buffer = state.buffer.slice(c);
+ else
+ state.buffer.length = 0;
+}
+
+Writable.prototype._write = function(chunk, encoding, cb) {
+ cb(new Error('not implemented'));
+};
+
+Writable.prototype.end = function(chunk, encoding, cb) {
+ var state = this._writableState;
+
+ if (typeof chunk === 'function') {
+ cb = chunk;
+ chunk = null;
+ encoding = null;
+ } else if (typeof encoding === 'function') {
+ cb = encoding;
+ encoding = null;
+ }
+
+ if (typeof chunk !== 'undefined' && chunk !== null)
+ this.write(chunk, encoding);
+
+ // ignore unnecessary end() calls.
+ if (!state.ending && !state.finished)
+ endWritable(this, state, cb);
+};
+
+
+function needFinish(stream, state) {
+ return (state.ending &&
+ state.length === 0 &&
+ !state.finished &&
+ !state.writing);
+}
+
+function finishMaybe(stream, state) {
+ var need = needFinish(stream, state);
+ if (need) {
+ state.finished = true;
+ stream.emit('finish');
+ }
+ return need;
+}
+
+function endWritable(stream, state, cb) {
+ state.ending = true;
+ finishMaybe(stream, state);
+ if (cb) {
+ if (state.finished)
+ setImmediate(cb);
+ else
+ stream.once('finish', cb);
+ }
+ state.ended = true;
+}
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/pkg-javascript/node-stream-browserify.git
More information about the Pkg-javascript-commits
mailing list