diff options
author | Anna Henningsen <anna@addaleax.net> | 2018-01-08 03:14:06 +0300 |
---|---|---|
committer | Ruben Bridgewater <ruben@bridgewater.de> | 2018-02-01 12:53:26 +0300 |
commit | 7c4b09b24bbe7d6a8cbad256f47b30a101a909ea (patch) | |
tree | 1aef41b1fd1cc0aad300b178e0a19e6da29615c8 /src/stream_base.cc | |
parent | 1b6cb947611de5865641d1a6780ee6930a4e1d69 (diff) |
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>
Diffstat (limited to 'src/stream_base.cc')
-rw-r--r-- | src/stream_base.cc | 49 |
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 |