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);
  });
}
Constructor