src/index.js
function Duplexer(options, writableStream, readableStream) {
  const _this = this;

  // Ensure new were used
  if (!(this instanceof Duplexer)) {
    return new (Duplexer.bind.apply(
      Duplexer, // eslint-disable-line
      [Duplexer].concat([].slice.call(arguments, 0))
    ))();
  }

  // Mapping args
  if (isStream(options)) {
    readableStream = writableStream;
    writableStream = options;
    options = {};
  } else {
    options = options || {};
  }
  this._reemitErrors =
    'boolean' === typeof options.reemitErrors ? options.reemitErrors : true;
  delete options.reemitErrors;

  // Checking arguments
  if (!isStream(writableStream, 'Writable', 'Duplex')) {
    throw new Error(
      'The writable stream must be an instanceof Writable or Duplex.'
    );
  }
  if (!isStream(readableStream, 'Readable')) {
    throw new Error('The readable stream must be an instanceof Readable.');
  }

  // Parent constructor
  Stream.Duplex.call(this, options);

  // Save streams refs
  this._writable = writableStream;
  this._readable = readableStream;

  // Internal state
  this._waitDatas = false;
  this._hasDatas = false;

  if ('undefined' == typeof this._readable._readableState) {
    this._readable = new Stream.Readable({
      objectMode: options.objectMode || false,
    }).wrap(this._readable);
  }

  if (this._reemitErrors) {
    this._writable.on('error', err => {
      _this.emit('error', err);
    });
    this._readable.on('error', err => {
      _this.emit('error', err);
    });
  }

  this._writable.on('drain', () => {
    _this.emit('drain');
  });

  this.once('finish', () => {
    _this._writable.end();
  });

  this._writable.once('finish', () => {
    _this.end();
  });

  this._readable.on('readable', () => {
    _this._hasDatas = true;
    if (_this._waitDatas) {
      _this._pushAll();
    }
  });

  this._readable.once('end', () => {
    _this.push(null);
  });
}
function Duplexer(options, writableStream, readableStream) {
  const _this = this;

  // Ensure new were used
  if (!(this instanceof Duplexer)) {
    return new (Duplexer.bind.apply(
      Duplexer, // eslint-disable-line
      [Duplexer].concat([].slice.call(arguments, 0))
    ))();
  }

  // Mapping args
  if (isStream(options)) {
    readableStream = writableStream;
    writableStream = options;
    options = {};
  } else {
    options = options || {};
  }
  this._reemitErrors =
    'boolean' === typeof options.reemitErrors ? options.reemitErrors : true;
  delete options.reemitErrors;

  // Checking arguments
  if (!isStream(writableStream, 'Writable', 'Duplex')) {
    throw new Error(
      'The writable stream must be an instanceof Writable or Duplex.'
    );
  }
  if (!isStream(readableStream, 'Readable')) {
    throw new Error('The readable stream must be an instanceof Readable.');
  }

  // Parent constructor
  Stream.Duplex.call(this, options);

  // Save streams refs
  this._writable = writableStream;
  this._readable = readableStream;

  // Internal state
  this._waitDatas = false;
  this._hasDatas = false;

  if ('undefined' == typeof this._readable._readableState) {
    this._readable = new Stream.Readable({
      objectMode: options.objectMode || false,
    }).wrap(this._readable);
  }

  if (this._reemitErrors) {
    this._writable.on('error', err => {
      _this.emit('error', err);
    });
    this._readable.on('error', err => {
      _this.emit('error', err);
    });
  }

  this._writable.on('drain', () => {
    _this.emit('drain');
  });

  this.once('finish', () => {
    _this._writable.end();
  });

  this._writable.once('finish', () => {
    _this.end();
  });

  this._readable.on('readable', () => {
    _this._hasDatas = true;
    if (_this._waitDatas) {
      _this._pushAll();
    }
  });

  this._readable.once('end', () => {
    _this.push(null);
  });
}
function Duplexer ( options , writableStream , readableStream ) { const _this = this ; // Ensure new were used if ( ! ( this instanceof Duplexer ) ) { return new ( Duplexer . bind . apply ( Duplexer , // eslint-disable-line [ Duplexer ] . concat ( [ ] . slice . call ( arguments , 0 ) ) ) ) ( ) ; } // Mapping args if ( isStream ( options ) ) { readableStream = writableStream ; writableStream = options ; options = { } ; } else { options = options || { } ; } this . _reemitErrors = 'boolean' === typeof options . reemitErrors ? options . reemitErrors : true ; delete options . reemitErrors ; // Checking arguments if ( ! isStream ( writableStream , 'Writable' , 'Duplex' ) ) { throw new Error ( 'The writable stream must be an instanceof Writable or Duplex.' ) ; } if ( ! isStream ( readableStream , 'Readable' ) ) { throw new Error ( 'The readable stream must be an instanceof Readable.' ) ; } // Parent constructor Stream . Duplex . call ( this , options ) ; // Save streams refs this . _writable = writableStream ; this . _readable = readableStream ; // Internal state this . _waitDatas = false ; this . _hasDatas = false ; if ( 'undefined' == typeof this . _readable . _readableState ) { this . _readable = new Stream . Readable ( { objectMode : options . objectMode || false , } ) . wrap ( this . _readable ) ; } if ( this . _reemitErrors ) { this . _writable . on ( 'error' , err => { _this . emit ( 'error' , err ) ; } ) ; this . _readable . on ( 'error' , err => { _this . emit ( 'error' , err ) ; } ) ; } this . _writable . on ( 'drain' , ( ) => { _this . emit ( 'drain' ) ; } ) ; this . once ( 'finish' , ( ) => { _this . _writable . end ( ) ; } ) ; this . _writable . once ( 'finish' , ( ) => { _this . end ( ) ; } ) ; this . _readable . on ( 'readable' , ( ) => { _this . _hasDatas = true ; if ( _this . _waitDatas ) { _this . _pushAll ( ) ; } } ) ; this . _readable . once ( 'end' , ( ) => { _this . push ( null ) ; } ) ; }
Constructor
Constructor