stream: error Duplex write/read if not writable/readable

lib/internal/bootstrap/switches/is_main_thread.js

@@ -148,11 +148,7 @@ function getStdin() {
   switch (guessHandleType(fd)) {
     case 'TTY':
       const tty = require('tty');
-      stdin = new tty.ReadStream(fd, {
-        highWaterMark: 0,
-        readable: true,
-        writable: false
-      });
+      stdin = new tty.ReadStream(fd);
       break;
 
     case 'FILE':

lib/internal/child_process.js

@@ -324,7 +324,7 @@ function flushStdio(subprocess) {
 
 
 function createSocket(pipe, readable) {
-  return net.Socket({ handle: pipe, readable, writable: !readable });
+  return net.Socket({ handle: pipe, readable });
 }
 
 
@@ -442,8 +442,6 @@ ChildProcess.prototype.spawn = function(options) {
     }
 
     if (stream.handle) {
-      // When i === 0 - we're dealing with stdin
-      // (which is the only one writable pipe).
       stream.socket = createSocket(this.pid !== 0 ?
         stream.handle : null, i > 0);
 

lib/internal/http2/core.js

@@ -358,18 +358,23 @@ function onSessionHeaders(handle, id, cat, flags, headers, sensitiveHeaders) {
       handle.destroy();
       return;
     }
-    const opts = { readable: !endOfStream };
     // session[kType] can be only one of two possible values
     if (type === NGHTTP2_SESSION_SERVER) {
-      stream = new ServerHttp2Stream(session, handle, id, opts, obj);
+      stream = new ServerHttp2Stream(session, handle, id, {}, obj);
+      if (endOfStream) {
+        stream.push(null);
+      }
       if (obj[HTTP2_HEADER_METHOD] === HTTP2_METHOD_HEAD) {
         // For head requests, there must not be a body...
         // end the writable side immediately.
         stream.end();
         stream[kState].flags |= STREAM_FLAGS_HEAD_REQUEST;
       }
     } else {
-      stream = new ClientHttp2Stream(session, handle, id, opts);
+      stream = new ClientHttp2Stream(session, handle, id, {});
+      if (endOfStream) {
+        stream.push(null);
+      }
       stream.end();
     }
     if (endOfStream)
@@ -2675,7 +2680,6 @@ class ServerHttp2Stream extends Http2Stream {
     let headRequest = false;
     if (headers[HTTP2_HEADER_METHOD] === HTTP2_METHOD_HEAD)
       headRequest = options.endStream = true;
-    options.readable = false;
 
     const headersList = mapToHeaders(headers);
 
@@ -2703,6 +2707,8 @@ class ServerHttp2Stream extends Http2Stream {
     const stream = new ServerHttp2Stream(session, ret, id, options, headers);
     stream[kSentHeaders] = headers;
 
+    stream.push(null);
+
     if (options.endStream)
       stream.end();
 

lib/internal/streams/destroy.js

@@ -206,8 +206,8 @@ function undestroy() {
     r.errored = null;
     r.errorEmitted = false;
     r.reading = false;
-    r.ended = false;
-    r.endEmitted = false;
+    r.ended = r.readable === false;
+    r.endEmitted = r.readable === false;
   }
 
   if (w) {
@@ -217,11 +217,11 @@ function undestroy() {
     w.closeEmitted = false;
     w.errored = null;
     w.errorEmitted = false;
-    w.ended = false;
-    w.ending = false;
     w.finalCalled = false;
     w.prefinished = false;
-    w.finished = false;
+    w.ended = w.writable === false;
+    w.ending = w.writable === false;
+    w.finished = w.writable === false;
   }
 }
 

lib/internal/streams/duplex.js

@@ -55,18 +55,20 @@ function Duplex(options) {
 
   Readable.call(this, options);
   Writable.call(this, options);
-  this.allowHalfOpen = true;
 
-  if (options) {
-    if (options.readable === false)
-      this.readable = false;
+  this.allowHalfOpen = options?.allowHalfOpen !== false;
 
-    if (options.writable === false)
-      this.writable = false;
+  if (options?.readable === false) {
+    this._readableState.readable = false;
+    this._readableState.ended = true;
+    this._readableState.endEmitted = true;
+  }
 
-    if (options.allowHalfOpen === false) {
-      this.allowHalfOpen = false;
-    }
+  if (options?.writable === false) {
+    this._writableState.writable = false;
+    this._writableState.ending = true;
+    this._writableState.ended = true;
+    this._writableState.finished = true;
   }
 }
 

lib/tty.js

@@ -51,16 +51,15 @@ function ReadStream(fd, options) {
     throw new ERR_INVALID_FD(fd);
 
   const ctx = {};
-  const tty = new TTY(fd, true, ctx);
+  const tty = new TTY(fd, ctx);
   if (ctx.code !== undefined) {
     throw new ERR_TTY_INIT_FAILED(ctx);
   }
 
   net.Socket.call(this, {
     highWaterMark: 0,
-    readable: true,
-    writable: false,
     handle: tty,
+    manualStart: true,
     ...options
   });
 
@@ -89,15 +88,15 @@ function WriteStream(fd) {
     throw new ERR_INVALID_FD(fd);
 
   const ctx = {};
-  const tty = new TTY(fd, false, ctx);
+  const tty = new TTY(fd, ctx);
   if (ctx.code !== undefined) {
     throw new ERR_TTY_INIT_FAILED(ctx);
   }
 
   net.Socket.call(this, {
+    highWaterMark: 0,
     handle: tty,
-    readable: false,
-    writable: true
+    manualStart: true
   });
 
   // Prevents interleaved or dropped stdout/stderr output for terminals.

src/tty_wrap.cc

@@ -121,9 +121,9 @@ void TTYWrap::New(const FunctionCallbackInfo<Value>& args) {
   CHECK_GE(fd, 0);
 
   int err = 0;
-  new TTYWrap(env, args.This(), fd, args[1]->IsTrue(), &err);
+  new TTYWrap(env, args.This(), fd, &err);
   if (err != 0) {
-    env->CollectUVExceptionInfo(args[2], err, "uv_tty_init");
+    env->CollectUVExceptionInfo(args[1], err, "uv_tty_init");
     args.GetReturnValue().SetUndefined();
   }
 }
@@ -132,13 +132,12 @@ void TTYWrap::New(const FunctionCallbackInfo<Value>& args) {
 TTYWrap::TTYWrap(Environment* env,
                  Local<Object> object,
                  int fd,
-                 bool readable,
                  int* init_err)
     : LibuvStreamWrap(env,
                       object,
                       reinterpret_cast<uv_stream_t*>(&handle_),
                       AsyncWrap::PROVIDER_TTYWRAP) {
-  *init_err = uv_tty_init(env->event_loop(), &handle_, fd, readable);
+  *init_err = uv_tty_init(env->event_loop(), &handle_, fd, 0);
   set_fd(fd);
   if (*init_err != 0)
     MarkAsUninitialized();

src/tty_wrap.h

@@ -46,7 +46,6 @@ class TTYWrap : public LibuvStreamWrap {
   TTYWrap(Environment* env,
           v8::Local<v8::Object> object,
           int fd,
-          bool readable,
           int* init_err);
 
   static void IsTTY(const v8::FunctionCallbackInfo<v8::Value>& args);

test/parallel/test-http2-compat-socket-set.js

@@ -26,9 +26,9 @@ server.on('request', common.mustCall(function(request, response) {
   assert.strictEqual(request.stream.destroyed, true);
   request.socket.destroyed = false;
 
-  assert.strictEqual(request.stream.readable, false);
-  request.socket.readable = true;
   assert.strictEqual(request.stream.readable, true);
+  request.socket.readable = false;
+  assert.strictEqual(request.stream.readable, false);
 
   assert.strictEqual(request.stream.writable, true);
   request.socket.writable = false;

test/parallel/test-stream-duplex-readable-writable.js

@@ -0,0 +1,46 @@
+'use strict';
+
+const common = require('../common');
+const { Duplex } = require('stream');
+const assert = require('assert');
+
+{
+  const duplex = new Duplex({
+    readable: false
+  });
+  assert.strictEqual(duplex.readable, false);
+  duplex.push('asd');
+  duplex.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_PUSH_AFTER_EOF');
+  }));
+  duplex.on('data', common.mustNotCall());
+  duplex.on('end', common.mustNotCall());
+}
+
+{
+  const duplex = new Duplex({
+    writable: false,
+    write: common.mustNotCall()
+  });
+  assert.strictEqual(duplex.writable, false);
+  duplex.write('asd');
+  duplex.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+  duplex.on('finish', common.mustNotCall());
+}
+
+{
+  const duplex = new Duplex({
+    readable: false
+  });
+  assert.strictEqual(duplex.readable, false);
+  duplex.on('data', common.mustNotCall());
+  duplex.on('end', common.mustNotCall());
+  async function run() {
+    for await (const chunk of duplex) {
+      assert(false, chunk);
+    }
+  }
+  run().then(common.mustCall());
+}

test/pseudo-tty/repl-dumb-tty.js

@@ -1,21 +1,13 @@
 'use strict';
 const common = require('../common');
+const process = require('process');
 
 process.env.TERM = 'dumb';
 
 const repl = require('repl');
 const ArrayStream = require('../common/arraystream');
 
 repl.start('> ');
-process.stdin.push('conso'); // No completion preview.
-process.stdin.push('le.log("foo")\n');
-process.stdin.push('1 + 2'); // No input preview.
-process.stdin.push('\n');
-process.stdin.push('"str"\n');
-process.stdin.push('console.dir({ a: 1 })\n');
-process.stdin.push('{ a: 1 }\n');
-process.stdin.push('\n');
-process.stdin.push('.exit\n');
 
 // Verify <ctrl> + D support.
 {
@@ -34,3 +26,13 @@ process.stdin.push('.exit\n');
   replServer.write(null, { ctrl: true, name: 's' });
   replServer.write(null, { ctrl: true, name: 'd' });
 }
+
+process.stdin.push('conso'); // No completion preview.
+process.stdin.push('le.log("foo")\n');
+process.stdin.push('1 + 2'); // No input preview.
+process.stdin.push('\n');
+process.stdin.push('"str"\n');
+process.stdin.push('console.dir({ a: 1 })\n');
+process.stdin.push('{ a: 1 }\n');
+process.stdin.push('\n');
+process.stdin.push('.exit\n');

test/pseudo-tty/repl-dumb-tty.out

@@ -1,4 +1,5 @@
-> console.log("foo")
+> >
+console.log("foo")
 foo
 undefined
 > 1 + 2
@@ -12,4 +13,3 @@ undefined
 { a: 1 }
 >
 > .exit
->