net: add bindings to libuv pipe API

diff --git a/deps/uv/src/uv-unix.c b/deps/uv/src/uv-unix.c
index 842fe32..75ecd6b 100644 (file)
--- a/deps/uv/src/uv-unix.c
+++ b/deps/uv/src/uv-unix.c
@@ -1904,7 +1904,7 @@ int uv_pipe_connect(uv_connect_t* req,
   status = 0;
 
 out:
-  uv__req_init((uv_req_t*)req);
+  if (0) uv__req_init((uv_req_t*)req);
   req->handle = (uv_stream_t*)handle;
   req->type = UV_CONNECT;
   req->cb = cb;

diff --git a/lib/net_uv.js b/lib/net_uv.js
index c273020..cfcf117 100644 (file)
--- a/lib/net_uv.js
+++ b/lib/net_uv.js
@@ -4,6 +4,7 @@ var timers = require('timers');
 var util = require('util');
 var assert = require('assert');
 var TCP = process.binding('tcp_wrap').TCP;
+var Pipe = process.binding('pipe_wrap').Pipe;
 
 /* Bit flags for socket._flags */
 var FLAG_GOT_EOF      = 1 << 0;
@@ -20,13 +21,25 @@ if (process.env.NODE_DEBUG && /net/.test(process.env.NODE_DEBUG)) {
 }
 
 
+function isPipeName(s) {
+  return typeof s === 'string' && toPort(s) === false;
+}
+
+
 exports.createServer = function() {
   return new Server(arguments[0], arguments[1]);
 };
 
 
 exports.connect = exports.createConnection = function(port /* [host], [cb] */) {
-  var s = new Socket();
+  var s;
+
+  if (isPipeName(port)) {
+    s = new Socket({handle:new Pipe});
+  } else {
+    s = new Socket();
+  }
+
   s.connect(port, arguments[1], arguments[2]);
   return s;
 };
@@ -52,7 +65,7 @@ function Socket(options) {
   if (options && options.handle) {
     this._handle = options.handle;
   } else {
-    this._handle = new TCP();
+    this._handle = new TCP;
   }
   this.allowHalfOpen = options ? (options.allowHalfOpen || false) : false;
 
@@ -348,22 +361,27 @@ function afterWrite(status, handle, req, buffer) {
 }
 
 
-function connectip(self, port, ip, addressType) {
-  self.remoteAddress = ip;
-  self.remotePort = port;
+function connect(self, address, port, addressType) {
+  if (port) {
+    self.remotePort = port;
+  }
+  self.remoteAddress = address;
 
   // TODO return promise from Socket.prototype.connect which
   // wraps _connectReq.
 
   assert.ok(self._connecting);
 
+  var connectReq;
   if (addressType == 6) {
-    var connectReq = self._handle.connect6(ip, port);
+    connectReq = self._handle.connect6(address, port);
+  } else if (addressType == 4) {
+    connectReq = self._handle.connect(address, port);
   } else {
-    var connectReq = self._handle.connect(ip, port);
+    connectReq = self._handle.connect(address, afterConnect);
   }
 
-  if (connectReq) {
+  if (connectReq !== null) {
     connectReq.oncomplete = afterConnect;
   } else {
     self.destroy(errnoException(errno, 'connect'));
@@ -374,11 +392,13 @@ function connectip(self, port, ip, addressType) {
 Socket.prototype.connect = function(port /* [host], [cb] */) {
   var self = this;
 
+  var pipe = isPipeName(port);
+
   if (this.destroyed) {
-    this._handle = new TCP();
+    this._handle = pipe ? new Pipe() : new TCP();
     initSocketHandle(this);
   }
-    
+
   var host;
   if (typeof arguments[1] === 'function') {
     self.on('connect', arguments[1]);
@@ -394,7 +414,10 @@ Socket.prototype.connect = function(port /* [host], [cb] */) {
   self._connecting = true;
   self.writable = true;
 
-  if (typeof host == 'string') {
+  if (pipe) {
+    connect(self, /*pipe_name=*/port);
+
+  } else if (typeof host == 'string') {
     debug("connect: find host " + host);
     require('dns').lookup(host, function(err, ip, addressType) {
       if (err) {
@@ -402,12 +425,13 @@ Socket.prototype.connect = function(port /* [host], [cb] */) {
       } else {
         timers.active(self);
 
-        connectip(self, port, ip || '127.0.0.1', ip ? addressType : 4);
+        connect(self, ip || '127.0.0.1', port, ip ? addressType : 4);
       }
     });
+
   } else {
     debug("connect: missing host");
-    connectip(self, port, '127.0.0.1', 4);
+    connect(self, '127.0.0.1', port, 4);
   }
 };
 
@@ -498,21 +522,19 @@ exports.Server = Server;
 function toPort(x) { return (x = Number(x)) >= 0 ? x : false; }
 
 
-function listenip(self, ip, port, addressType) {
+function listen(self, address, port, addressType) {
   var r = 0;
 
   // assign handle in listen, and clean up if bind or listen fails
-  self._handle = new TCP();
+  self._handle = (port == -1 && addressType == -1) ? new Pipe : new TCP;
   self._handle.socket = self;
   self._handle.onconnection = onconnection;
 
-  if (ip && port) {
-    debug("bind to " + ip);
-    if (addressType == 6) {
-      r = self._handle.bind6(ip, port);
-    } else {
-      r = self._handle.bind(ip, port);
-    }
+  debug("bind to " + address);
+  if (addressType == 6) {
+    r = self._handle.bind6(address, port);
+  } else {
+    r = self._handle.bind(address, port);
   }
 
   if (r) {
@@ -552,12 +574,16 @@ Server.prototype.listen = function() {
   if (arguments.length == 0 || typeof arguments[0] == 'function') {
     // Don't bind(). OS will assign a port with INADDR_ANY.
     // The port can be found with server.address()
-    listenip(self, null, null);
+    listen(self, null, null);
+
+  } else if (isPipeName(arguments[0])) {
+    // UNIX socket or Windows pipe.
+    listen(self, arguments[0], -1, -1);
 
   } else if (typeof arguments[1] == 'undefined' ||
              typeof arguments[1] == 'function') {
     // The first argument is the port, no IP given.
-    listenip(self, '0.0.0.0', port, 4);
+    listen(self, '0.0.0.0', port, 4);
 
   } else {
     // The first argument is the port, the second an IP
@@ -565,7 +591,7 @@ Server.prototype.listen = function() {
       if (err) {
         self.emit('error', err);
       } else {
-        listenip(self, ip || '0.0.0.0', port, ip ? addressType : 4);
+        listen(self, ip || '0.0.0.0', port, ip ? addressType : 4);
       }
     });
   }

diff --git a/src/pipe_wrap.cc b/src/pipe_wrap.cc
index 5d0c512..15de2a3 100644 (file)
--- a/src/pipe_wrap.cc
+++ b/src/pipe_wrap.cc
@@ -85,6 +85,7 @@ class PipeWrap : StreamWrap {
     int r = uv_pipe_init(&handle_);
     assert(r == 0); // How do we proxy this error up to javascript?
                     // Suggestion: uv_pipe_init() returns void.
+    handle_.data = reinterpret_cast<void*>(this);
     UpdateWriteQueueSize();
   }