stdio binding + javascript to enable process.stdin.listen()

lib/net_uv.js

@@ -79,6 +79,13 @@ exports.Socket = Socket;
 exports.Stream = Socket; // Legacy naming.
 
 
+Socket.prototype.listen = function() {
+  var self = this;
+  self.on('connection', arguments[0]);
+  listen(self, null, null);
+};
+
+
 Socket.prototype.setTimeout = function(msecs, callback) {
   if (msecs > 0) {
     timers.enroll(this, msecs);
@@ -535,8 +542,11 @@ function toPort(x) { return (x = Number(x)) >= 0 ? x : false; }
 function listen(self, address, port, addressType) {
   var r = 0;
 
-  // assign handle in listen, and clean up if bind or listen fails
+  if (!self._handle) {
-  self._handle = (port == -1 && addressType == -1) ? new Pipe : new TCP;
+    // assign handle in listen, and clean up if bind or listen fails
+    self._handle = (port == -1 && addressType == -1) ? new Pipe : new TCP;
+  }
+
   self._handle.socket = self;
   self._handle.onconnection = onconnection;
 

src/handle_wrap.cc

@@ -50,7 +50,9 @@ Handle<Value> HandleWrap::Close(const Arguments& args) {
 
 HandleWrap::HandleWrap(Handle<Object> object, uv_handle_t* h) {
   handle__ = h;
-  h->data = this;
+  if (h) {
+    h->data = this;
+  }
 
   HandleScope scope;
   assert(object_.IsEmpty());
@@ -60,6 +62,12 @@ HandleWrap::HandleWrap(Handle<Object> object, uv_handle_t* h) {
 }
 
 
+void HandleWrap::SetHandle(uv_handle_t* h) {
+  handle__ = h;
+  h->data = this;
+}
+
+
 HandleWrap::~HandleWrap() {
   assert(object_.IsEmpty());
 }

src/handle_wrap.h

@@ -32,6 +32,7 @@ class HandleWrap {
     HandleWrap(v8::Handle<v8::Object> object, uv_handle_t* handle);
     virtual ~HandleWrap();
 
+    virtual void SetHandle(uv_handle_t* h);
     virtual void StateChange() {}
 
     v8::Persistent<v8::Object> object_;

src/node_extensions.h

@@ -46,6 +46,7 @@ NODE_EXT_LIST_ITEM(node_timer_wrap)
 NODE_EXT_LIST_ITEM(node_tcp_wrap)
 NODE_EXT_LIST_ITEM(node_pipe_wrap)
 NODE_EXT_LIST_ITEM(node_cares_wrap)
+NODE_EXT_LIST_ITEM(node_stdio_wrap)
 
 NODE_EXT_LIST_END
 

src/pipe_wrap.cc

@@ -30,7 +30,7 @@ using v8::Context;
 using v8::Arguments;
 using v8::Integer;
 
-static Persistent<Function> constructor;
+Persistent<Function> pipeConstructor;
 
 
 // TODO share with TCPWrap?
@@ -61,9 +61,9 @@ class PipeWrap : StreamWrap {
     NODE_SET_PROTOTYPE_METHOD(t, "listen", Listen);
     NODE_SET_PROTOTYPE_METHOD(t, "connect", Connect);
 
-    constructor = Persistent<Function>::New(t->GetFunction());
+    pipeConstructor = Persistent<Function>::New(t->GetFunction());
 
-    target->Set(String::NewSymbol("Pipe"), constructor);
+    target->Set(String::NewSymbol("Pipe"), pipeConstructor);
   }
 
  private:
@@ -137,7 +137,7 @@ class PipeWrap : StreamWrap {
     }
 
     // Instanciate the client javascript object and handle.
-    Local<Object> client_obj = constructor->NewInstance();
+    Local<Object> client_obj = pipeConstructor->NewInstance();
 
     // Unwrap the client javascript object.
     assert(client_obj->InternalFieldCount() > 0);

src/stdio_wrap.cc

@@ -0,0 +1,157 @@
+#include <node.h>
+#include <node_buffer.h>
+#include <req_wrap.h>
+#include <handle_wrap.h>
+#include <stream_wrap.h>
+
+#define UNWRAP \
+  assert(!args.Holder().IsEmpty()); \
+  assert(args.Holder()->InternalFieldCount() > 0); \
+  StdIOWrap* wrap =  \
+      static_cast<StdIOWrap*>(args.Holder()->GetPointerFromInternalField(0)); \
+  if (!wrap) { \
+    SetErrno(UV_EBADF); \
+    return scope.Close(Integer::New(-1)); \
+  }
+
+namespace node {
+
+using v8::Object;
+using v8::Handle;
+using v8::Local;
+using v8::Persistent;
+using v8::Value;
+using v8::HandleScope;
+using v8::FunctionTemplate;
+using v8::String;
+using v8::Function;
+using v8::TryCatch;
+using v8::Context;
+using v8::Arguments;
+using v8::Integer;
+using v8::Undefined;
+
+extern Persistent<Function> tcpConstructor;
+extern Persistent<Function> pipeConstructor;
+static Persistent<Function> constructor;
+
+
+class StdIOWrap : StreamWrap {
+ public:
+  static void Initialize(Handle<Object> target) {
+    StreamWrap::Initialize(target);
+
+    HandleScope scope;
+
+    Local<FunctionTemplate> t = FunctionTemplate::New(New);
+    t->SetClassName(String::NewSymbol("StdIO"));
+
+    t->InstanceTemplate()->SetInternalFieldCount(1);
+
+    NODE_SET_PROTOTYPE_METHOD(t, "readStart", StreamWrap::ReadStart);
+    NODE_SET_PROTOTYPE_METHOD(t, "readStop", StreamWrap::ReadStop);
+    NODE_SET_PROTOTYPE_METHOD(t, "write", StreamWrap::Write);
+    NODE_SET_PROTOTYPE_METHOD(t, "listen", Listen);
+
+    constructor = Persistent<Function>::New(t->GetFunction());
+
+    target->Set(String::NewSymbol("StdIO"), constructor);
+  }
+
+ private:
+  static Handle<Value> New(const Arguments& args) {
+    // This constructor should not be exposed to public javascript.
+    // Therefore we assert that we are not trying to call this as a
+    // normal function.
+    assert(args.IsConstructCall());
+
+    uv_std_type stdHandleType = (uv_std_type)args[0]->Int32Value();
+
+    assert(stdHandleType == UV_STDIN || stdHandleType == UV_STDOUT || stdHandleType == UV_STDERR);
+
+    uv_stream_t* stdHandle = uv_std_handle(stdHandleType);
+    if (stdHandle) {
+      HandleScope scope;
+      StdIOWrap* wrap = new StdIOWrap(args.This());
+      assert(wrap);
+
+      wrap->handle_ = stdHandle;
+      wrap->SetHandle((uv_handle_t*)stdHandle);
+      wrap->UpdateWriteQueueSize();
+
+      return scope.Close(args.This());
+    } else {
+      return Undefined();
+    }
+  }
+
+  StdIOWrap(Handle<Object> object) : StreamWrap(object, NULL) {
+  }
+
+  static Handle<Value> Listen(const Arguments& args) {
+    HandleScope scope;
+
+    UNWRAP
+
+    int backlog = args[0]->Int32Value();
+
+    int r = uv_listen(wrap->handle_, SOMAXCONN, OnConnection);
+
+    // Error starting the pipe.
+    if (r) SetErrno(uv_last_error().code);
+
+    return scope.Close(Integer::New(r));
+  }
+
+  // TODO maybe share with TCPWrap?
+  static void OnConnection(uv_stream_t* handle, int status) {
+    HandleScope scope;
+    Local<Object> client_obj;
+
+    StdIOWrap* wrap = static_cast<StdIOWrap*>(handle->data);
+    assert(wrap->handle_ == handle);
+
+    // We should not be getting this callback if someone as already called
+    // uv_close() on the handle.
+    assert(wrap->object_.IsEmpty() == false);
+
+    if (status != 0) {
+      // TODO Handle server error (set errno and call onconnection with NULL)
+      assert(0);
+      return;
+    }
+
+    // Instanciate the client javascript object and handle.
+    switch (handle->type) {
+      case UV_TCP:
+        client_obj = tcpConstructor->NewInstance();
+        break;
+      case UV_NAMED_PIPE:
+        client_obj = pipeConstructor->NewInstance();
+        break;
+      default:
+        assert(0);
+        return;
+    }
+
+    // Unwrap the client javascript object.
+    assert(client_obj->InternalFieldCount() > 0);
+    StreamWrap* client_wrap =
+        static_cast<StreamWrap*>(client_obj->GetPointerFromInternalField(0));
+
+    int r = uv_accept(handle, client_wrap->GetStream());
+
+    // uv_accept should always work.
+    assert(r == 0);
+
+    // Successful accept. Call the onconnection callback in JavaScript land.
+    Local<Value> argv[1] = { client_obj };
+    MakeCallback(wrap->object_, "onconnection", 1, argv);
+  }
+
+  uv_stream_t* handle_;
+};
+
+}  // namespace node
+
+NODE_MODULE(node_stdio_wrap, node::StdIOWrap::Initialize);

src/stream_wrap.cc

@@ -71,7 +71,16 @@ void StreamWrap::Initialize(Handle<Object> target) {
 StreamWrap::StreamWrap(Handle<Object> object, uv_stream_t* stream)
     : HandleWrap(object, (uv_handle_t*)stream) {
   stream_ = stream;
-  stream->data = this;
+  if (stream) {
+    stream->data = this;
+  }
+}
+
+
+void StreamWrap::SetHandle(uv_handle_t* h) {
+  HandleWrap::SetHandle(h);
+  stream_ = (uv_stream_t*)h;
+  stream_->data = this;
 }
 
 

src/stream_wrap.h

@@ -9,6 +9,8 @@ namespace node {
 
 class StreamWrap : public HandleWrap {
  public:
+  uv_stream_t* GetStream() { return stream_; }
+
   static void Initialize(v8::Handle<v8::Object> target);
 
   // JavaScript functions
@@ -20,6 +22,7 @@ class StreamWrap : public HandleWrap {
  protected:
   StreamWrap(v8::Handle<v8::Object> object, uv_stream_t* stream);
   virtual ~StreamWrap() { }
+  virtual void SetHandle(uv_handle_t* h);
   void StateChange() { }
   void UpdateWriteQueueSize();
 

src/tcp_wrap.cc

@@ -45,7 +45,7 @@ using v8::Context;
 using v8::Arguments;
 using v8::Integer;
 
-static Persistent<Function> constructor;
+Persistent<Function> tcpConstructor;
 
 static Persistent<String> family_symbol;
 static Persistent<String> address_symbol;
@@ -83,13 +83,13 @@ class TCPWrap : public StreamWrap {
     NODE_SET_PROTOTYPE_METHOD(t, "connect6", Connect6);
     NODE_SET_PROTOTYPE_METHOD(t, "getsockname", GetSockName);
 
-    constructor = Persistent<Function>::New(t->GetFunction());
+    tcpConstructor = Persistent<Function>::New(t->GetFunction());
 
     family_symbol = NODE_PSYMBOL("family");
     address_symbol = NODE_PSYMBOL("address");
     port_symbol = NODE_PSYMBOL("port");
 
-    target->Set(String::NewSymbol("TCP"), constructor);
+    target->Set(String::NewSymbol("TCP"), tcpConstructor);
   }
 
  private:
@@ -221,7 +221,7 @@ class TCPWrap : public StreamWrap {
     }
 
     // Instanciate the client javascript object and handle.
-    Local<Object> client_obj = constructor->NewInstance();
+    Local<Object> client_obj = tcpConstructor->NewInstance();
 
     // Unwrap the client javascript object.
     assert(client_obj->InternalFieldCount() > 0);

wscript

@@ -867,6 +867,7 @@ def build(bld):
     src/tcp_wrap.cc
     src/pipe_wrap.cc
     src/cares_wrap.cc
+    src/stdio_wrap.cc
   """
 
   if sys.platform.startswith("win32"):