diff --git a/lib/internal/streams/transform.js b/lib/internal/streams/transform.js
index 2527ce94c8f44d..02646f44344901 100644
--- a/lib/internal/streams/transform.js
+++ b/lib/internal/streams/transform.js
@@ -182,7 +182,12 @@ Transform.prototype._write = function(chunk, encoding, callback) {
       this.push(val);
     }
 
-    if (
+    if (rState.ended) {
+      // If user has called this.push(null) we have to
+      // delay the callback to properly progate the new
+      // state.
+      process.nextTick(callback);
+    } else if (
       wState.ended || // Backwards compat.
       length === rState.length || // Backwards compat.
       rState.length < rState.highWaterMark
diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js
index 8b44731b462a95..9851b77bd8fa98 100644
--- a/lib/internal/streams/writable.js
+++ b/lib/internal/streams/writable.js
@@ -517,14 +517,6 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
 
   state.length += len;
 
-  // stream._write resets state.length
-  const ret = state.length < state.highWaterMark;
-
-  // We must ensure that previous needDrain will not be reset to false.
-  if (!ret) {
-    state.state |= kNeedDrain;
-  }
-
   if ((state.state & (kWriting | kErrored | kCorked | kConstructed)) !== kConstructed) {
     state.buffered.push({ chunk, encoding, callback });
     state.state |= kHasBuffer;
@@ -544,6 +536,12 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
     state.state &= ~kSync;
   }
 
+  const ret = state.length < state.highWaterMark;
+
+  if (!ret) {
+    state.state |= kNeedDrain;
+  }
+
   // Return false if errored or destroyed in order to break
   // any synchronous while(stream.write(data)) loops.
   return ret && (state.state & (kDestroyed | kErrored)) === 0;