diff options
author | Brian White <mscdex@mscdex.net> | 2019-01-10 23:52:27 +0300 |
---|---|---|
committer | Brian White <mscdex@mscdex.net> | 2019-08-24 00:05:52 +0300 |
commit | 8292b280ec9e6b8c2444cbe49350facc77f5fefa (patch) | |
tree | 2f40ac41f20b53b748a7039c4f00e419efb8784b /src/stream_base.cc | |
parent | 9d21b0395cc248a0e5537a11cc84f61919eccca6 (diff) |
net: allow reading data into a static buffer
Co-Authored-By: Anna Henningsen <anna@addaleax.net>
PR-URL: https://github.com/nodejs/node/pull/25436
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Ben Noordhuis <info@bnoordhuis.nl>
Diffstat (limited to 'src/stream_base.cc')
-rw-r--r-- | src/stream_base.cc | 59 |
1 files changed, 50 insertions, 9 deletions
diff --git a/src/stream_base.cc b/src/stream_base.cc index 695d19c123e..52163e2e43f 100644 --- a/src/stream_base.cc +++ b/src/stream_base.cc @@ -26,6 +26,7 @@ using v8::FunctionCallbackInfo; using v8::HandleScope; using v8::Integer; using v8::Local; +using v8::MaybeLocal; using v8::Object; using v8::ReadOnly; using v8::String; @@ -50,6 +51,13 @@ int StreamBase::ReadStopJS(const FunctionCallbackInfo<Value>& args) { return ReadStop(); } +int StreamBase::UseUserBuffer(const FunctionCallbackInfo<Value>& args) { + CHECK(Buffer::HasInstance(args[0])); + + uv_buf_t buf = uv_buf_init(Buffer::Data(args[0]), Buffer::Length(args[0])); + PushStreamListener(new CustomBufferJSListener(buf)); + return 0; +} int StreamBase::Shutdown(const FunctionCallbackInfo<Value>& args) { CHECK(args[0]->IsObject()); @@ -291,19 +299,22 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) { } -void StreamBase::CallJSOnreadMethod(ssize_t nread, - Local<ArrayBuffer> ab, - size_t offset) { +MaybeLocal<Value> StreamBase::CallJSOnreadMethod(ssize_t nread, + Local<ArrayBuffer> ab, + size_t offset, + StreamBaseJSChecks checks) { Environment* env = env_; DCHECK_EQ(static_cast<int32_t>(nread), nread); DCHECK_LE(offset, INT32_MAX); - if (ab.IsEmpty()) { - DCHECK_EQ(offset, 0); - DCHECK_LE(nread, 0); - } else { - DCHECK_GE(nread, 0); + if (checks == DONT_SKIP_NREAD_CHECKS) { + if (ab.IsEmpty()) { + DCHECK_EQ(offset, 0); + DCHECK_LE(nread, 0); + } else { + DCHECK_GE(nread, 0); + } } env->stream_base_state()[kReadBytesOrError] = nread; @@ -317,7 +328,7 @@ void StreamBase::CallJSOnreadMethod(ssize_t nread, CHECK_NOT_NULL(wrap); Local<Value> onread = wrap->object()->GetInternalField(kOnReadFunctionField); CHECK(onread->IsFunction()); - wrap->MakeCallback(onread.As<Function>(), arraysize(argv), argv); + return wrap->MakeCallback(onread.As<Function>(), arraysize(argv), argv); } @@ -366,6 +377,9 @@ void StreamBase::AddMethods(Environment* env, Local<FunctionTemplate> t) { env->SetProtoMethod(t, "readStart", JSMethod<&StreamBase::ReadStartJS>); env->SetProtoMethod(t, "readStop", JSMethod<&StreamBase::ReadStopJS>); env->SetProtoMethod(t, "shutdown", JSMethod<&StreamBase::Shutdown>); + env->SetProtoMethod(t, + "useUserBuffer", + JSMethod<&StreamBase::UseUserBuffer>); env->SetProtoMethod(t, "writev", JSMethod<&StreamBase::Writev>); env->SetProtoMethod(t, "writeBuffer", JSMethod<&StreamBase::WriteBuffer>); env->SetProtoMethod( @@ -445,6 +459,7 @@ void StreamResource::ClearError() { // No-op } + uv_buf_t EmitToJSStreamListener::OnStreamAlloc(size_t suggested_size) { CHECK_NOT_NULL(stream_); Environment* env = static_cast<StreamBase*>(stream_)->stream_env(); @@ -472,6 +487,32 @@ void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { } +uv_buf_t CustomBufferJSListener::OnStreamAlloc(size_t suggested_size) { + return buffer_; +} + + +void CustomBufferJSListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) { + CHECK_NOT_NULL(stream_); + CHECK_EQ(buf.base, buffer_.base); + + StreamBase* stream = static_cast<StreamBase*>(stream_); + Environment* env = stream->stream_env(); + HandleScope handle_scope(env->isolate()); + Context::Scope context_scope(env->context()); + + MaybeLocal<Value> ret = stream->CallJSOnreadMethod(nread, + Local<ArrayBuffer>(), + 0, + StreamBase::SKIP_NREAD_CHECKS); + Local<Value> next_buf_v; + if (ret.ToLocal(&next_buf_v) && !next_buf_v->IsUndefined()) { + buffer_.base = Buffer::Data(next_buf_v); + buffer_.len = Buffer::Length(next_buf_v); + } +} + + void ReportWritesToJSStreamListener::OnStreamAfterReqFinished( StreamReq* req_wrap, int status) { StreamBase* stream = static_cast<StreamBase*>(stream_); |