diff --git a/doc/api/stream.md b/doc/api/stream.md
index 6dda48f9a25d64..1c9a40bc9fca70 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -550,8 +550,7 @@ added: v9.3.0
 
 * {number}
 
-Return the value of `highWaterMark` passed when constructing this
-`Writable`.
+Return the value of `highWaterMark` passed when creating this `Writable`.
 
 ##### `writable.writableLength`
 <!-- YAML
@@ -1193,8 +1192,7 @@ added: v9.3.0
 
 * {number}
 
-Returns the value of `highWaterMark` passed when constructing this
-`Readable`.
+Returns the value of `highWaterMark` passed when creating this `Readable`.
 
 ##### `readable.readableLength`
 <!-- YAML
@@ -1792,7 +1790,7 @@ expectations.
 added: v1.2.0
 -->
 
-For many simple cases, it is possible to construct a stream without relying on
+For many simple cases, it is possible to create a stream without relying on
 inheritance. This can be accomplished by directly creating instances of the
 `stream.Writable`, `stream.Readable`, `stream.Duplex` or `stream.Transform`
 objects and passing appropriate methods as constructor options.
@@ -1801,8 +1799,14 @@ objects and passing appropriate methods as constructor options.
 const { Writable } = require('stream');
 
 const myWritable = new Writable({
+  construct(callback) {
+    // Initialize state and load resources...
+  },
   write(chunk, encoding, callback) {
     // ...
+  },
+  destroy() {
+    // Free resources...
   }
 });
 ```
@@ -1861,6 +1865,8 @@ changes:
     [`stream._destroy()`][writable-_destroy] method.
   * `final` {Function} Implementation for the
     [`stream._final()`][stream-_final] method.
+  * `construct` {Function} Implementation for the
+    [`stream._construct()`][writable-_construct] method.
   * `autoDestroy` {boolean} Whether this stream should automatically call
     `.destroy()` on itself after ending. **Default:** `true`.
 
@@ -1906,6 +1912,56 @@ const myWritable = new Writable({
 });
 ```
 
+#### `writable._construct(callback)`
+<!-- YAML
+added: REPLACEME
+-->
+
+* `callback` {Function} Call this function (optionally with an error
+  argument) when the stream has finished initializing.
+
+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 in a tick after the stream constructor
+has returned, delaying any `_write`, `_final` and `_destroy` calls until
+`callback` is called. This is useful to initialize 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();
+    this.filename = filename;
+    this.fd = fd;
+  }
+  _construct(callback) {
+    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)`
 <!-- YAML
 changes:
@@ -2130,6 +2186,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:** `true`.
 
@@ -2172,6 +2230,63 @@ const myReadable = new Readable({
 });
 ```
 
+#### `readable._construct(callback)`
+<!-- YAML
+added: REPLACEME
+-->
+
+* `callback` {Function} Call this function (optionally with an error
+  argument) when the stream has finished initializing.
+
+The `_construct()` method MUST NOT be called directly. It may be implemented
+by child classes, and if so, will be called by the internal `Readable`
+class methods only.
+
+This optional function will be scheduled in the next tick by the stream
+constructor, delaying any `_read` and `_destroy` calls until `callback` is
+called. This is useful to initialize 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();
+    this.filename = filename;
+    this.fd = null;
+  }
+  _construct(callback) {
+    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(0, 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
@@ -2427,6 +2542,46 @@ const myDuplex = new Duplex({
 });
 ```
 
+When using pipeline:
+
+```js
+const { Transform, pipeline } = require('stream');
+const fs = require('fs');
+
+pipeline(
+  fs.createReadStream('object.json')
+    .setEncoding('utf-8'),
+  new Transform({
+    decodeStrings: false, // Accept string input rather than Buffers
+    construct(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'),
+  (err) => {
+    if (err) {
+      console.error('failed', err);
+    } else {
+      console.log('completed');
+    }
+  }
+);
+```
+
 #### An Example Duplex Stream
 
 The following illustrates a simple example of a `Duplex` stream that wraps a
@@ -2706,8 +2861,8 @@ unhandled post-destroy errors.
 
 #### Creating Readable Streams with Async Generators
 
-We can construct a Node.js Readable Stream from an asynchronous generator
-using the `Readable.from()` utility method:
+A Node.js Readable Stream can be created from an asynchronous generator using
+the `Readable.from()` utility method:
 
 ```js
 const { Readable } = require('stream');
@@ -2960,6 +3115,7 @@ contain multi-byte characters.
 [http-incoming-message]: http.html#http_class_http_incomingmessage
 [hwm-gotcha]: #stream_highwatermark_discrepancy_after_calling_readable_setencoding
 [object-mode]: #stream_object_mode
+[readable-_construct]: #stream_readable_construct_callback
 [readable-_destroy]: #stream_readable_destroy_err_callback
 [readable-destroy]: #stream_readable_destroy_error
 [stream-_final]: #stream_writable_final_callback
@@ -2976,6 +3132,7 @@ contain multi-byte characters.
 [stream-uncork]: #stream_writable_uncork
 [stream-write]: #stream_writable_write_chunk_encoding_callback
 [Stream Three States]: #stream_three_states
+[writable-_construct]: #stream_writable_construct_callback
 [writable-_destroy]: #stream_writable_destroy_err_callback
 [writable-destroy]: #stream_writable_destroy_error
 [writable-new]: #stream_constructor_new_stream_writable_options
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 166d3bb4404c7d..f978113bb3b43c 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -118,6 +118,12 @@ function ReadableState(options, stream, isDuplex) {
   this.endEmitted = false;
   this.reading = false;
 
+  // Stream is still being constructed and cannot be
+  // destroyed until construction finished or failed.
+  // Async construction is opt in, therefore we start as
+  // constructed.
+  this.constructed = true;
+
   // 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
@@ -197,9 +203,16 @@ function Readable(options) {
 
     if (typeof options.destroy === 'function')
       this._destroy = options.destroy;
+
+    if (typeof options.construct === 'function')
+      this._construct = options.construct;
   }
 
   Stream.call(this, options);
+
+  destroyImpl.construct(this, () => {
+    maybeReadMore(this, this._readableState);
+  });
 }
 
 Readable.prototype.destroy = destroyImpl.destroy;
@@ -461,11 +474,12 @@ Readable.prototype.read = function(n) {
   }
 
   // However, if we've ended, then there's no point, if we're already
-  // reading, then it's unnecessary, and if we're destroyed or errored,
-  // then it's not allowed.
-  if (state.ended || state.reading || state.destroyed || state.errored) {
+  // reading, then it's unnecessary, if we're constructing we have to wait,
+  // and if we're destroyed or errored, then it's not allowed,
+  if (state.ended || state.reading || state.destroyed || state.errored ||
+      !state.constructed) {
     doRead = false;
-    debug('reading or ended', doRead);
+    debug('reading, ended or constructing', doRead);
   } else if (doRead) {
     debug('do read');
     state.reading = true;
@@ -587,7 +601,7 @@ function emitReadable_(stream) {
 // 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) {
+  if (!state.readingMore && state.constructed) {
     state.readingMore = true;
     process.nextTick(maybeReadMore_, stream, state);
   }
diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
index dfbf4f6ff48203..916e9b87d9c17a 100644
--- a/lib/_stream_writable.js
+++ b/lib/_stream_writable.js
@@ -155,6 +155,12 @@ function WritableState(options, stream, isDuplex) {
   // this must be 0 before 'finish' can be emitted.
   this.pendingcb = 0;
 
+  // Stream is still being constructed and cannot be
+  // destroyed until construction finished or failed.
+  // Async construction is opt in, therefore we start as
+  // constructed.
+  this.constructed = true;
+
   // Emit prefinish if the only thing we're waiting for is _write cbs
   // This is relevant for synchronous Transform streams.
   this.prefinished = false;
@@ -249,9 +255,22 @@ function Writable(options) {
 
     if (typeof options.final === 'function')
       this._final = options.final;
+
+    if (typeof options.construct === 'function')
+      this._construct = options.construct;
   }
 
   Stream.call(this, options);
+
+  destroyImpl.construct(this, () => {
+    const state = this._writableState;
+
+    if (!state.writing) {
+      clearBuffer(this, state);
+    }
+
+    finishMaybe(this, state);
+  });
 }
 
 // Otherwise people can pipe Writable streams, which is just wrong.
@@ -342,7 +361,7 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
 
   state.length += len;
 
-  if (state.writing || state.corked || state.errored) {
+  if (state.writing || state.corked || state.errored || !state.constructed) {
     state.buffered.push({ chunk, encoding, callback });
     if (state.allBuffers && encoding !== 'buffer') {
       state.allBuffers = false;
@@ -492,7 +511,10 @@ function errorBuffer(state, err) {
 
 // If there's something in the buffer waiting, then process it.
 function clearBuffer(stream, state) {
-  if (state.corked || state.bufferProcessing || state.destroyed) {
+  if (state.corked ||
+      state.bufferProcessing ||
+      state.destroyed ||
+      !state.constructed) {
     return;
   }
 
@@ -600,6 +622,7 @@ Writable.prototype.end = function(chunk, encoding, cb) {
 
 function needFinish(state) {
   return (state.ending &&
+          state.constructed &&
           state.length === 0 &&
           !state.errored &&
           state.buffered.length === 0 &&
diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js
index 9ce21721faad80..9e6050139dc79a 100644
--- a/lib/internal/fs/streams.js
+++ b/lib/internal/fs/streams.js
@@ -10,11 +10,11 @@ const {
 
 const {
   ERR_INVALID_ARG_TYPE,
-  ERR_OUT_OF_RANGE,
-  ERR_STREAM_DESTROYED
+  ERR_OUT_OF_RANGE
 } = require('internal/errors').codes;
 const { deprecate } = require('internal/util');
 const { validateInteger } = require('internal/validators');
+const { errorOrDestroy } = require('internal/streams/destroy');
 const fs = require('fs');
 const { Buffer } = require('buffer');
 const {
@@ -49,6 +49,57 @@ function roundUpToMultipleOf8(n) {
   return (n + 7) & ~7;  // Align to 8 byte boundary.
 }
 
+function _construct(callback) {
+  const stream = this;
+  if (typeof stream.fd === 'number') {
+    callback();
+    return;
+  }
+
+  if (stream.open !== openWriteFs && stream.open !== openReadFs) {
+    // Backwards compat for monkey patching open().
+    const orgEmit = stream.emit;
+    stream.emit = function(...args) {
+      if (args[0] === 'open') {
+        this.emit = orgEmit;
+        callback();
+        orgEmit.apply(this, args);
+      } else if (args[0] === 'error') {
+        this.emit = orgEmit;
+        callback(args[1]);
+      } else {
+        orgEmit.apply(this, args);
+      }
+    };
+    stream.open();
+  } else {
+    stream[kFs].open(stream.path, stream.flags, stream.mode, (er, fd) => {
+      if (er) {
+        callback(er);
+      } else {
+        stream.fd = fd;
+        callback();
+        stream.emit('open', stream.fd);
+        stream.emit('ready');
+      }
+    });
+  }
+}
+
+function close(stream, err, cb) {
+  if (!stream.fd) {
+    // TODO(ronag)
+    // stream.closed = true;
+    cb(err);
+  } else {
+    stream[kFs].close(stream.fd, (er) => {
+      stream.closed = true;
+      cb(er || err);
+    });
+    stream.fd = null;
+  }
+}
+
 function ReadStream(path, options) {
   if (!(this instanceof ReadStream))
     return new ReadStream(path, options);
@@ -79,7 +130,8 @@ function ReadStream(path, options) {
                                    this[kFs].close);
   }
 
-  Readable.call(this, options);
+  options.autoDestroy = options.autoClose === undefined ?
+    true : options.autoClose;
 
   // Path will be ignored when fd is specified, so it can be falsy
   this.path = toPathIfFileURL(path);
@@ -89,7 +141,6 @@ function ReadStream(path, options) {
 
   this.start = options.start;
   this.end = options.end;
-  this.autoClose = options.autoClose === undefined ? true : options.autoClose;
   this.pos = undefined;
   this.bytesRead = 0;
   this.closed = false;
@@ -115,56 +166,28 @@ function ReadStream(path, options) {
     }
   }
 
-  if (typeof this.fd !== 'number')
-    _openReadFs(this);
-
-  this.on('end', function() {
-    if (this.autoClose) {
-      this.destroy();
-    }
-  });
+  Readable.call(this, options);
 }
 ObjectSetPrototypeOf(ReadStream.prototype, Readable.prototype);
 ObjectSetPrototypeOf(ReadStream, Readable);
 
+ObjectDefineProperty(ReadStream.prototype, 'autoClose', {
+  get() {
+    return this._readableState.autoDestroy;
+  },
+  set(val) {
+    this._readableState.autoDestroy = val;
+  }
+});
+
 const openReadFs = deprecate(function() {
-  _openReadFs(this);
+  // Noop.
 }, 'ReadStream.prototype.open() is deprecated', 'DEP0135');
 ReadStream.prototype.open = openReadFs;
 
-function _openReadFs(stream) {
-  // Backwards compat for overriden open.
-  if (stream.open !== openReadFs) {
-    stream.open();
-    return;
-  }
-
-  stream[kFs].open(stream.path, stream.flags, stream.mode, (er, fd) => {
-    if (er) {
-      if (stream.autoClose) {
-        stream.destroy();
-      }
-      stream.emit('error', er);
-      return;
-    }
-
-    stream.fd = fd;
-    stream.emit('open', fd);
-    stream.emit('ready');
-    // Start the flow of data.
-    stream.read();
-  });
-}
+ReadStream.prototype._construct = _construct;
 
 ReadStream.prototype._read = function(n) {
-  if (typeof this.fd !== 'number') {
-    return this.once('open', function() {
-      this._read(n);
-    });
-  }
-
-  if (this.destroyed) return;
-
   if (!pool || pool.length - pool.used < kMinPoolSpace) {
     // Discard the old pool.
     allocNewPool(this.readableHighWaterMark);
@@ -189,17 +212,14 @@ ReadStream.prototype._read = function(n) {
 
   // the actual read.
   this[kIsPerformingIO] = true;
-  this[kFs].read(
-    this.fd, pool, pool.used, toRead, this.pos, (er, bytesRead) => {
+  this[kFs]
+    .read(this.fd, pool, pool.used, toRead, this.pos, (er, bytesRead) => {
       this[kIsPerformingIO] = false;
       // Tell ._destroy() that it's safe to close the fd now.
       if (this.destroyed) return this.emit(kIoDone, er);
 
       if (er) {
-        if (this.autoClose) {
-          this.destroy();
-        }
-        this.emit('error', er);
+        errorOrDestroy(this, er);
       } else {
         let b = null;
         // Now that we know how much data we have actually read, re-wind the
@@ -235,28 +255,13 @@ 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));
-    return;
-  }
-
   if (this[kIsPerformingIO]) {
-    this.once(kIoDone, (er) => closeFsStream(this, cb, err || er));
-    return;
+    this.once(kIoDone, (er) => close(this, err || er, cb));
+  } else {
+    close(this, err, cb);
   }
-
-  closeFsStream(this, cb, err);
 };
 
-function closeFsStream(stream, cb, err) {
-  stream[kFs].close(stream.fd, (er) => {
-    stream.closed = true;
-    cb(er || err);
-  });
-
-  stream.fd = null;
-}
-
 ReadStream.prototype.close = function(cb) {
   if (typeof cb === 'function') finished(this, cb);
   this.destroy();
@@ -276,11 +281,6 @@ function WriteStream(path, options) {
   // Only buffers are supported.
   options.decodeStrings = true;
 
-  if (options.autoDestroy === undefined) {
-    options.autoDestroy = options.autoClose === undefined ?
-      true : (options.autoClose || false);
-  }
-
   this[kFs] = options.fs || fs;
   if (typeof this[kFs].open !== 'function') {
     throw new ERR_INVALID_ARG_TYPE('options.fs.open', 'function',
@@ -315,7 +315,8 @@ function WriteStream(path, options) {
     this._writev = null;
   }
 
-  Writable.call(this, options);
+  options.autoDestroy = options.autoClose === undefined ?
+    true : options.autoClose;
 
   // Path will be ignored when fd is specified, so it can be falsy
   this.path = toPathIfFileURL(path);
@@ -324,7 +325,6 @@ function WriteStream(path, options) {
   this.mode = options.mode === undefined ? 0o666 : options.mode;
 
   this.start = options.start;
-  this.autoClose = options.autoDestroy;
   this.pos = undefined;
   this.bytesWritten = 0;
   this.closed = false;
@@ -336,67 +336,36 @@ function WriteStream(path, options) {
     this.pos = this.start;
   }
 
+  Writable.call(this, options);
+
   if (options.encoding)
     this.setDefaultEncoding(options.encoding);
-
-  if (typeof this.fd !== 'number')
-    _openWriteFs(this);
 }
 ObjectSetPrototypeOf(WriteStream.prototype, Writable.prototype);
 ObjectSetPrototypeOf(WriteStream, Writable);
 
-WriteStream.prototype._final = function(callback) {
-  if (typeof this.fd !== 'number') {
-    return this.once('open', function() {
-      this._final(callback);
-    });
+ObjectDefineProperty(WriteStream.prototype, 'autoClose', {
+  get() {
+    return this._writableState.autoDestroy;
+  },
+  set(val) {
+    this._writableState.autoDestroy = val;
   }
-
-  callback();
-};
+});
 
 const openWriteFs = deprecate(function() {
-  _openWriteFs(this);
+  // Noop.
 }, 'WriteStream.prototype.open() is deprecated', 'DEP0135');
 WriteStream.prototype.open = openWriteFs;
 
-function _openWriteFs(stream) {
-  // Backwards compat for overriden open.
-  if (stream.open !== openWriteFs) {
-    stream.open();
-    return;
-  }
-
-  stream[kFs].open(stream.path, stream.flags, stream.mode, (er, fd) => {
-    if (er) {
-      if (stream.autoClose) {
-        stream.destroy();
-      }
-      stream.emit('error', er);
-      return;
-    }
-
-    stream.fd = fd;
-    stream.emit('open', fd);
-    stream.emit('ready');
-  });
-}
-
+WriteStream.prototype._construct = _construct;
 
 WriteStream.prototype._write = function(data, encoding, cb) {
-  if (typeof this.fd !== 'number') {
-    return this.once('open', function() {
-      this._write(data, encoding, cb);
-    });
-  }
-
-  if (this.destroyed) return cb(new ERR_STREAM_DESTROYED('write'));
-
   this[kIsPerformingIO] = true;
   this[kFs].write(this.fd, data, 0, data.length, this.pos, (er, bytes) => {
     this[kIsPerformingIO] = false;
-    // Tell ._destroy() that it's safe to close the fd now.
     if (this.destroyed) {
+      // Tell ._destroy() that it's safe to close the fd now.
       cb(er);
       return this.emit(kIoDone, er);
     }
@@ -404,6 +373,7 @@ WriteStream.prototype._write = function(data, encoding, cb) {
     if (er) {
       return cb(er);
     }
+
     this.bytesWritten += bytes;
     cb();
   });
@@ -412,16 +382,7 @@ WriteStream.prototype._write = function(data, encoding, cb) {
     this.pos += data.length;
 };
 
-
 WriteStream.prototype._writev = function(data, cb) {
-  if (typeof this.fd !== 'number') {
-    return this.once('open', function() {
-      this._writev(data, cb);
-    });
-  }
-
-  if (this.destroyed) return cb(new ERR_STREAM_DESTROYED('write'));
-
   const len = data.length;
   const chunks = new Array(len);
   let size = 0;
@@ -436,18 +397,16 @@ WriteStream.prototype._writev = function(data, cb) {
   this[kIsPerformingIO] = true;
   this[kFs].writev(this.fd, chunks, this.pos, (er, bytes) => {
     this[kIsPerformingIO] = false;
-    // Tell ._destroy() that it's safe to close the fd now.
     if (this.destroyed) {
+      // Tell ._destroy() that it's safe to close the fd now.
       cb(er);
       return this.emit(kIoDone, er);
     }
 
     if (er) {
-      if (this.autoClose) {
-        this.destroy(er);
-      }
       return cb(er);
     }
+
     this.bytesWritten += bytes;
     cb();
   });
@@ -456,8 +415,13 @@ WriteStream.prototype._writev = function(data, cb) {
     this.pos += size;
 };
 
-
-WriteStream.prototype._destroy = ReadStream.prototype._destroy;
+WriteStream.prototype._destroy = function(err, cb) {
+  if (this[kIsPerformingIO]) {
+    this.once(kIoDone, (er) => close(this, err || er, cb));
+  } else {
+    close(this, err, cb);
+  }
+};
 WriteStream.prototype.close = function(cb) {
   if (cb) {
     if (this.closed) {
diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index ce476467a151dc..58e3f5199445ed 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -1,10 +1,20 @@
 'use strict';
 
+const {
+  ERR_MULTIPLE_CALLBACK
+} = require('internal/errors').codes;
+const { Symbol } = primordials;
+
+const kDestroy = Symbol('kDestroy');
+const kConstruct = Symbol('kConstruct');
+
 // Backwards compat. cb() is undocumented and unused in core but
 // unfortunately might be used by modules.
 function destroy(err, cb) {
   const r = this._readableState;
   const w = this._writableState;
+  // With duplex streams we use the writable side for state.
+  const s = w || r;
 
   if ((w && w.destroyed) || (r && r.destroyed)) {
     if (typeof cb === 'function') {
@@ -33,7 +43,23 @@ function destroy(err, cb) {
     r.destroyed = true;
   }
 
-  this._destroy(err || null, (err) => {
+  // If still constructing then defer calling _destroy.
+  if (!s.constructed) {
+    this.once(kDestroy, function(er) {
+      _destroy(this, err || er, cb);
+    });
+  } 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;
+
     if (err) {
       if (w) {
         w.errored = true;
@@ -55,13 +81,11 @@ function destroy(err, cb) {
     }
 
     if (err) {
-      process.nextTick(emitErrorCloseNT, this, err);
+      process.nextTick(emitErrorCloseNT, self, err);
     } else {
-      process.nextTick(emitCloseNT, this);
+      process.nextTick(emitCloseNT, self);
     }
   });
-
-  return this;
 }
 
 function emitErrorCloseNT(self, err) {
@@ -108,6 +132,7 @@ function undestroy() {
   const w = this._writableState;
 
   if (r) {
+    r.constructed = true;
     r.closed = false;
     r.closeEmitted = false;
     r.destroyed = false;
@@ -119,6 +144,7 @@ function undestroy() {
   }
 
   if (w) {
+    w.constructed = true;
     w.destroyed = false;
     w.closed = false;
     w.closeEmitted = false;
@@ -155,7 +181,6 @@ function errorOrDestroy(stream, err, sync) {
     if (r) {
       r.errored = true;
     }
-
     if (sync) {
       process.nextTick(emitErrorNT, stream, err);
     } else {
@@ -164,6 +189,66 @@ function errorOrDestroy(stream, err, sync) {
   }
 }
 
+function construct(stream, cb) {
+  if (typeof stream._construct !== 'function') {
+    return;
+  }
+
+  stream.once(kConstruct, cb);
+
+  if (stream.listenerCount(kConstruct) > 1) {
+    // Duplex
+    return;
+  }
+
+  const r = stream._readableState;
+  const w = stream._writableState;
+
+  if (r) {
+    r.constructed = false;
+  }
+  if (w) {
+    w.constructed = false;
+  }
+
+  process.nextTick(constructNT, stream);
+}
+
+function constructNT(stream) {
+  const r = stream._readableState;
+  const w = stream._writableState;
+  // With duplex streams we use the writable side for state.
+  const s = w || r;
+
+  let called = false;
+  stream._construct((err) => {
+    if (r) {
+      r.constructed = true;
+    }
+    if (w) {
+      w.constructed = true;
+    }
+
+    if (called) {
+      err = new ERR_MULTIPLE_CALLBACK();
+    } else {
+      called = true;
+    }
+
+    if (s.destroyed) {
+      stream.emit(kDestroy, err);
+    } else if (err) {
+      errorOrDestroy(stream, err, true);
+    } else {
+      process.nextTick(emitConstructNT, stream);
+    }
+  });
+}
+
+function emitConstructNT(stream) {
+  stream.emit(kConstruct);
+}
+
 function isRequest(stream) {
   return stream && stream.setHeader && typeof stream.abort === 'function';
 }
@@ -177,6 +262,7 @@ function destroyer(stream, err) {
 }
 
 module.exports = {
+  construct,
   destroyer,
   destroy,
   undestroy,
diff --git a/test/parallel/test-file-write-stream.js b/test/parallel/test-file-write-stream.js
index 6cab71a4353ff2..09f2063e5623f3 100644
--- a/test/parallel/test-file-write-stream.js
+++ b/test/parallel/test-file-write-stream.js
@@ -62,12 +62,12 @@ file
     assert.strictEqual(file.bytesWritten, EXPECTED.length * 2);
 
     callbacks.close++;
-    console.error('write after end should not be allowed');
     file.write('should not work anymore', common.expectsError({
       code: 'ERR_STREAM_WRITE_AFTER_END',
       name: 'Error',
       message: 'write after end'
     }));
+    file.on('error', common.mustNotCall());
 
     fs.unlinkSync(fn);
   });
diff --git a/test/parallel/test-fs-read-stream-patch-open.js b/test/parallel/test-fs-read-stream-patch-open.js
index c658105132b155..6fa97737b187a3 100644
--- a/test/parallel/test-fs-read-stream-patch-open.js
+++ b/test/parallel/test-fs-read-stream-patch-open.js
@@ -10,6 +10,8 @@ const s = fs.createReadStream('asd')
   .on('error', () => {});
 s.open();
 
-// Allow overriding open().
-fs.ReadStream.prototype.open = common.mustCall();
-fs.createReadStream('asd');
+process.nextTick(() => {
+  // Allow overriding open().
+  fs.ReadStream.prototype.open = common.mustCall();
+  fs.createReadStream('asd');
+});
diff --git a/test/parallel/test-fs-stream-construct.js b/test/parallel/test-fs-stream-construct.js
new file mode 100644
index 00000000000000..4e6bf0285fc91e
--- /dev/null
+++ b/test/parallel/test-fs-stream-construct.js
@@ -0,0 +1,209 @@
+'use strict';
+
+const common = require('../common');
+const fs = require('fs');
+const assert = require('assert');
+const fixtures = require('../common/fixtures');
+
+const tmpdir = require('../common/tmpdir');
+tmpdir.refresh();
+
+const examplePath = fixtures.path('x.txt');
+
+{
+  // 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;
+      // Make sure 'ready' is emitted in same tick as 'open'.
+      assert.strictEqual(ticked, false);
+    }))
+    .on('error', common.mustNotCall())
+    .on('open', common.mustCall((fd) => {
+      process.nextTick(() => {
+        ticked = true;
+        r.destroy();
+      });
+      assert.strictEqual(readyCalled, false);
+      assert.strictEqual(fd, r.fd);
+    }));
+}
+
+{
+  // 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(`${tmpdir.path}/dummy`)
+    .on('ready', common.mustCall(() => {
+      readyCalled = true;
+      // Make sure 'ready' is emitted in same tick as 'open'.
+      assert.strictEqual(ticked, false);
+    }))
+    .on('error', common.mustNotCall())
+    .on('open', common.mustCall((fd) => {
+      process.nextTick(() => {
+        ticked = true;
+        w.destroy();
+      });
+      assert.strictEqual(readyCalled, false);
+      assert.strictEqual(fd, w.fd);
+    }));
+}
+
+{
+  // 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();
+      }
+    });
+  });
+
+  const r = new ReadStream(examplePath)
+    .on('open', common.mustCall((fd) => {
+      assert.strictEqual(fd, r.fd);
+      r.destroy();
+    }));
+}
+
+{
+  // 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);
+      }
+    });
+  });
+
+  const w = new WriteStream(`${tmpdir.path}/dummy2`)
+    .on('open', common.mustCall((fd) => {
+      assert.strictEqual(fd, w.fd);
+      w.destroy();
+    }));
+}
+
+{
+  // 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', err);
+    });
+  });
+
+  const r = new ReadStream('/doesnotexist', { emitClose: true })
+    .on('error', common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ENOENT');
+      assert.strictEqual(r.destroyed, true);
+      r.on('close', common.mustCall());
+    }));
+}
+
+{
+  // 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', err);
+    });
+  });
+
+  const w = new WriteStream(`${tmpdir.path}/dummy2`,
+                            { flags: 'wx+', emitClose: true })
+    .on('error', common.mustCall((err) => {
+      assert.strictEqual(err.code, 'EEXIST');
+      w.destroy();
+      w.on('close', common.mustCall());
+    }));
+}
diff --git a/test/parallel/test-fs-write-stream-patch-open.js b/test/parallel/test-fs-write-stream-patch-open.js
index ac683785cbfec7..9e7bb06af528a5 100644
--- a/test/parallel/test-fs-write-stream-patch-open.js
+++ b/test/parallel/test-fs-write-stream-patch-open.js
@@ -29,6 +29,8 @@ common.expectWarning(
 const s = fs.createWriteStream(`${tmpdir.path}/out`);
 s.open();
 
-// Allow overriding open().
-fs.WriteStream.prototype.open = common.mustCall();
-fs.createWriteStream('asd');
+process.nextTick(() => {
+  // Allow overriding open().
+  fs.WriteStream.prototype.open = common.mustCall();
+  fs.createWriteStream('asd');
+});
diff --git a/test/parallel/test-stream-construct.js b/test/parallel/test-stream-construct.js
new file mode 100644
index 00000000000000..a3e2b4d2e3e8bd
--- /dev/null
+++ b/test/parallel/test-stream-construct.js
@@ -0,0 +1,244 @@
+'use strict';
+
+const common = require('../common');
+const { Writable, Readable, Duplex } = require('stream');
+const assert = require('assert');
+
+{
+  // Multiple callback.
+  new Writable({
+    construct: common.mustCall((callback) => {
+      callback();
+      callback();
+    })
+  }).on('error', common.expectsError({
+    name: 'Error',
+    code: 'ERR_MULTIPLE_CALLBACK'
+  }));
+}
+
+{
+  // Multiple callback.
+  new Readable({
+    construct: common.mustCall((callback) => {
+      callback();
+      callback();
+    })
+  }).on('error', common.expectsError({
+    name: 'Error',
+    code: 'ERR_MULTIPLE_CALLBACK'
+  }));
+}
+
+{
+  // Synchronous error.
+
+  new Writable({
+    construct: common.mustCall((callback) => {
+      callback(new Error('test'));
+    })
+  }).on('error', common.expectsError({
+    name: 'Error',
+    message: 'test'
+  }));
+}
+
+{
+  // Synchronous error.
+
+  new Readable({
+    construct: common.mustCall((callback) => {
+      callback(new Error('test'));
+    })
+  }).on('error', common.expectsError({
+    name: 'Error',
+    message: 'test'
+  }));
+}
+
+{
+  // Asynchronous error.
+
+  new Writable({
+    construct: common.mustCall((callback) => {
+      process.nextTick(callback, new Error('test'));
+    })
+  }).on('error', common.expectsError({
+    name: 'Error',
+    message: 'test'
+  }));
+}
+
+{
+  // Asynchronous error.
+
+  new Readable({
+    construct: common.mustCall((callback) => {
+      process.nextTick(callback, new Error('test'));
+    })
+  }).on('error', common.expectsError({
+    name: 'Error',
+    message: 'test'
+  }));
+}
+
+function testDestroy(factory) {
+  {
+    let constructed = false;
+    const s = factory({
+      construct: common.mustCall((cb) => {
+        constructed = true;
+        process.nextTick(cb);
+      })
+    });
+    s.on('close', common.mustCall(() => {
+      assert.strictEqual(constructed, true);
+    }));
+    s.destroy();
+  }
+
+  {
+    let constructed = false;
+    const s = factory({
+      construct: common.mustCall((cb) => {
+        constructed = true;
+        process.nextTick(cb);
+      })
+    });
+    s.on('close', common.mustCall(() => {
+      assert.strictEqual(constructed, true);
+    }));
+    s.destroy(null, () => {
+      assert.strictEqual(constructed, true);
+    });
+  }
+
+  {
+    let constructed = false;
+    const s = factory({
+      construct: common.mustCall((cb) => {
+        constructed = true;
+        process.nextTick(cb);
+      })
+    });
+    s.on('close', common.mustCall(() => {
+      assert.strictEqual(constructed, true);
+    }));
+    s.destroy();
+  }
+
+
+  {
+    let constructed = false;
+    const s = factory({
+      construct: common.mustCall((cb) => {
+        constructed = true;
+        process.nextTick(cb);
+      })
+    });
+    s.on('close', common.mustCall(() => {
+      assert.strictEqual(constructed, true);
+    }));
+    s.on('error', common.mustCall((err) => {
+      assert.strictEqual(err.message, 'kaboom');
+    }));
+    s.destroy(new Error('kaboom'), (err) => {
+      assert.strictEqual(err.message, 'kaboom');
+      assert.strictEqual(constructed, true);
+    });
+  }
+
+  {
+    let constructed = false;
+    const s = factory({
+      construct: common.mustCall((cb) => {
+        constructed = true;
+        process.nextTick(cb);
+      })
+    });
+    s.on('error', common.mustCall(() => {
+      assert.strictEqual(constructed, true);
+    }));
+    s.on('close', common.mustCall(() => {
+      assert.strictEqual(constructed, true);
+    }));
+    s.destroy(new Error());
+  }
+}
+testDestroy((opts) => new Readable({
+  read: common.mustNotCall(),
+  ...opts
+}));
+testDestroy((opts) => new Writable({
+  write: common.mustNotCall(),
+  final: common.mustNotCall(),
+  ...opts
+}));
+
+{
+  let constructed = false;
+  const r = new Readable({
+    autoDestroy: true,
+    construct: common.mustCall((cb) => {
+      constructed = true;
+      process.nextTick(cb);
+    }),
+    read: common.mustCall(() => {
+      assert.strictEqual(constructed, true);
+      r.push(null);
+    })
+  });
+  r.on('close', common.mustCall(() => {
+    assert.strictEqual(constructed, true);
+  }));
+  r.on('data', common.mustNotCall());
+}
+
+{
+  let constructed = false;
+  const w = new Writable({
+    autoDestroy: true,
+    construct: common.mustCall((cb) => {
+      constructed = true;
+      process.nextTick(cb);
+    }),
+    write: common.mustCall((chunk, encoding, cb) => {
+      assert.strictEqual(constructed, true);
+      process.nextTick(cb);
+    }),
+    final: common.mustCall((cb) => {
+      assert.strictEqual(constructed, true);
+      process.nextTick(cb);
+    })
+  });
+  w.on('close', common.mustCall(() => {
+    assert.strictEqual(constructed, true);
+  }));
+  w.end('data');
+}
+
+{
+  let constructed = false;
+  const w = new Writable({
+    autoDestroy: true,
+    construct: common.mustCall((cb) => {
+      constructed = true;
+      process.nextTick(cb);
+    }),
+    write: common.mustNotCall(),
+    final: common.mustCall((cb) => {
+      assert.strictEqual(constructed, true);
+      process.nextTick(cb);
+    })
+  });
+  w.on('close', common.mustCall(() => {
+    assert.strictEqual(constructed, true);
+  }));
+  w.end();
+}
+
+{
+  new Duplex({
+    construct: common.mustCall()
+  });
+}