diff --git a/doc/api/stream.md b/doc/api/stream.md
index 6510dec5410114..8f8e73fc2ea7e0 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -544,6 +544,15 @@ added: v12.3.0
 
 Getter for the property `objectMode` of a given `Writable` stream.
 
+##### writable.writableReady
+<!-- YAML
+added: REPLACEME
+-->
+
+* {boolean}
+
+Is set to `true` immediately before the [`'ready'`][] event is emitted.
+
 ##### writable.write(chunk[, encoding][, callback])
 <!-- YAML
 added: v0.9.4
@@ -1169,6 +1178,15 @@ added: v12.3.0
 
 Getter for the property `objectMode` of a given `Readable` stream.
 
+##### readable.readableReady
+<!-- YAML
+added: REPLACEME
+-->
+
+* {boolean}
+
+Is set to `true` immediately before the [`'ready'`][] event is emitted.
+
 ##### readable.resume()
 <!-- YAML
 added: v0.9.4
@@ -1622,8 +1640,8 @@ on the type of stream being created, as detailed in the chart below:
 | Use-case | Class | Method(s) to implement |
 | -------- | ----- | ---------------------- |
 | Reading only | [`Readable`] | <code>[_read()][stream-_read]</code> |
-| Writing only | [`Writable`] | <code>[_write()][stream-_write]</code>, <code>[_writev()][stream-_writev]</code>, <code>[_final()][stream-_final]</code> |
-| Reading and writing | [`Duplex`] | <code>[_read()][stream-_read]</code>, <code>[_write()][stream-_write]</code>, <code>[_writev()][stream-_writev]</code>, <code>[_final()][stream-_final]</code> |
+| Writing only | [`Writable`] | <code>[_write()][stream-_write]</code>, <code>[_writev()][stream-_writev]</code>, <code>[_final()][stream-_final]</code>, <code>[_construct()][stream-_construct]</code> |
+| Reading and writing | [`Duplex`] | <code>[_read()][stream-_read]</code>, <code>[_write()][stream-_write]</code>, <code>[_writev()][stream-_writev]</code>, <code>[_final()][stream-_final]</code>, <code>[_construct()][stream-_construct]</code> |
 | Operate on written data, then read the result | [`Transform`] | <code>[_transform()][stream-_transform]</code>, <code>[_flush()][stream-_flush]</code>, <code>[_final()][stream-_final]</code> |
 
 The implementation code for a stream should *never* call the "public" methods
@@ -1645,8 +1663,14 @@ objects and passing appropriate methods as constructor options.
 const { Writable } = require('stream');
 
 const myWritable = new Writable({
+  construct(options, callback) {
+    // Initialize state and load resources...
+  },
   write(chunk, encoding, callback) {
     // ...
+  },
+  destroy() {
+    // Free resources...
   }
 });
 ```
@@ -1700,6 +1724,8 @@ changes:
     [`stream._destroy()`][writable-_destroy] method.
   * `final` {Function} Implementation for the
     [`stream._final()`][stream-_final] method.
+  * `construct` {Function} Implementation for the
+    [`stream._construct()`][stream-_construct] method.
   * `autoDestroy` {boolean} Whether this stream should automatically call
     `.destroy()` on itself after ending. **Default:** `false`.
 
@@ -1745,6 +1771,57 @@ const myWritable = new Writable({
 });
 ```
 
+#### writable.\_construct(options, callback)
+<!-- YAML
+added: REPLACEME
+-->
+
+* `options` {Object} Options passed to constructor.
+* `callback` {Function} Call this function (optionally with an error
+  argument) when finished writing any remaining data.
+
+The `_construct()` method **must not** be called directly. It may be implemented
+by child classes, and if so, will be called by the internal `Writable`
+class methods only.
+
+This optional function will be called by the stream constructor,
+delaying the `'ready'` event until `callback` is called. This is useful to
+initalize state or asynchronously initialize resources before the stream
+can be used.
+
+```js
+const { Writable } = require('stream');
+const fs = require('fs');
+
+class WriteStream extends Writable {
+  constructor(filename) {
+    super({ filename, autoDestroy: true });
+  }
+  _construct({ filename }, callback) {
+    this.filename = filename;
+    this.fd = null;
+    fs.open(this.filename, (fd, err) => {
+      if (err) {
+        callback(err);
+      } else {
+        this.fd = fd;
+        callback();
+      }
+    });
+  }
+  _write(chunk, encoding, callback) {
+    fs.write(this.fd, chunk, callback);
+  }
+  _destroy(err, callback) {
+    if (this.fd) {
+      fs.close(this.fd, (er) => callback(er || err));
+    } else {
+      callback(err);
+    }
+  }
+}
+```
+
 #### writable.\_write(chunk, encoding, callback)
 
 * `chunk` {Buffer|string|any} The `Buffer` to be written, converted from the
@@ -1958,6 +2035,8 @@ changes:
     method.
   * `destroy` {Function} Implementation for the
     [`stream._destroy()`][readable-_destroy] method.
+  * `construct` {Function} Implementation for the
+    [`stream._construct()`][readable-_construct] method.
   * `autoDestroy` {boolean} Whether this stream should automatically call
     `.destroy()` on itself after ending. **Default:** `false`.
 
@@ -2000,6 +2079,64 @@ const myReadable = new Readable({
 });
 ```
 
+#### readable.\_construct(options, callback)
+<!-- YAML
+added: REPLACEME
+-->
+
+* `options` {Object} Options passed to constructor.
+* `callback` {Function} Call this function (optionally with an error
+  argument) when finished writing any remaining data.
+
+The `_construct()` method **must not** be called directly. It may be implemented
+by child classes, and if so, will be called by the internal `Writable`
+class methods only.
+
+This optional function will be called by the stream constructor,
+delaying the `'ready'` event until `callback` is called. This is useful to
+initalize state or asynchronously initialize resources before the stream
+can be used.
+
+```js
+const { Readable } = require('stream');
+const fs = require('fs');
+
+class ReadStream extends Readable {
+  constructor(filename) {
+    super({ autoDestroy: true, filename });
+  }
+  _construct({ filename }, callback) {
+    this.filename = filename;
+    this.fd = null;
+    fs.open(this.filename, (fd, err) => {
+      if (err) {
+        callback(err);
+      } else {
+        this.fd = fd;
+        callback();
+      }
+    });
+  }
+  _read(n) {
+    const buf = Buffer.alloc(n);
+    fs.read(this.fd, buf, 0, n, null, (err, bytesRead) => {
+      if (err) {
+        this.destroy(err);
+      } else {
+        this.push(bytesRead > 0 ? buf.slice(bytesRead) : null);
+      }
+    });
+  }
+  _destroy(err, callback) {
+    if (this.fd) {
+      fs.close(this.fd, (er) => callback(er || err));
+    } else {
+      callback(err);
+    }
+  }
+}
+```
+
 #### readable.\_read(size)
 <!-- YAML
 added: v0.9.4
@@ -2255,6 +2392,38 @@ const myDuplex = new Duplex({
 });
 ```
 
+When using pipeline:
+
+```js
+const { Duplex } = require('stream');
+const fs = require('fs');
+
+stream.pipeline(
+  fs.createReadStream('object.json')
+    .setEncoding('utf-8'),
+  new Duplex({
+    construct(options, callback) {
+      this.data = '';
+      callback();
+    },
+    transform(chunk, encoding, callback) {
+      this.data += chunk;
+      callback();
+    },
+    flush(callback) {
+      try {
+        // Make sure is valid json.
+        JSON.parse(this.data);
+        this.push(this.data);
+      } catch (err) {
+        callback(err);
+      }
+    }
+  }),
+  fs.createWriteStream('valid-object.json')
+);
+```
+
 #### An Example Duplex Stream
 
 The following illustrates a simple example of a `Duplex` stream that wraps a
@@ -2748,6 +2917,7 @@ contain multi-byte characters.
 [object-mode]: #stream_object_mode
 [readable-_destroy]: #stream_readable_destroy_err_callback
 [readable-destroy]: #stream_readable_destroy_error
+[stream-_construct]: #stream_writable_construct_options_callback
 [stream-_final]: #stream_writable_final_callback
 [stream-_flush]: #stream_transform_flush_callback
 [stream-_read]: #stream_readable_read_size_1
diff --git a/lib/_stream_duplex.js b/lib/_stream_duplex.js
index b96a0439f785fa..f5b1eedcdd4459 100644
--- a/lib/_stream_duplex.js
+++ b/lib/_stream_duplex.js
@@ -78,6 +78,16 @@ Object.defineProperty(Duplex.prototype, 'writableHighWaterMark', {
   }
 });
 
+Object.defineProperty(Duplex.prototype, 'writableReady', {
+  // Making it explicit this property is not enumerable
+  // because otherwise some prototype manipulation in
+  // userland will fail
+  enumerable: false,
+  get: function() {
+    return this._writableState && this._writableState.status === 2;
+  }
+});
+
 Object.defineProperty(Duplex.prototype, 'writableBuffer', {
   // Making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index cfa36731e3d661..7f36a66bf368f8 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -103,6 +103,18 @@ function ReadableState(options, stream, isDuplex) {
   this.endEmitted = false;
   this.reading = false;
 
+  // Stream is still being constructed and no operations
+  // can take place until construction finished.
+  this.ready = false;
+
+  // Stream is still being constructed and cannot be
+  // destroyed until construction finished or failed.
+  this.pending = true;
+
+  // Callback to continue destruction after constrution
+  // has finished or failed.
+  this.destroyCallback = null;
+
   // A flag to be able to tell if the event 'readable'/'data' is emitted
   // immediately, or on a later tick.  We set this to true at first, because
   // any actions that shouldn't happen until "later" should generally also
@@ -178,9 +190,22 @@ function Readable(options) {
 
     if (typeof options.destroy === 'function')
       this._destroy = options.destroy;
+
+    if (typeof options.construct === 'function')
+      this._construct = options.construct;
   }
 
   Stream.call(this);
+
+  if (typeof this._construct === 'function') {
+    this.once('ready', function() {
+      maybeReadMore(this, this._readableState);
+    });
+    destroyImpl.construct(this, options);
+  } else {
+    this._readableState.pending = false;
+    this._readableState.ready = true;
+  }
 }
 
 Object.defineProperty(Readable.prototype, 'destroyed', {
@@ -489,9 +514,9 @@ Readable.prototype.read = function(n) {
 
   // 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) {
+  if (state.ended || state.reading || state.pending) {
     doRead = false;
-    debug('reading or ended', doRead);
+    debug('reading, ended or pending', doRead);
   } else if (doRead) {
     debug('do read');
     state.reading = true;
@@ -643,7 +668,7 @@ function maybeReadMore_(stream, state) {
   //   called push() with new data. In this case we skip performing more
   //   read()s. The execution ends in this method again after the _read() ends
   //   up calling push() with more data.
-  while (!state.reading && !state.ended &&
+  while (!state.reading && !state.ended && !state.destroyed &&
          (state.length < state.highWaterMark ||
           (state.flowing && state.length === 0))) {
     const len = state.length;
@@ -1077,6 +1102,12 @@ Readable.prototype[Symbol.asyncIterator] = function() {
   return createReadableStreamAsyncIterator(this);
 };
 
+Object.defineProperty(Readable.prototype, 'readableReady', {
+  get() {
+    return this._readableState.ready;
+  }
+});
+
 Object.defineProperty(Readable.prototype, 'readableHighWaterMark', {
   // Making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in
diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
index e212881c4ac555..85487ae80a1a2a 100644
--- a/lib/_stream_writable.js
+++ b/lib/_stream_writable.js
@@ -142,6 +142,18 @@ function WritableState(options, stream, isDuplex) {
   // this must be 0 before 'finish' can be emitted
   this.pendingcb = 0;
 
+  // Stream is still being constructed and no operations
+  // can take place until construction finished.
+  this.ready = false;
+
+  // Stream is still being constructed and cannot be
+  // destroyed until construction finished or failed.
+  this.pending = true;
+
+  // Callback to continue destruction after constrution
+  // has finished or failed.
+  this.destroyCallback = null;
+
   // Emit prefinish if the only thing we're waiting for is _write cbs
   // This is relevant for synchronous Transform streams
   this.prefinished = false;
@@ -236,9 +248,27 @@ function Writable(options) {
 
     if (typeof options.final === 'function')
       this._final = options.final;
+
+    if (typeof options.construct === 'function')
+      this._construct = options.construct;
   }
 
   Stream.call(this);
+
+  if (typeof this._construct === 'function') {
+    this.once('ready', function() {
+      clearBuffer(this, this._writableState);
+      finishMaybe(this, this._writableState);
+    });
+    destroyImpl.construct(this, options);
+  } else {
+    if (this._readableState) {
+      this._readableState.ready = true;
+      this._readableState.pending = false;
+    }
+    this._writableState.ready = true;
+    this._writableState.pending = false;
+  }
 }
 
 // Otherwise people can pipe Writable streams, which is just wrong.
@@ -353,6 +383,12 @@ function decodeChunk(state, chunk, encoding) {
   return chunk;
 }
 
+Object.defineProperty(Writable.prototype, 'writableReady', {
+  get() {
+    return this._writableState.ready;
+  }
+});
+
 Object.defineProperty(Writable.prototype, 'writableEnded', {
   // Making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in
@@ -394,7 +430,7 @@ function writeOrBuffer(stream, state, isBuf, chunk, encoding, cb) {
   if (!ret)
     state.needDrain = true;
 
-  if (state.writing || state.corked) {
+  if (state.writing || state.corked || state.pending) {
     var last = state.lastBufferedRequest;
     state.lastBufferedRequest = {
       chunk,
@@ -500,6 +536,9 @@ function afterWrite(stream, state, cb) {
 
 // If there's something in the buffer waiting, then process it
 function clearBuffer(stream, state) {
+  if (state.destroyed || state.pending) {
+    return;
+  }
   state.bufferProcessing = true;
   var entry = state.bufferedRequest;
 
@@ -616,7 +655,8 @@ Object.defineProperty(Writable.prototype, 'writableLength', {
 });
 
 function needFinish(state) {
-  return (state.ending &&
+  return (!state.pending &&
+          state.ending &&
           state.length === 0 &&
           state.bufferedRequest === null &&
           !state.finished &&
diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js
index d51fbb1b580113..bb5a8e79e6def4 100644
--- a/lib/internal/fs/streams.js
+++ b/lib/internal/fs/streams.js
@@ -17,6 +17,7 @@ const { Readable, Writable } = require('stream');
 const { toPathIfFileURL } = require('internal/url');
 
 const kMinPoolSpace = 128;
+const kFd = Symbol('kFd');
 
 let pool;
 // It can happen that we expect to read a large chunk of data, and reserve
@@ -51,6 +52,60 @@ function roundUpToMultipleOf8(n) {
   return (n + 7) & ~7;  // Align to 8 byte boundary.
 }
 
+function open(stream, callback) {
+  if (stream[kFd]) {
+    callback(null, stream[kFd]);
+    return;
+  }
+
+  // Backwards compat. Should emit 'open' before 'ready'.
+  stream.once('ready', function() {
+    this.emit('open', this[kFd]);
+  });
+
+  if (typeof stream.open === 'function') {
+    // Backwards compat for monkey patching open().
+
+    // This is to trick legacy implementations to emit
+    // events in correct order (i.e. 'open' before 'ready').
+    // The legacy implementations look like:
+    // fs.open(..., (err) => {
+    //   if (er) {
+    //     ...
+    //   }
+    //   this.fd = fd;
+    //   this.emit('open', fd);
+    //   this.emit('ready'); // This line is not always there.
+    // });
+    const orgEmit = stream.emit;
+    stream.emit = function(ev, err) {
+      if (ev === 'open') {
+        stream.emit = orgEmit;
+        callback();
+        // Ignore any further 'ready' emits.
+        this.removeAllListeners('ready');
+      } else if (ev === 'error') {
+        // We can't register a regular 'error' listener as that
+        // would block the unhandled error behaviour.
+        stream.emit = orgEmit;
+        callback(err);
+      } else {
+        orgEmit.call(stream, arguments);
+      }
+    };
+    stream.open();
+  } else {
+    fs.open(stream.path, stream.flags, stream.mode, (er, fd) => {
+      if (er) {
+        callback(er);
+      } else {
+        stream[kFd] = fd;
+        callback();
+      }
+    });
+  }
+}
+
 function ReadStream(path, options) {
   if (!(this instanceof ReadStream))
     return new ReadStream(path, options);
@@ -65,10 +120,37 @@ function ReadStream(path, options) {
     options.emitClose = false;
   }
 
+  options.autoDestroy = options.autoClose === undefined ?
+    true : options.autoClose;
+  options.path = path;
   Readable.call(this, options);
+}
+Object.setPrototypeOf(ReadStream.prototype, Readable.prototype);
+Object.setPrototypeOf(ReadStream, Readable);
+
+// Backwards compat.
+Object.defineProperty(ReadStream.prototype, 'fd', {
+  get() {
+    return this[kFd];
+  },
+  set(val) {
+    this[kFd] = val;
+  }
+});
 
+// Backwards compat.
+Object.defineProperty(ReadStream.prototype, 'autoClose', {
+  get() {
+    return this._readableState.autoDestroy;
+  },
+  set(val) {
+    this._readableState.autoDestroy = val;
+  }
+});
+
+ReadStream.prototype._construct = function(options, callback) {
   // Path will be ignored when fd is specified, so it can be falsy
-  this.path = toPathIfFileURL(path);
+  this.path = toPathIfFileURL(options.path);
   this.fd = options.fd === undefined ? null : options.fd;
   this.flags = options.flags === undefined ? 'r' : options.flags;
   this.mode = options.mode === undefined ? 0o666 : options.mode;
@@ -100,43 +182,10 @@ function ReadStream(path, options) {
     }
   }
 
-  if (typeof this.fd !== 'number')
-    this.open();
-
-  this.on('end', function() {
-    if (this.autoClose) {
-      this.destroy();
-    }
-  });
-}
-Object.setPrototypeOf(ReadStream.prototype, Readable.prototype);
-Object.setPrototypeOf(ReadStream, Readable);
-
-ReadStream.prototype.open = function() {
-  fs.open(this.path, this.flags, this.mode, (er, fd) => {
-    if (er) {
-      if (this.autoClose) {
-        this.destroy();
-      }
-      this.emit('error', er);
-      return;
-    }
-
-    this.fd = fd;
-    this.emit('open', fd);
-    this.emit('ready');
-    // Start the flow of data.
-    this.read();
-  });
+  open(this, callback);
 };
 
 ReadStream.prototype._read = function(n) {
-  if (typeof this.fd !== 'number') {
-    return this.once('open', function() {
-      this._read(n);
-    });
-  }
-
   if (this.destroyed)
     return;
 
@@ -166,9 +215,10 @@ ReadStream.prototype._read = function(n) {
   fs.read(this.fd, pool, pool.used, toRead, this.pos, (er, bytesRead) => {
     if (er) {
       if (this.autoClose) {
-        this.destroy();
+        this.destroy(er);
+      } else {
+        this.emit('error', er);
       }
-      this.emit('error', er);
     } else {
       let b = null;
       // Now that we know how much data we have actually read, re-wind the
@@ -204,17 +254,18 @@ ReadStream.prototype._read = function(n) {
 };
 
 ReadStream.prototype._destroy = function(err, cb) {
-  if (typeof this.fd !== 'number') {
-    this.once('open', closeFsStream.bind(null, this, cb, err));
+  if (!this[kFd]) {
+    cb(err);
+    if (!err)
+      this.emit('close');
     return;
   }
-
   closeFsStream(this, cb, err);
-  this.fd = null;
+  this[kFd] = null;
 };
 
 function closeFsStream(stream, cb, err) {
-  fs.close(stream.fd, (er) => {
+  fs.close(stream[kFd], (er) => {
     er = er || err;
     cb(er);
     stream.closed = true;
@@ -227,6 +278,7 @@ ReadStream.prototype.close = function(cb) {
   this.destroy(null, cb);
 };
 
+// Backwards compat.
 Object.defineProperty(ReadStream.prototype, 'pending', {
   get() { return this.fd === null; },
   configurable: true
@@ -243,10 +295,39 @@ function WriteStream(path, options) {
     options.emitClose = false;
   }
 
+  options.autoDestroy = options.autoClose === undefined ?
+    true : options.autoClose;
+  options.path = path;
   Writable.call(this, options);
+}
+Object.setPrototypeOf(WriteStream.prototype, Writable.prototype);
+Object.setPrototypeOf(WriteStream, Writable);
+
+// Backwards compat.
+Object.defineProperty(WriteStream.prototype, 'fd', {
+  get() {
+    // Backwards compat. Expects fd to be cleared before 'finish' and not
+    // through _destroy.
+    return this.autoClose && this._writableState.finished ? null : this[kFd];
+  },
+  set(val) {
+    this[kFd] = val;
+  }
+});
 
+// Backwards compat.
+Object.defineProperty(ReadStream.prototype, 'autoClose', {
+  get() {
+    return this._writableState.autoDestroy;
+  },
+  set(val) {
+    this._writableState.autoDestroy = val;
+  }
+});
+
+WriteStream.prototype._construct = function(options, callback) {
   // Path will be ignored when fd is specified, so it can be falsy
-  this.path = toPathIfFileURL(path);
+  this.path = toPathIfFileURL(options.path);
   this.fd = options.fd === undefined ? null : options.fd;
   this.flags = options.flags === undefined ? 'w' : options.flags;
   this.mode = options.mode === undefined ? 0o666 : options.mode;
@@ -266,54 +347,17 @@ function WriteStream(path, options) {
   if (options.encoding)
     this.setDefaultEncoding(options.encoding);
 
-  if (typeof this.fd !== 'number')
-    this.open();
-}
-Object.setPrototypeOf(WriteStream.prototype, Writable.prototype);
-Object.setPrototypeOf(WriteStream, Writable);
-
-WriteStream.prototype._final = function(callback) {
-  if (this.autoClose) {
-    this.destroy();
-  }
-
-  callback();
+  open(this, callback);
 };
 
-WriteStream.prototype.open = function() {
-  fs.open(this.path, this.flags, this.mode, (er, fd) => {
-    if (er) {
-      if (this.autoClose) {
-        this.destroy();
-      }
-      this.emit('error', er);
-      return;
-    }
-
-    this.fd = fd;
-    this.emit('open', fd);
-    this.emit('ready');
-  });
-};
-
-
 WriteStream.prototype._write = function(data, encoding, cb) {
   if (!(data instanceof Buffer)) {
     const err = new ERR_INVALID_ARG_TYPE('data', 'Buffer', data);
     return this.emit('error', err);
   }
 
-  if (typeof this.fd !== 'number') {
-    return this.once('open', function() {
-      this._write(data, encoding, cb);
-    });
-  }
-
   fs.write(this.fd, data, 0, data.length, this.pos, (er, bytes) => {
     if (er) {
-      if (this.autoClose) {
-        this.destroy();
-      }
       return cb(er);
     }
     this.bytesWritten += bytes;
@@ -326,12 +370,6 @@ WriteStream.prototype._write = function(data, encoding, cb) {
 
 
 WriteStream.prototype._writev = function(data, cb) {
-  if (typeof this.fd !== 'number') {
-    return this.once('open', function() {
-      this._writev(data, cb);
-    });
-  }
-
   const self = this;
   const len = data.length;
   const chunks = new Array(len);
@@ -346,7 +384,6 @@ WriteStream.prototype._writev = function(data, cb) {
 
   fs.writev(this.fd, chunks, this.pos, function(er, bytes) {
     if (er) {
-      self.destroy();
       return cb(er);
     }
     self.bytesWritten += bytes;
@@ -383,6 +420,7 @@ WriteStream.prototype.close = function(cb) {
 // There is no shutdown() for files.
 WriteStream.prototype.destroySoon = WriteStream.prototype.end;
 
+// Backwards compat.
 Object.defineProperty(WriteStream.prototype, 'pending', {
   get() { return this.fd === null; },
   configurable: true
diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index 27985482cee70b..1e15a89ef46c7c 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -26,6 +26,7 @@ function needError(stream, err) {
 function destroy(err, cb) {
   const r = this._readableState;
   const w = this._writableState;
+  const s = w || r;
 
   if ((w && w.destroyed) || (r && r.destroyed)) {
     if (cb) {
@@ -47,21 +48,42 @@ function destroy(err, cb) {
     r.destroyed = true;
   }
 
-  this._destroy(err || null, (err) => {
+  // Check if stream has finished initializing before destroying it.
+  // If still pending then register a `destroyCallback` which will be invoked
+  // once initialized.
+  if (s.pending) {
+    if (err || cb) {
+      s.destroyCallback = (stream, er) => {
+        _destroy(stream, err || er, cb);
+      };
+    } else {
+      // Fast path. Avoid allocating closure.
+      s.destroyCallback = _destroy;
+    }
+  } else {
+    _destroy(this, err, cb);
+  }
+
+  return this;
+}
+
+function _destroy(self, err, cb) {
+  self._destroy(err || null, (err) => {
+    const r = self._readableState;
+    const w = self._writableState;
+
     const emitClose = (w && w.emitClose) || (r && r.emitClose);
     if (cb) {
       if (emitClose) {
-        process.nextTick(emitCloseNT, this);
+        process.nextTick(emitCloseNT, self);
       }
       cb(err);
-    } else if (needError(this, err)) {
-      process.nextTick(emitClose ? emitErrorCloseNT : emitErrorNT, this, err);
+    } else if (needError(self, err)) {
+      process.nextTick(emitClose ? emitErrorCloseNT : emitErrorNT, self, err);
     } else if (emitClose) {
-      process.nextTick(emitCloseNT, this);
+      process.nextTick(emitCloseNT, self);
     }
   });
-
-  return this;
 }
 
 function emitErrorCloseNT(self, err) {
@@ -116,8 +138,47 @@ function errorOrDestroy(stream, err) {
     stream.emit('error', err);
 }
 
+function construct_(stream, err) {
+  const r = stream._readableState;
+  const w = stream._writableState;
+  const s = w || r;
+
+  if (r) {
+    r.pending = false;
+  }
+  if (w) {
+    w.pending = false;
+  }
+
+  if (s.destroyCallback) {
+    s.destroyCallback(stream, err);
+  } else if (err) {
+    errorOrDestroy(stream, err);
+  } else if (!s.destroyed) {
+    if (r) {
+      r.ready = true;
+    }
+    if (w) {
+      w.ready = true;
+    }
+    stream.emit('ready');
+  }
+}
+
+function construct(stream, options) {
+  let sync = true;
+  stream._construct(options, (err) => {
+    if (sync) {
+      process.nextTick(construct_, stream, err);
+    } else {
+      construct_(stream, err);
+    }
+  });
+  sync = false;
+}
 
 module.exports = {
+  construct,
   destroy,
   undestroy,
   errorOrDestroy
diff --git a/test/parallel/test-file-write-stream.js b/test/parallel/test-file-write-stream.js
index d66657d690f1b2..4ba0eecbee62c4 100644
--- a/test/parallel/test-file-write-stream.js
+++ b/test/parallel/test-file-write-stream.js
@@ -46,9 +46,6 @@ file
     callbacks.open++;
     assert.strictEqual(typeof fd, 'number');
   })
-  .on('error', function(err) {
-    throw err;
-  })
   .on('drain', function() {
     console.error('drain!', callbacks.drain);
     callbacks.drain++;
@@ -65,17 +62,16 @@ file
     assert.strictEqual(file.bytesWritten, EXPECTED.length * 2);
 
     callbacks.close++;
-    common.expectsError(
-      () => {
-        console.error('write after end should not be allowed');
-        file.write('should not work anymore');
-      },
-      {
-        code: 'ERR_STREAM_WRITE_AFTER_END',
-        type: Error,
-        message: 'write after end'
-      }
-    );
+    file.on('error', common.expectsError({
+      code: 'ERR_STREAM_WRITE_AFTER_END',
+      type: Error,
+      message: 'write after end'
+    }));
+    file.write('should not work anymore', common.expectsError({
+      code: 'ERR_STREAM_WRITE_AFTER_END',
+      type: Error,
+      message: 'write after end'
+    }));
 
     fs.unlinkSync(fn);
   });
diff --git a/test/parallel/test-fs-stream-construct.js b/test/parallel/test-fs-stream-construct.js
new file mode 100644
index 00000000000000..c49fc273e1904a
--- /dev/null
+++ b/test/parallel/test-fs-stream-construct.js
@@ -0,0 +1,231 @@
+'use strict';
+
+const common = require('../common');
+const fs = require('fs');
+const assert = require('assert');
+const path = require('path');
+const tmpdir = require('../common/tmpdir');
+const fixtures = require('../common/fixtures');
+
+const examplePath = fixtures.path('x.txt');
+const dummyPath = path.join(tmpdir.path, 'x.txt');
+
+tmpdir.refresh();
+
+{
+  // Compat with old node.
+
+  function ReadStream(...args) {
+    fs.ReadStream.call(this, ...args);
+  }
+  Object.setPrototypeOf(ReadStream.prototype, fs.ReadStream.prototype);
+  Object.setPrototypeOf(ReadStream, fs.ReadStream);
+
+  ReadStream.prototype.open = common.mustCall(function() {
+    fs.open(this.path, this.flags, this.mode, (er, fd) => {
+      if (er) {
+        if (this.autoClose) {
+          this.destroy();
+        }
+        this.emit('error', er);
+        return;
+      }
+
+      this.fd = fd;
+      this.emit('open', fd);
+      this.emit('ready');
+    });
+  });
+
+  let readyCalled = false;
+  let ticked = false;
+  const r = new ReadStream(examplePath)
+    .on('ready', common.mustCall(() => {
+      readyCalled = true;
+      assert.strictEqual(r.readableReady, true);
+      // Make sure 'ready' is emitted in same tick as 'open'.
+      assert.strictEqual(ticked, false);
+    }))
+    .on('open', common.mustCall((fd) => {
+      process.nextTick(() => {
+        ticked = true;
+      });
+      assert.strictEqual(readyCalled, false);
+      assert.strictEqual(fd, r.fd);
+    }));
+  assert.strictEqual(r.readableReady, false);
+}
+
+{
+  // Compat with old node.
+
+  function WriteStream(...args) {
+    fs.WriteStream.call(this, ...args);
+  }
+  Object.setPrototypeOf(WriteStream.prototype, fs.WriteStream.prototype);
+  Object.setPrototypeOf(WriteStream, fs.WriteStream);
+
+  WriteStream.prototype.open = common.mustCall(function() {
+    fs.open(this.path, this.flags, this.mode, (er, fd) => {
+      if (er) {
+        if (this.autoClose) {
+          this.destroy();
+        }
+        this.emit('error', er);
+        return;
+      }
+
+      this.fd = fd;
+      this.emit('open', fd);
+      this.emit('ready');
+    });
+  });
+
+  let readyCalled = false;
+  let ticked = false;
+  const w = new WriteStream(dummyPath)
+    .on('ready', common.mustCall(() => {
+      readyCalled = true;
+      assert.strictEqual(w.writableReady, true);
+      // Make sure 'ready' is emitted in same tick as 'open'.
+      assert.strictEqual(ticked, false);
+    }))
+    .on('open', common.mustCall((fd) => {
+      process.nextTick(() => {
+        ticked = true;
+      });
+      assert.strictEqual(readyCalled, false);
+      assert.strictEqual(fd, w.fd);
+    }));
+  assert.strictEqual(w.writableReady, false);
+}
+
+{
+  // Compat with graceful-fs.
+
+  function ReadStream(...args) {
+    fs.ReadStream.call(this, ...args);
+  }
+  Object.setPrototypeOf(ReadStream.prototype, fs.ReadStream.prototype);
+  Object.setPrototypeOf(ReadStream, fs.ReadStream);
+
+  ReadStream.prototype.open = common.mustCall(function ReadStream$open() {
+    const that = this;
+    fs.open(that.path, that.flags, that.mode, (err, fd) => {
+      if (err) {
+        if (that.autoClose)
+          that.destroy();
+
+        that.emit('error', err);
+      } else {
+        that.fd = fd;
+        that.emit('open', fd);
+        that.read();
+      }
+    });
+  });
+
+  let readyCalled = false;
+  let ticked = false;
+  const r = new ReadStream(examplePath)
+    .on('ready', common.mustCall(() => {
+      readyCalled = true;
+      assert.strictEqual(r.readableReady, true);
+      // Make sure 'ready' is emitted in same tick as 'open'.
+      assert.strictEqual(ticked, false);
+    }))
+    .on('open', common.mustCall((fd) => {
+      process.nextTick(() => {
+        ticked = true;
+      });
+      assert.strictEqual(readyCalled, false);
+      assert.strictEqual(fd, r.fd);
+    }));
+}
+
+
+{
+  // Compat with graceful-fs.
+
+  function WriteStream(...args) {
+    fs.WriteStream.call(this, ...args);
+  }
+  Object.setPrototypeOf(WriteStream.prototype, fs.WriteStream.prototype);
+  Object.setPrototypeOf(WriteStream, fs.WriteStream);
+
+  WriteStream.prototype.open = common.mustCall(function WriteStream$open() {
+    const that = this;
+    fs.open(that.path, that.flags, that.mode, function (err, fd) {
+      if (err) {
+        that.destroy();
+        that.emit('error', err);
+      } else {
+        that.fd = fd
+        that.emit('open', fd);
+      }
+    });
+  });
+
+  let readyCalled = false;
+  let ticked = false;
+  const w = new WriteStream(dummyPath)
+    .on('ready', common.mustCall(() => {
+      readyCalled = true;
+      assert.strictEqual(w.writableReady, true);
+      // Make sure 'ready' is emitted in same tick as 'open'.
+      assert.strictEqual(ticked, false);
+    }))
+    .on('open', common.mustCall((fd) => {
+      process.nextTick(() => {
+        ticked = true;
+      });
+      assert.strictEqual(readyCalled, false);
+      assert.strictEqual(fd, w.fd);
+    }));
+}
+
+{
+  // Compat error.
+
+  function ReadStream(...args) {
+    fs.ReadStream.call(this, ...args);
+  }
+  Object.setPrototypeOf(ReadStream.prototype, fs.ReadStream.prototype);
+  Object.setPrototypeOf(ReadStream, fs.ReadStream);
+
+  ReadStream.prototype.open = common.mustCall(function ReadStream$open() {
+    const that = this;
+    fs.open(that.path, that.flags, that.mode, (err, fd) => {
+      that.emit('error', new Error('test'));
+    });
+  });
+
+  new ReadStream(examplePath)
+    .on('error', common.expectsError({
+      type: Error,
+      message: 'test'
+    }));
+}
+
+{
+  // Compat error.
+
+  function WriteStream(...args) {
+    fs.WriteStream.call(this, ...args);
+  }
+  Object.setPrototypeOf(WriteStream.prototype, fs.WriteStream.prototype);
+  Object.setPrototypeOf(WriteStream, fs.WriteStream);
+
+  WriteStream.prototype.open = common.mustCall(function WriteStream$open() {
+    const that = this;
+    fs.open(that.path, that.flags, that.mode, (err, fd) => {
+      that.emit('error', new Error('test'));
+    });
+  });
+
+  new WriteStream(examplePath)
+    .on('error', common.expectsError({
+      type: Error,
+      message: 'test'
+    }));
+}
diff --git a/test/parallel/test-fs-stream-double-close.js b/test/parallel/test-fs-stream-double-close.js
index 8c0037b24312ff..1e37f5f2a35ebd 100644
--- a/test/parallel/test-fs-stream-double-close.js
+++ b/test/parallel/test-fs-stream-double-close.js
@@ -40,10 +40,9 @@ function test1(stream) {
 }
 
 function test2(stream) {
+  // Don't emit 'open' after destroy.
   stream.destroy();
-  stream.on('open', common.mustCall(function(fd) {
-    stream.destroy();
-  }));
+  stream.on('open', common.mustNotCall());
 }
 
 function test3(stream) {
diff --git a/test/parallel/test-stream-construct.js b/test/parallel/test-stream-construct.js
new file mode 100644
index 00000000000000..fca3f26204ac22
--- /dev/null
+++ b/test/parallel/test-stream-construct.js
@@ -0,0 +1,211 @@
+'use strict';
+
+const common = require('../common');
+const { Writable, Readable } = require('stream');
+const assert = require('assert');
+
+{
+  // Pass options as is to construct.
+
+  const options2 = {
+    test: 'foo',
+    construct(options, callback) {
+      assert.strictEqual(options, options2);
+      callback();
+    }
+  };
+  const w = new Writable(options2);
+  w.on('ready', common.mustCall());
+}
+
+{
+  // Pass options as is to construct.
+
+  const options2 = {
+    test: 'foo',
+    construct(options, callback) {
+      assert.strictEqual(options, options2);
+      callback();
+    },
+    read() {}
+  };
+  const r = new Readable(options2);
+  r.on('ready', common.mustCall());
+}
+
+{
+  // Emit ready in next tick when sync construct.
+
+  const w = new Writable({
+    construct(options, callback) {
+      callback();
+    }
+  });
+  assert.strictEqual(w.writableReady, false);
+  w.on('ready', common.mustCall(() => {
+    assert.strictEqual(w.writableReady, true);
+  }));
+}
+
+{
+  // Emit ready in next tick when sync construct.
+
+  const r = new Readable({
+    construct(options, callback) {
+      callback();
+    },
+    read() {}
+  });
+  assert.strictEqual(r.readableReady, false);
+  r.on('ready', common.mustCall(() => {
+    assert.strictEqual(r.readableReady, true);
+  }));
+}
+
+{
+  // Emit ready on async construct.
+
+  const w = new Writable({
+    construct(options, callback) {
+      process.nextTick(callback);
+    }
+  });
+  assert.strictEqual(w.writableReady, false);
+  w.on('ready', common.mustCall(() => {
+    assert.strictEqual(w.writableReady, true);
+  }));
+}
+
+{
+  // Emit ready on async construct.
+
+  const r = new Readable({
+    construct(options, callback) {
+      process.nextTick(callback);
+    },
+    read() {}
+  });
+  assert.strictEqual(r.readableReady, false);
+  r.on('ready', common.mustCall(() => {
+    assert.strictEqual(r.readableReady, true);
+  }));
+}
+
+{
+  // No construct.
+
+  const w = new Writable({
+  });
+  assert.strictEqual(w.writableReady, true);
+  w.on('ready', common.mustNotCall());
+}
+
+{
+  // No construct.
+
+  const r = new Readable({
+    read() {}
+  });
+  assert.strictEqual(r.readableReady, true);
+  r.on('ready', common.mustNotCall());
+}
+
+{
+  // Don't emit ready on destroy.
+
+  const w = new Writable({
+    construct(options, callback) {
+      callback();
+    }
+  });
+  w.on('ready', common.mustNotCall());
+  w.destroy();
+}
+
+{
+  // Don't emit ready on destroy.
+
+  const r = new Readable({
+    construct(options, callback) {
+      callback();
+    },
+    read() {}
+  });
+  r.on('ready', common.mustNotCall());
+  r.destroy();
+}
+
+{
+  // Don't emit ready on destroy.
+
+  let pending = true;
+  const w = new Writable({
+    construct(options, callback) {
+      process.nextTick(() => {
+        pending = false;
+        callback();
+      });
+    },
+    destroy() {
+      // Don't destroy before constructed.
+      assert.strictEqual(pending, false);
+      // Not ready if destroy()ed before ready.
+      assert.strictEqual(w.writableReady, false);
+    }
+  });
+  w.on('ready', common.mustNotCall());
+  w.destroy();
+}
+
+{
+  // Don't emit ready on destroy.
+
+  let pending = true;
+  const r = new Readable({
+    construct(options, callback) {
+      process.nextTick(() => {
+        pending = false;
+        callback();
+      });
+    },
+    destroy() {
+      // Don't destroy before constructed.
+      assert.strictEqual(pending, false);
+      // Not ready if destroy()ed before ready.
+      assert.strictEqual(r.readableReady, false);
+    },
+    read() {}
+  });
+  r.on('ready', common.mustNotCall());
+  r.destroy();
+}
+
+{
+  // Throw synchronously
+
+  common.expectsError(function() {
+    new Writable({
+      construct(options, callback) {
+        throw new Error('test');
+      }
+    });
+  }, {
+    type: Error,
+    message: 'test'
+  });
+}
+
+{
+  // Throw synchronously
+
+  common.expectsError(function() {
+    new Readable({
+      construct(options, callback) {
+        throw new Error('test');
+      }
+    });
+  }, {
+    type: Error,
+    message: 'test'
+  });
+}