lib/streams/compound-duplex.js
function CompoundDuplex(writable, readable, options) {
	var self = this;
	var convertToZStream = require('../index');	// for circular dependencies

	if(!readable || typeof readable.read !== 'function') {
		options = readable;
		readable = writable;
		writable = null;
	}

	if(writable && !writable._isZStream) {
		writable = convertToZStream(writable);
	}
	if(!readable._isZStream) {
		readable = convertToZStream(readable);
	}

	if(!writable) {
		if(typeof readable.getStreamChain !== 'function') {
			throw new Error('Can only use shorthand CompoundDuplex constructor if pipeline is all zstreams');
		}
		writable = readable.getStreamChain().getStreams()[0];
	}

	if(!options) options = {};

	options.readableObjectMode = readable.isReadableObjectMode();
	options.writableObjectMode = writable.isWritableObjectMode();
	Duplex.call(this, options);

	this._compoundReadable = readable;
	this._compoundWritable = writable;

	this._waitingForReadableData = false;

	writable.on('chainerror', function(error) {
		// Forward the error on; if the chain is to be destructed, the compound stream's _abortStream() method will be called
		this.ignoreError();
		self.emit('error', error);
	});

	readable.on('readable', function() {
		if(self._waitingForReadableData) {
			self._waitingForReadableData = false;
			self._readSomeData();
		}
	});

	readable.on('end', function() {
		self.push(null);
	});
}
function CompoundDuplex(writable, readable, options) {
	var self = this;
	var convertToZStream = require('../index');	// for circular dependencies

	if(!readable || typeof readable.read !== 'function') {
		options = readable;
		readable = writable;
		writable = null;
	}

	if(writable && !writable._isZStream) {
		writable = convertToZStream(writable);
	}
	if(!readable._isZStream) {
		readable = convertToZStream(readable);
	}

	if(!writable) {
		if(typeof readable.getStreamChain !== 'function') {
			throw new Error('Can only use shorthand CompoundDuplex constructor if pipeline is all zstreams');
		}
		writable = readable.getStreamChain().getStreams()[0];
	}

	if(!options) options = {};

	options.readableObjectMode = readable.isReadableObjectMode();
	options.writableObjectMode = writable.isWritableObjectMode();
	Duplex.call(this, options);

	this._compoundReadable = readable;
	this._compoundWritable = writable;

	this._waitingForReadableData = false;

	writable.on('chainerror', function(error) {
		// Forward the error on; if the chain is to be destructed, the compound stream's _abortStream() method will be called
		this.ignoreError();
		self.emit('error', error);
	});

	readable.on('readable', function() {
		if(self._waitingForReadableData) {
			self._waitingForReadableData = false;
			self._readSomeData();
		}
	});

	readable.on('end', function() {
		self.push(null);
	});
}
function CompoundDuplex ( writable , readable , options ) { var self = this ; var convertToZStream = require ( '../index' ) ; // for circular dependencies if ( ! readable || typeof readable . read !== 'function' ) { options = readable ; readable = writable ; writable = null ; } if ( writable && ! writable . _isZStream ) { writable = convertToZStream ( writable ) ; } if ( ! readable . _isZStream ) { readable = convertToZStream ( readable ) ; } if ( ! writable ) { if ( typeof readable . getStreamChain !== 'function' ) { throw new Error ( 'Can only use shorthand CompoundDuplex constructor if pipeline is all zstreams' ) ; } writable = readable . getStreamChain ( ) . getStreams ( ) [ 0 ] ; } if ( ! options ) options = { } ; options . readableObjectMode = readable . isReadableObjectMode ( ) ; options . writableObjectMode = writable . isWritableObjectMode ( ) ; Duplex . call ( this , options ) ; this . _compoundReadable = readable ; this . _compoundWritable = writable ; this . _waitingForReadableData = false ; writable . on ( 'chainerror' , function ( error ) { // Forward the error on; if the chain is to be destructed, the compound stream's _abortStream() method will be called this . ignoreError ( ) ; self . emit ( 'error' , error ) ; } ) ; readable . on ( 'readable' , function ( ) { if ( self . _waitingForReadableData ) { self . _waitingForReadableData = false ; self . _readSomeData ( ) ; } } ) ; readable . on ( 'end' , function ( ) { self . push ( null ) ; } ) ; }
Stream that allows encapsulating a set of streams piped together as a single stream.

@class CompoundDuplex
@constructor
@param {Writable} writable - The first stream in the pipeline to encapsulate.  This is optional.  If all
streams in the pipeline are ZStreams, the readable alone can be used to determine the first stream in the pipeline.
@param {Readable} readable - The last stream in the pipeline.
@param {Object} options - Stream options.
Stream that allows encapsulating a set of streams piped together as a single stream .