tty/stdin: Refactor for streams2

lib/tty.js

@@ -40,42 +40,47 @@ exports.setRawMode = util.deprecate(function(flag) {
 }, 'tty.setRawMode: Use `process.stdin.setRawMode()` instead.');
 
 
-function ReadStream(fd) {
-  if (!(this instanceof ReadStream)) return new ReadStream(fd);
-  net.Socket.call(this, {
+function ReadStream(fd, options) {
+  if (!(this instanceof ReadStream))
+    return new ReadStream(fd, options);
+
+  options = util._extend({
+    highWaterMark: 0,
+    lowWaterMark: 0,
     handle: new TTY(fd, true)
-  });
+  }, options);
+
+  net.Socket.call(this, options);
 
   this.readable = true;
   this.writable = false;
   this.isRaw = false;
+  this.isTTY = true;
+
+  // this.read = function(orig) { return function(n) {
+  //   var ret = orig.apply(this, arguments);
+  //   console.trace('TTY read(' + n + ') -> ' + ret);
+  //   return ret;
+  // } }(this.read);
 }
 inherits(ReadStream, net.Socket);
 
 exports.ReadStream = ReadStream;
 
-ReadStream.prototype.pause = function() {
-  return net.Socket.prototype.pause.call(this);
-};
-
-ReadStream.prototype.resume = function() {
-  return net.Socket.prototype.resume.call(this);
-};
-
 ReadStream.prototype.setRawMode = function(flag) {
   flag = !!flag;
   this._handle.setRawMode(flag);
   this.isRaw = flag;
 };
 
-ReadStream.prototype.isTTY = true;
-
 
 
 function WriteStream(fd) {
   if (!(this instanceof WriteStream)) return new WriteStream(fd);
   net.Socket.call(this, {
-    handle: new TTY(fd, false)
+    handle: new TTY(fd, false),
+    readable: false,
+    writable: true
   });
 
   this.readable = false;

src/node.js

@@ -140,7 +140,6 @@
 
       } else {
         // Read all of stdin - execute it.
-        process.stdin.resume();
         process.stdin.setEncoding('utf8');
 
         var code = '';
@@ -497,17 +496,20 @@
       switch (tty_wrap.guessHandleType(fd)) {
         case 'TTY':
           var tty = NativeModule.require('tty');
-          stdin = new tty.ReadStream(fd);
+          stdin = new tty.ReadStream(fd, {
+            highWaterMark: 0,
+            lowWaterMark: 0
+          });
           break;
 
         case 'FILE':
           var fs = NativeModule.require('fs');
-          stdin = new fs.ReadStream(null, {fd: fd});
+          stdin = new fs.ReadStream(null, { fd: fd });
           break;
 
         case 'PIPE':
           var net = NativeModule.require('net');
-          stdin = new net.Stream(fd);
+          stdin = new net.Stream({ fd: fd });
           stdin.readable = true;
           break;
 
@@ -520,16 +522,23 @@
       stdin.fd = fd;
 
       // stdin starts out life in a paused state, but node doesn't
-      // know yet.  Call pause() explicitly to unref() it.
-      stdin.pause();
+      // know yet.  Explicitly to readStop() it to put it in the
+      // not-reading state.
+      if (stdin._handle && stdin._handle.readStop) {
+        stdin._handle.reading = false;
+        stdin._readableState.reading = false;
+        stdin._handle.readStop();
+      }
 
-      // when piping stdin to a destination stream,
-      // let the data begin to flow.
-      var pipe = stdin.pipe;
-      stdin.pipe = function(dest, opts) {
-        stdin.resume();
-        return pipe.call(stdin, dest, opts);
-      };
+      // if the user calls stdin.pause(), then we need to stop reading
+      // immediately, so that the process can close down.
+      stdin.on('pause', function() {
+        if (!stdin._handle)
+          return;
+        stdin._readableState.reading = false;
+        stdin._handle.reading = false;
+        stdin._handle.readStop();
+      });
 
       return stdin;
     });