#ifndef SRC_STREAM_BASE_INL_H_ #define SRC_STREAM_BASE_INL_H_ #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS #include "stream_base.h" #include "node.h" #include "env-inl.h" #include "v8.h" namespace node { using v8::Signature; using v8::External; using v8::FunctionCallbackInfo; using v8::FunctionTemplate; using v8::HandleScope; using v8::Local; using v8::Object; using v8::PropertyAttribute; using v8::PropertyCallbackInfo; using v8::String; using v8::Value; using AsyncHooks = Environment::AsyncHooks; inline void StreamReq::AttachToObject(v8::Local req_wrap_obj) { CHECK_EQ(req_wrap_obj->GetAlignedPointerFromInternalField(kStreamReqField), nullptr); req_wrap_obj->SetAlignedPointerInInternalField(kStreamReqField, this); } inline StreamReq* StreamReq::FromObject(v8::Local req_wrap_obj) { return static_cast( req_wrap_obj->GetAlignedPointerFromInternalField(kStreamReqField)); } inline void StreamReq::Dispose() { object()->SetAlignedPointerInInternalField(kStreamReqField, nullptr); delete this; } inline v8::Local StreamReq::object() { return GetAsyncWrap()->object(); } inline StreamListener::~StreamListener() { if (stream_ != nullptr) stream_->RemoveStreamListener(this); } inline void StreamListener::PassReadErrorToPreviousListener(ssize_t nread) { CHECK_NE(previous_listener_, nullptr); previous_listener_->OnStreamRead(nread, uv_buf_init(nullptr, 0)); } inline void StreamListener::OnStreamAfterShutdown(ShutdownWrap* w, int status) { CHECK_NE(previous_listener_, nullptr); previous_listener_->OnStreamAfterShutdown(w, status); } inline void StreamListener::OnStreamAfterWrite(WriteWrap* w, int status) { CHECK_NE(previous_listener_, nullptr); previous_listener_->OnStreamAfterWrite(w, status); } inline StreamResource::~StreamResource() { while (listener_ != nullptr) { listener_->OnStreamDestroy(); RemoveStreamListener(listener_); } } inline void StreamResource::PushStreamListener(StreamListener* listener) { CHECK_NE(listener, nullptr); CHECK_EQ(listener->stream_, nullptr); listener->previous_listener_ = listener_; listener->stream_ = this; listener_ = listener; } inline void StreamResource::RemoveStreamListener(StreamListener* listener) { CHECK_NE(listener, nullptr); StreamListener* previous; StreamListener* current; // Remove from the linked list. for (current = listener_, previous = nullptr; /* No loop condition because we want a crash if listener is not found */ ; previous = current, current = current->previous_listener_) { CHECK_NE(current, nullptr); if (current == listener) { if (previous != nullptr) previous->previous_listener_ = current->previous_listener_; else listener_ = listener->previous_listener_; break; } } listener->stream_ = nullptr; listener->previous_listener_ = nullptr; } inline uv_buf_t StreamResource::EmitAlloc(size_t suggested_size) { return listener_->OnStreamAlloc(suggested_size); } inline void StreamResource::EmitRead(ssize_t nread, const uv_buf_t& buf) { if (nread > 0) bytes_read_ += static_cast(nread); listener_->OnStreamRead(nread, buf); } inline void StreamResource::EmitAfterWrite(WriteWrap* w, int status) { listener_->OnStreamAfterWrite(w, status); } inline void StreamResource::EmitAfterShutdown(ShutdownWrap* w, int status) { listener_->OnStreamAfterShutdown(w, status); } inline StreamBase::StreamBase(Environment* env) : env_(env) { PushStreamListener(&default_listener_); } inline Environment* StreamBase::stream_env() const { return env_; } inline void StreamBase::AfterWrite(WriteWrap* req_wrap, int status) { AfterRequest(req_wrap, [&]() { EmitAfterWrite(req_wrap, status); }); } inline void StreamBase::AfterShutdown(ShutdownWrap* req_wrap, int status) { AfterRequest(req_wrap, [&]() { EmitAfterShutdown(req_wrap, status); }); } template inline void StreamBase::AfterRequest(Wrap* req_wrap, EmitEvent emit) { Environment* env = stream_env(); v8::HandleScope handle_scope(env->isolate()); v8::Context::Scope context_scope(env->context()); emit(); req_wrap->Dispose(); } inline int StreamBase::Shutdown(v8::Local req_wrap_obj) { Environment* env = stream_env(); if (req_wrap_obj.IsEmpty()) { req_wrap_obj = env->shutdown_wrap_constructor_function() ->NewInstance(env->context()).ToLocalChecked(); } AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope( env, GetAsyncWrap()->get_async_id()); ShutdownWrap* req_wrap = CreateShutdownWrap(req_wrap_obj); int err = DoShutdown(req_wrap); if (err != 0) { req_wrap->Dispose(); } const char* msg = Error(); if (msg != nullptr) { req_wrap_obj->Set(env->error_string(), OneByteString(env->isolate(), msg)); ClearError(); } return err; } inline StreamWriteResult StreamBase::Write( uv_buf_t* bufs, size_t count, uv_stream_t* send_handle, v8::Local req_wrap_obj) { Environment* env = stream_env(); int err; if (send_handle == nullptr) { err = DoTryWrite(&bufs, &count); if (err != 0 || count == 0) { return StreamWriteResult { false, err, nullptr }; } } if (req_wrap_obj.IsEmpty()) { req_wrap_obj = env->write_wrap_constructor_function() ->NewInstance(env->context()).ToLocalChecked(); } AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope( env, GetAsyncWrap()->get_async_id()); WriteWrap* req_wrap = CreateWriteWrap(req_wrap_obj); err = DoWrite(req_wrap, bufs, count, send_handle); bool async = err == 0; if (!async) { req_wrap->Dispose(); req_wrap = nullptr; } const char* msg = Error(); if (msg != nullptr) { req_wrap_obj->Set(env->error_string(), OneByteString(env->isolate(), msg)); ClearError(); } req_wrap_obj->Set(env->async(), v8::Boolean::New(env->isolate(), async)); return StreamWriteResult { async, err, req_wrap }; } template SimpleShutdownWrap::SimpleShutdownWrap( StreamBase* stream, v8::Local req_wrap_obj) : ShutdownWrap(stream, req_wrap_obj), OtherBase(stream->stream_env(), req_wrap_obj, AsyncWrap::PROVIDER_SHUTDOWNWRAP) { Wrap(req_wrap_obj, static_cast(this)); } template SimpleShutdownWrap::~SimpleShutdownWrap() { ClearWrap(static_cast(this)->object()); } inline ShutdownWrap* StreamBase::CreateShutdownWrap( v8::Local object) { return new SimpleShutdownWrap(this, object); } template SimpleWriteWrap::SimpleWriteWrap( StreamBase* stream, v8::Local req_wrap_obj) : WriteWrap(stream, req_wrap_obj), OtherBase(stream->stream_env(), req_wrap_obj, AsyncWrap::PROVIDER_WRITEWRAP) { Wrap(req_wrap_obj, static_cast(this)); } template SimpleWriteWrap::~SimpleWriteWrap() { ClearWrap(static_cast(this)->object()); } inline WriteWrap* StreamBase::CreateWriteWrap( v8::Local object) { return new SimpleWriteWrap(this, object); } template void StreamBase::AddMethods(Environment* env, Local t, int flags) { HandleScope scope(env->isolate()); enum PropertyAttribute attributes = static_cast( v8::ReadOnly | v8::DontDelete | v8::DontEnum); Local signature = Signature::New(env->isolate(), t); Local get_fd_templ = FunctionTemplate::New(env->isolate(), GetFD, env->as_external(), signature); Local get_external_templ = FunctionTemplate::New(env->isolate(), GetExternal, env->as_external(), signature); Local get_bytes_read_templ = FunctionTemplate::New(env->isolate(), GetBytesRead, env->as_external(), signature); t->PrototypeTemplate()->SetAccessorProperty(env->fd_string(), get_fd_templ, Local(), attributes); t->PrototypeTemplate()->SetAccessorProperty(env->external_stream_string(), get_external_templ, Local(), attributes); t->PrototypeTemplate()->SetAccessorProperty(env->bytes_read_string(), get_bytes_read_templ, Local(), attributes); env->SetProtoMethod(t, "readStart", JSMethod); env->SetProtoMethod(t, "readStop", JSMethod); if ((flags & kFlagNoShutdown) == 0) env->SetProtoMethod(t, "shutdown", JSMethod); if ((flags & kFlagHasWritev) != 0) env->SetProtoMethod(t, "writev", JSMethod); env->SetProtoMethod(t, "writeBuffer", JSMethod); env->SetProtoMethod(t, "writeAsciiString", JSMethod >); env->SetProtoMethod(t, "writeUtf8String", JSMethod >); env->SetProtoMethod(t, "writeUcs2String", JSMethod >); env->SetProtoMethod(t, "writeLatin1String", JSMethod >); } template void StreamBase::GetFD(const FunctionCallbackInfo& args) { // Mimic implementation of StreamBase::GetFD() and UDPWrap::GetFD(). Base* handle; ASSIGN_OR_RETURN_UNWRAP(&handle, args.This(), args.GetReturnValue().Set(UV_EINVAL)); StreamBase* wrap = static_cast(handle); if (!wrap->IsAlive()) return args.GetReturnValue().Set(UV_EINVAL); args.GetReturnValue().Set(wrap->GetFD()); } template void StreamBase::GetBytesRead(const FunctionCallbackInfo& args) { // The handle instance hasn't been set. So no bytes could have been read. Base* handle; ASSIGN_OR_RETURN_UNWRAP(&handle, args.This(), args.GetReturnValue().Set(0)); StreamBase* wrap = static_cast(handle); // uint64_t -> double. 53bits is enough for all real cases. args.GetReturnValue().Set(static_cast(wrap->bytes_read_)); } template void StreamBase::GetExternal(const FunctionCallbackInfo& args) { Base* handle; ASSIGN_OR_RETURN_UNWRAP(&handle, args.This()); StreamBase* wrap = static_cast(handle); Local ext = External::New(args.GetIsolate(), wrap); args.GetReturnValue().Set(ext); } template & args)> void StreamBase::JSMethod(const FunctionCallbackInfo& args) { Base* handle; ASSIGN_OR_RETURN_UNWRAP(&handle, args.Holder()); StreamBase* wrap = static_cast(handle); if (!wrap->IsAlive()) return args.GetReturnValue().Set(UV_EINVAL); AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope( handle->env(), handle->get_async_id()); args.GetReturnValue().Set((wrap->*Method)(args)); } inline void ShutdownWrap::OnDone(int status) { stream()->AfterShutdown(this, status); } inline void WriteWrap::SetAllocatedStorage(char* data, size_t size) { CHECK_EQ(storage_, nullptr); storage_ = data; storage_size_ = size; } inline char* WriteWrap::Storage() { return storage_; } inline size_t WriteWrap::StorageSize() const { return storage_size_; } inline void WriteWrap::OnDone(int status) { stream()->AfterWrite(this, status); } inline void StreamReq::Done(int status, const char* error_str) { AsyncWrap* async_wrap = GetAsyncWrap(); Environment* env = async_wrap->env(); if (error_str != nullptr) { async_wrap->object()->Set(env->error_string(), OneByteString(env->isolate(), error_str)); } OnDone(status); } } // namespace node #endif // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS #endif // SRC_STREAM_BASE_INL_H_