#include "js_stream.h" #include "async_wrap.h" #include "env-inl.h" #include "node_buffer.h" #include "stream_base-inl.h" #include "v8.h" namespace node { using v8::Array; using v8::Context; using v8::FunctionCallbackInfo; using v8::FunctionTemplate; using v8::HandleScope; using v8::Local; using v8::MaybeLocal; using v8::Object; using v8::String; using v8::Value; JSStream::JSStream(Environment* env, Local obj) : AsyncWrap(env, obj, AsyncWrap::PROVIDER_JSSTREAM), StreamBase(env) { node::Wrap(obj, this); MakeWeak(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(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(), Local()); return; } if (nread == 0) { if (buf->base != nullptr) free(buf->base); return; } CHECK_LE(static_cast(nread), buf->len); char* base = node::Realloc(buf->base, nread); CHECK_EQ(pending, UV_UNKNOWN_HANDLE); Local obj = Buffer::New(env, base, nread).ToLocalChecked(); wrap->EmitData(nread, obj, Local()); } void* JSStream::Cast() { return static_cast(this); } AsyncWrap* JSStream::GetAsyncWrap() { return static_cast(this); } bool JSStream::IsAlive() { HandleScope scope(env()->isolate()); Context::Scope context_scope(env()->context()); v8::Local fn = object()->Get(env()->isalive_string()); if (!fn->IsFunction()) return false; return MakeCallback(fn.As(), 0, nullptr) .ToLocalChecked()->IsTrue(); } bool JSStream::IsClosing() { HandleScope scope(env()->isolate()); Context::Scope context_scope(env()->context()); return MakeCallback(env()->isclosing_string(), 0, nullptr) .ToLocalChecked()->IsTrue(); } int JSStream::ReadStart() { HandleScope scope(env()->isolate()); Context::Scope context_scope(env()->context()); return MakeCallback(env()->onreadstart_string(), 0, nullptr) .ToLocalChecked()->Int32Value(); } int JSStream::ReadStop() { HandleScope scope(env()->isolate()); Context::Scope context_scope(env()->context()); return MakeCallback(env()->onreadstop_string(), 0, nullptr) .ToLocalChecked()->Int32Value(); } int JSStream::DoShutdown(ShutdownWrap* req_wrap) { HandleScope scope(env()->isolate()); Context::Scope context_scope(env()->context()); Local argv[] = { req_wrap->object() }; req_wrap->Dispatched(); MaybeLocal res = MakeCallback(env()->onshutdown_string(), arraysize(argv), argv); return res.ToLocalChecked()->Int32Value(); } int JSStream::DoWrite(WriteWrap* w, uv_buf_t* bufs, size_t count, uv_stream_t* send_handle) { CHECK_EQ(send_handle, nullptr); HandleScope scope(env()->isolate()); Context::Scope context_scope(env()->context()); Local bufs_arr = Array::New(env()->isolate(), count); Local buf; for (size_t i = 0; i < count; i++) { buf = Buffer::Copy(env(), bufs[i].base, bufs[i].len).ToLocalChecked(); bufs_arr->Set(i, buf); } Local argv[] = { w->object(), bufs_arr }; w->Dispatched(); MaybeLocal res = MakeCallback(env()->onwrite_string(), arraysize(argv), argv); return res.ToLocalChecked()->Int32Value(); } void JSStream::New(const FunctionCallbackInfo& args) { // This constructor should not be exposed to public javascript. // Therefore we assert that we are not trying to call this as a // normal function. CHECK(args.IsConstructCall()); Environment* env = Environment::GetCurrent(args); new JSStream(env, args.This()); } void JSStream::DoAfterWrite(const FunctionCallbackInfo& args) { JSStream* wrap; CHECK(args[0]->IsObject()); WriteWrap* w; ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder()); ASSIGN_OR_RETURN_UNWRAP(&w, args[0].As()); wrap->OnAfterWrite(w); } template void JSStream::Finish(const FunctionCallbackInfo& args) { Wrap* w; CHECK(args[0]->IsObject()); ASSIGN_OR_RETURN_UNWRAP(&w, args[0].As()); w->Done(args[1]->Int32Value()); } void JSStream::ReadBuffer(const FunctionCallbackInfo& args) { JSStream* wrap; ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder()); CHECK(Buffer::HasInstance(args[0])); char* data = Buffer::Data(args[0]); int len = Buffer::Length(args[0]); do { uv_buf_t buf; ssize_t avail = len; wrap->OnAlloc(len, &buf); if (static_cast(buf.len) < avail) avail = buf.len; memcpy(buf.base, data, avail); data += avail; len -= avail; wrap->OnRead(avail, &buf); } while (len != 0); } void JSStream::EmitEOF(const FunctionCallbackInfo& args) { JSStream* wrap; ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder()); wrap->OnRead(UV_EOF, nullptr); } void JSStream::Initialize(Local target, Local unused, Local context) { Environment* env = Environment::GetCurrent(context); Local t = env->NewFunctionTemplate(New); Local jsStreamString = FIXED_ONE_BYTE_STRING(env->isolate(), "JSStream"); t->SetClassName(jsStreamString); t->InstanceTemplate()->SetInternalFieldCount(1); AsyncWrap::AddWrapMethods(env, t); env->SetProtoMethod(t, "doAfterWrite", DoAfterWrite); env->SetProtoMethod(t, "finishWrite", Finish); env->SetProtoMethod(t, "finishShutdown", Finish); env->SetProtoMethod(t, "readBuffer", ReadBuffer); env->SetProtoMethod(t, "emitEOF", EmitEOF); StreamBase::AddMethods(env, t, StreamBase::kFlagHasWritev); target->Set(jsStreamString, t->GetFunction()); } } // namespace node NODE_BUILTIN_MODULE_CONTEXT_AWARE(js_stream, node::JSStream::Initialize)