Skip to content
Browse files

WIP

  • Loading branch information...
1 parent 31ad1d2 commit 33641d06a72c0de4403fd6c6585bbfaf147205f2 @bnoordhuis committed Mar 9, 2012
Showing with 125 additions and 32 deletions.
  1. +32 −6 deps/uv/src/unix/stream.c
  2. +7 −0 src/pipe_wrap.cc
  3. +1 −0 src/pipe_wrap.h
  4. +28 −26 src/stream_wrap.cc
  5. +57 −0 test/simple/test-cluster-http-pipe.js
View
38 deps/uv/src/unix/stream.c
@@ -22,14 +22,17 @@
#include "uv.h"
#include "internal.h"
-#include <assert.h>
-#include <errno.h>
-#include <unistd.h>
+#include <stdio.h>
#include <stdlib.h>
#include <string.h>
-#include <sys/uio.h>
+#include <assert.h>
+#include <errno.h>
-#include <stdio.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+#include <sys/un.h>
+#include <unistd.h>
static void uv__stream_connect(uv_stream_t*);
@@ -513,6 +516,28 @@ static void uv__write_callbacks(uv_stream_t* stream) {
}
+static uv_handle_type uv__handle_type(int fd) {
+ struct sockaddr_storage ss;
+ socklen_t len;
+
+ memset(&ss, 0, sizeof(ss));
+ len = sizeof(ss);
+
+ if (getsockname(fd, (struct sockaddr*)&ss, &len))
+ return UV_UNKNOWN_HANDLE;
+
+ switch (ss.ss_family) {
+ case AF_UNIX:
+ return UV_NAMED_PIPE;
+ case AF_INET:
+ case AF_INET6:
+ return UV_TCP;
+ }
+
+ return UV_UNKNOWN_HANDLE;
+}
+
+
static void uv__read(uv_stream_t* stream) {
uv_buf_t buf;
ssize_t nread;
@@ -633,7 +658,8 @@ static void uv__read(uv_stream_t* stream) {
if (stream->accepted_fd >= 0) {
- stream->read2_cb((uv_pipe_t*)stream, nread, buf, UV_TCP);
+ stream->read2_cb((uv_pipe_t*)stream, nread, buf,
+ uv__handle_type(stream->accepted_fd));
} else {
stream->read2_cb((uv_pipe_t*)stream, nread, buf, UV_UNKNOWN_HANDLE);
}
View
7 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);
View
1 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);
View
54 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 (pending == UV_TCP) {
- // Instantiate the client javascript object and handle.
- Local<Object> pending_obj = TCPWrap::Instantiate();
+ if (nread == 0) return;
- // 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);
+ int argc = 3;
+ Local<Value> argv[4] = {
+ slab_v,
+ Integer::New(wrap->slab_offset_),
+ Integer::New(nread)
+ };
- argv[3] = pending_obj;
- argc++;
- } else {
- // We only support sending UV_TCP right now.
- assert(pending == UV_UNKNOWN_HANDLE);
- }
+ 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);
+ }
- MakeCallback(wrap->object_, "onread", argc, argv);
+ 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);
}
View
57 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();
+ });
+ });
+});

0 comments on commit 33641d0

Please sign in to comment.
Something went wrong with that request. Please try again.