src: refactor stream callbacks and ownership

lib/_http_server.js

@@ -666,7 +666,7 @@ function onSocketPause() {
 function unconsume(parser, socket) {
   if (socket._handle) {
     if (parser._consumed)
-      parser.unconsume(socket._handle._externalStream);
+      parser.unconsume();
     parser._consumed = false;
     socket.removeListener('pause', onSocketPause);
     socket.removeListener('resume', onSocketResume);

lib/internal/child_process.js

@@ -465,7 +465,10 @@ function setupChannel(target, channel) {
   var jsonBuffer = '';
   var pendingHandle = null;
   channel.buffering = false;
-  channel.onread = function(nread, pool, recvHandle) {
+  channel.pendingHandle = null;
+  channel.onread = function(nread, pool) {
+    const recvHandle = channel.pendingHandle;
+    channel.pendingHandle = null;
     // TODO(bnoordhuis) Check that nread > 0.
     if (pool) {
       if (recvHandle)

src/connection_wrap.cc

@@ -3,6 +3,7 @@
 #include "connect_wrap.h"
 #include "env-inl.h"
 #include "pipe_wrap.h"
+#include "stream_base-inl.h"
 #include "stream_wrap.h"
 #include "tcp_wrap.h"
 #include "util-inl.h"

src/env.h

@@ -210,6 +210,7 @@ class ModuleWrap;
   V(owner_string, "owner")                                                    \
   V(parse_error_string, "Parse Error")                                        \
   V(path_string, "path")                                                      \
+  V(pending_handle_string, "pendingHandle")                                   \
   V(pbkdf2_error_string, "PBKDF2 Error")                                      \
   V(pid_string, "pid")                                                        \
   V(pipe_string, "pipe")                                                      \

src/js_stream.cc

@@ -25,55 +25,13 @@ JSStream::JSStream(Environment* env, Local<Object> obj)
       StreamBase(env) {
   node::Wrap(obj, this);
   MakeWeak<JSStream>(this);
-
-  set_alloc_cb({ OnAllocImpl, this });
-  set_read_cb({ OnReadImpl, this });
 }
 
 
 JSStream::~JSStream() {
 }
 
 
-void JSStream::OnAllocImpl(size_t size, uv_buf_t* buf, void* ctx) {
-  buf->base = Malloc(size);
-  buf->len = size;
-}
-
-
-void JSStream::OnReadImpl(ssize_t nread,
-                          const uv_buf_t* buf,
-                          uv_handle_type pending,
-                          void* ctx) {
-  JSStream* wrap = static_cast<JSStream*>(ctx);
-  CHECK_NE(wrap, nullptr);
-  Environment* env = wrap->env();
-  HandleScope handle_scope(env->isolate());
-  Context::Scope context_scope(env->context());
-
-  if (nread < 0)  {
-    if (buf != nullptr && buf->base != nullptr)
-      free(buf->base);
-    wrap->EmitData(nread, Local<Object>(), Local<Object>());
-    return;
-  }
-
-  if (nread == 0) {
-    if (buf->base != nullptr)
-      free(buf->base);
-    return;
-  }
-
-  CHECK_LE(static_cast<size_t>(nread), buf->len);
-  char* base = node::Realloc(buf->base, nread);
-
-  CHECK_EQ(pending, UV_UNKNOWN_HANDLE);
-
-  Local<Object> obj = Buffer::New(env, base, nread).ToLocalChecked();
-  wrap->EmitData(nread, obj, Local<Object>());
-}
-
-
 AsyncWrap* JSStream::GetAsyncWrap() {
   return static_cast<AsyncWrap*>(this);
 }
@@ -212,26 +170,27 @@ void JSStream::ReadBuffer(const FunctionCallbackInfo<Value>& args) {
   char* data = Buffer::Data(args[0]);
   int len = Buffer::Length(args[0]);
 
-  do {
-    uv_buf_t buf;
+  // Repeatedly ask the stream's owner for memory, copy the data that we
+  // just read from JS into those buffers and emit them as reads.
+  while (len != 0) {
+    uv_buf_t buf = wrap->EmitAlloc(len);
     ssize_t avail = len;
-    wrap->EmitAlloc(len, &buf);
     if (static_cast<ssize_t>(buf.len) < avail)
       avail = buf.len;
 
     memcpy(buf.base, data, avail);
     data += avail;
     len -= avail;
-    wrap->EmitRead(avail, &buf);
-  } while (len != 0);
+    wrap->EmitRead(avail, buf);
+  }
 }
 
 
 void JSStream::EmitEOF(const FunctionCallbackInfo<Value>& args) {
   JSStream* wrap;
   ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
 
-  wrap->EmitRead(UV_EOF, nullptr);
+  wrap->EmitRead(UV_EOF);
 }