cluster: support passing of named pipes

Fixes triggered assertion:

  Assertion failed: (0 && "bad address family"), function GetPeerName,
  file ../src/tcp_wrap.cc, line 237.

Fixes #2870.

src/pipe_wrap.cc

@@ -67,6 +67,13 @@ uv_pipe_t* PipeWrap::UVHandle() {
 }
 
 
+Local<Object> PipeWrap::Instantiate() {
+  HandleScope scope;
+  assert(!pipeConstructor.IsEmpty());
+  return scope.Close(pipeConstructor->NewInstance());
+}
+
+
 PipeWrap* PipeWrap::Unwrap(Local<Object> obj) {
   assert(!obj.IsEmpty());
   assert(obj->InternalFieldCount() > 0);

src/pipe_wrap.h

@@ -29,6 +29,7 @@ class PipeWrap : StreamWrap {
  public:
   uv_pipe_t* UVHandle();
 
+  static v8::Local<v8::Object> Instantiate();
   static PipeWrap* Unwrap(v8::Local<v8::Object> obj);
   static void Initialize(v8::Handle<v8::Object> target);
 

src/stream_wrap.cc

@@ -23,9 +23,12 @@
 #include <node_buffer.h>
 #include <handle_wrap.h>
 #include <stream_wrap.h>
+#include <pipe_wrap.h>
 #include <tcp_wrap.h>
 #include <req_wrap.h>
 
+#include <stdlib.h> // abort()
+
 
 namespace node {
 
@@ -237,36 +240,35 @@ void StreamWrap::OnReadCommon(uv_stream_t* handle, ssize_t nread,
     slab_used -= (buf.len - nread);
   }
 
-  if (nread > 0) {
-    int argc = 3;
-    Local<Value> argv[4] = {
-      slab_v,
-      Integer::New(wrap->slab_offset_),
-      Integer::New(nread)
-    };
+  if (nread == 0) return;
 
+  int argc = 3;
+  Local<Value> argv[4] = {
+    slab_v,
+    Integer::New(wrap->slab_offset_),
+    Integer::New(nread)
+  };
 
-    if (pending == UV_TCP) {
-      // Instantiate the client javascript object and handle.
-      Local<Object> pending_obj = TCPWrap::Instantiate();
-
-      // Unwrap the client javascript object.
-      assert(pending_obj->InternalFieldCount() > 0);
-      TCPWrap* pending_wrap =
-          static_cast<TCPWrap*>(pending_obj->GetPointerFromInternalField(0));
-
-      int r = uv_accept(handle, pending_wrap->GetStream());
-      assert(r == 0);
-
-      argv[3] = pending_obj;
-      argc++;
-    } else {
-      // We only support sending UV_TCP right now.
-      assert(pending == UV_UNKNOWN_HANDLE);
-    }
-
-    MakeCallback(wrap->object_, "onread", argc, argv);
+  Local<Object> pending_obj;
+  if (pending == UV_TCP) {
+    pending_obj = TCPWrap::Instantiate();
+  } else if (pending == UV_NAMED_PIPE) {
+    pending_obj = PipeWrap::Instantiate();
+  } else {
+    // We only support sending UV_TCP and UV_NAMED_PIPE right now.
+    assert(pending == UV_UNKNOWN_HANDLE);
   }
+
+  if (!pending_obj.IsEmpty()) {
+    assert(pending_obj->InternalFieldCount() > 0);
+    StreamWrap* pending_wrap =
+      static_cast<StreamWrap*>(pending_obj->GetPointerFromInternalField(0));
+    if (uv_accept(handle, pending_wrap->GetStream())) abort();
+    argv[3] = pending_obj;
+    argc++;
+  }
+
+  MakeCallback(wrap->object_, "onread", argc, argv);
 }
 
 

test/simple/test-cluster-http-pipe.js

@@ -0,0 +1,57 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+var common = require('../common');
+var assert = require('assert');
+var cluster = require('cluster');
+var http = require('http');
+
+if (cluster.isMaster) {
+  var ok = false;
+  var worker = cluster.fork();
+  worker.on('message', function(msg) {
+    assert.equal(msg, 'DONE');
+    ok = true;
+  });
+  worker.on('death', function() {
+    process.exit();
+  });
+  process.on('exit', function() {
+    assert(ok);
+  });
+  return;
+}
+
+http.createServer(function(req, res) {
+  assert.equal(req.connection.remoteAddress, undefined);
+  assert.equal(req.connection.localAddress, undefined); // TODO common.PIPE?
+  res.writeHead(200);
+  res.end('OK');
+}).listen(common.PIPE, function() {
+  var self = this;
+  http.get({ socketPath: common.PIPE, path: '/' }, function(res) {
+    res.on('end', function(err) {
+      if (err) throw err;
+      process.send('DONE');
+      process.exit();
+    });
+  });
+});