src: refactor stream callbacks and ownership

Instead of setting individual callbacks on streams and tracking
stream ownership through a boolean `consume_` flag, always have
one specific listener object in charge of a stream, and call
methods on that object rather than generic C-style callbacks.

PR-URL: https://github.com/nodejs/node/pull/18334
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Anatoli Papirovski <apapirovski@mac.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

1 files changed, 44 insertions, 5 deletions

diff --git a/src/stream_base.cc b/src/stream_base.cc
index 0fb801ddd57..9acf2273abd 100644
--- a/src/stream_base.cc
+++ b/src/stream_base.cc
@@ -34,12 +34,12 @@ template int StreamBase::WriteString<LATIN1>(
     const FunctionCallbackInfo<Value>& args);
 
 
-int StreamBase::ReadStart(const FunctionCallbackInfo<Value>& args) {
+int StreamBase::ReadStartJS(const FunctionCallbackInfo<Value>& args) {
   return ReadStart();
 }
 
 
-int StreamBase::ReadStop(const FunctionCallbackInfo<Value>& args) {
+int StreamBase::ReadStopJS(const FunctionCallbackInfo<Value>& args) {
   return ReadStop();
 }
 
@@ -437,9 +437,9 @@ void StreamBase::AfterWrite(WriteWrap* req_wrap, int status) {
 }
 
 
-void StreamBase::EmitData(ssize_t nread,
-                          Local<Object> buf,
-                          Local<Object> handle) {
+void StreamBase::CallJSOnreadMethod(ssize_t nread,
+                                    Local<Object> buf,
+                                    Local<Object> handle) {
   Environment* env = env_;
 
   Local<Value> argv[] = {
@@ -490,4 +490,43 @@ void StreamResource::ClearError() {
   // No-op
 }
 
+
+uv_buf_t StreamListener::OnStreamAlloc(size_t suggested_size) {
+  return uv_buf_init(Malloc(suggested_size), suggested_size);
+}
+
+void StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
+  // This cannot be virtual because it is just as valid to override the other
+  // OnStreamRead() callback.
+  CHECK(0 && "OnStreamRead() needs to be implemented");
+}
+
+void StreamListener::OnStreamRead(ssize_t nread,
+                                  const uv_buf_t& buf,
+                                  uv_handle_type pending) {
+  CHECK_EQ(pending, UV_UNKNOWN_HANDLE);
+  OnStreamRead(nread, buf);
+}
+
+
+void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
+  CHECK_NE(stream_, nullptr);
+  StreamBase* stream = static_cast<StreamBase*>(stream_);
+  Environment* env = stream->stream_env();
+  HandleScope handle_scope(env->isolate());
+  Context::Scope context_scope(env->context());
+
+  if (nread <= 0)  {
+    free(buf.base);
+    if (nread < 0)
+      stream->CallJSOnreadMethod(nread, Local<Object>());
+    return;
+  }
+
+  CHECK_LE(static_cast<size_t>(nread), buf.len);
+
+  Local<Object> obj = Buffer::New(env, buf.base, nread).ToLocalChecked();
+  stream->CallJSOnreadMethod(nread, obj);
+}
+
 }  // namespace node