text: #include "stream_wrap.h"
  #include "stream_base.h"
  #include "stream_base-inl.h"
  
  #include "env-inl.h"
  #include "env.h"
  #include "handle_wrap.h"
  #include "node_buffer.h"
  #include "node_counters.h"
  #include "pipe_wrap.h"
  #include "req-wrap.h"
  #include "req-wrap-inl.h"
  #include "tcp_wrap.h"
  #include "udp_wrap.h"
  #include "util.h"
  #include "util-inl.h"
  
  #include <stdlib.h>  // abort()
  #include <string.h>  // memcpy()
  #include <limits.h>  // INT_MAX
  
  
  namespace node {
  
  using v8::Array;
  using v8::Context;
  using v8::EscapableHandleScope;
  using v8::FunctionCallbackInfo;
  using v8::FunctionTemplate;
  using v8::Handle;
  using v8::HandleScope;
  using v8::Integer;
  using v8::Local;
  using v8::Number;
  using v8::Object;
  using v8::PropertyCallbackInfo;
  using v8::String;
  using v8::True;
  using v8::Undefined;
  using v8::Value;
  
  
  void StreamWrap::Initialize(Handle<Object> target,
                              Handle<Value> unused,
                              Handle<Context> context) {
    Environment* env = Environment::GetCurrent(context);
  
    Local<FunctionTemplate> sw =
        FunctionTemplate::New(env->isolate(), ShutdownWrap::NewShutdownWrap);
    sw->InstanceTemplate()->SetInternalFieldCount(1);
    sw->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), "ShutdownWrap"));
    target->Set(FIXED_ONE_BYTE_STRING(env->isolate(), "ShutdownWrap"),
                sw->GetFunction());
  
    Local<FunctionTemplate> ww =
        FunctionTemplate::New(env->isolate(), WriteWrap::NewWriteWrap);
    ww->InstanceTemplate()->SetInternalFieldCount(1);
    ww->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), "WriteWrap"));
    target->Set(FIXED_ONE_BYTE_STRING(env->isolate(), "WriteWrap"),
                ww->GetFunction());
    env->set_write_wrap_constructor_function(ww->GetFunction());
  }
  
  
  StreamWrap::StreamWrap(Environment* env,
                         Local<Object> object,
                         uv_stream_t* stream,
                         AsyncWrap::ProviderType provider,
                         AsyncWrap* parent)
      : HandleWrap(env,
                   object,
                   reinterpret_cast<uv_handle_t*>(stream),
                   provider,
                   parent),
        StreamBase(env),
        stream_(stream) {
    set_after_write_cb(OnAfterWriteImpl, this);
    set_alloc_cb(OnAllocImpl, this);
    set_read_cb(OnReadImpl, this);
  }
  
  
  void StreamWrap::AddMethods(Environment* env,
                              v8::Handle<v8::FunctionTemplate> target,
                              int flags) {
    env->SetProtoMethod(target, "setBlocking", SetBlocking);
    StreamBase::AddMethods<StreamWrap>(env, target, flags);
  }
  
  
  int StreamWrap::GetFD() {
    int fd = -1;
  #if !defined(_WIN32)
    if (stream() != nullptr)
      fd = stream()->io_watcher.fd;
  #endif
    return fd;
  }
  
  
  bool StreamWrap::IsAlive() {
    return HandleWrap::IsAlive(this);
  }
  
  
  bool StreamWrap::IsClosing() {
    return uv_is_closing(reinterpret_cast<uv_handle_t*>(stream()));
  }
  
  
  void* StreamWrap::Cast() {
    return reinterpret_cast<void*>(this);
  }
  
  
  AsyncWrap* StreamWrap::GetAsyncWrap() {
    return static_cast<AsyncWrap*>(this);
  }
  
  
  bool StreamWrap::IsIPCPipe() {
    return is_named_pipe_ipc();
  }
  
  
  void StreamWrap::UpdateWriteQueueSize() {
    HandleScope scope(env()->isolate());
    Local<Integer> write_queue_size =
        Integer::NewFromUnsigned(env()->isolate(), stream()->write_queue_size);
    object()->Set(env()->write_queue_size_string(), write_queue_size);
  }
  
  
  int StreamWrap::ReadStart() {
    return uv_read_start(stream(), OnAlloc, OnRead);
  }
  
  
  int StreamWrap::ReadStop() {
    return uv_read_stop(stream());
  }
  
  
  void StreamWrap::OnAlloc(uv_handle_t* handle,
                           size_t suggested_size,
                           uv_buf_t* buf) {
    StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
    HandleScope scope(wrap->env()->isolate());
    Context::Scope context_scope(wrap->env()->context());
  
    CHECK_EQ(wrap->stream(), reinterpret_cast<uv_stream_t*>(handle));
  
    return static_cast<StreamBase*>(wrap)->OnAlloc(suggested_size, buf);
  }
  
  
  void StreamWrap::OnAllocImpl(size_t size, uv_buf_t* buf, void* ctx) {
    buf->base = static_cast<char*>(malloc(size));
    buf->len = size;
  
    if (buf->base == nullptr && size > 0) {
      FatalError(
          "node::StreamWrap::DoAlloc(size_t, uv_buf_t*, void*)",
          "Out Of Memory");
    }
  }
  
  
  template <class WrapType, class UVType>
  static Local<Object> AcceptHandle(Environment* env, StreamWrap* parent) {
    EscapableHandleScope scope(env->isolate());
    Local<Object> wrap_obj;
    UVType* handle;
  
    wrap_obj = WrapType::Instantiate(env, parent);
    if (wrap_obj.IsEmpty())
      return Local<Object>();
  
    WrapType* wrap = Unwrap<WrapType>(wrap_obj);
    handle = wrap->UVHandle();
  
    if (uv_accept(parent->stream(), reinterpret_cast<uv_stream_t*>(handle)))
      abort();
  
    return scope.Escape(wrap_obj);
  }
  
  
  void StreamWrap::OnReadImpl(ssize_t nread,
                              const uv_buf_t* buf,
                              uv_handle_type pending,
                              void* ctx) {
    StreamWrap* wrap = static_cast<StreamWrap*>(ctx);
    Environment* env = wrap->env();
    HandleScope handle_scope(env->isolate());
    Context::Scope context_scope(env->context());
  
    Local<Object> pending_obj;
  
    if (nread < 0)  {
      if (buf->base != nullptr)
        free(buf->base);
      wrap->EmitData(nread, Local<Object>(), pending_obj);
      return;
    }
  
    if (nread == 0) {
      if (buf->base != nullptr)
        free(buf->base);
      return;
    }
  
    char* base = static_cast<char*>(realloc(buf->base, nread));
    CHECK_LE(static_cast<size_t>(nread), buf->len);
  
    if (pending == UV_TCP) {
      pending_obj = AcceptHandle<TCPWrap, uv_tcp_t>(env, wrap);
    } else if (pending == UV_NAMED_PIPE) {
      pending_obj = AcceptHandle<PipeWrap, uv_pipe_t>(env, wrap);
    } else if (pending == UV_UDP) {
      pending_obj = AcceptHandle<UDPWrap, uv_udp_t>(env, wrap);
    } else {
      CHECK_EQ(pending, UV_UNKNOWN_HANDLE);
    }
  
    wrap->EmitData(nread, Buffer::Use(env, base, nread), pending_obj);
  }
  
  
  void StreamWrap::OnReadCommon(uv_stream_t* handle,
                                ssize_t nread,
                                const uv_buf_t* buf,
                                uv_handle_type pending) {
    StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
    HandleScope scope(wrap->env()->isolate());
  
    // We should not be getting this callback if someone as already called
    // uv_close() on the handle.
    CHECK_EQ(wrap->persistent().IsEmpty(), false);
  
    if (nread > 0) {
      if (wrap->is_tcp()) {
        NODE_COUNT_NET_BYTES_RECV(nread);
      } else if (wrap->is_named_pipe()) {
        NODE_COUNT_PIPE_BYTES_RECV(nread);
      }
    }
  
    static_cast<StreamBase*>(wrap)->OnRead(nread, buf, pending);
  }
  
  
  void StreamWrap::OnRead(uv_stream_t* handle,
                          ssize_t nread,
                          const uv_buf_t* buf) {
    StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
    uv_handle_type type = UV_UNKNOWN_HANDLE;
  
    if (wrap->is_named_pipe_ipc() &&
        uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(handle)) > 0) {
      type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(handle));
    }
  
    OnReadCommon(handle, nread, buf, type);
  }
  
  
  void StreamWrap::SetBlocking(const FunctionCallbackInfo<Value>& args) {
    StreamWrap* wrap = Unwrap<StreamWrap>(args.Holder());
  
    CHECK_GT(args.Length(), 0);
    if (!wrap->IsAlive())
      return args.GetReturnValue().Set(UV_EINVAL);
  
    bool enable = args[0]->IsTrue();
    args.GetReturnValue().Set(uv_stream_set_blocking(wrap->stream(), enable));
  }
  
  
  int StreamWrap::DoShutdown(ShutdownWrap* req_wrap) {
    return uv_shutdown(&req_wrap->req_, stream(), AfterShutdown);
  }
  
  
  void StreamWrap::AfterShutdown(uv_shutdown_t* req, int status) {
    ShutdownWrap* req_wrap = ContainerOf(&ShutdownWrap::req_, req);
    HandleScope scope(req_wrap->env()->isolate());
    Context::Scope context_scope(req_wrap->env()->context());
    req_wrap->Done(status);
  }
  
  
  // NOTE: Call to this function could change both `buf`'s and `count`'s
  // values, shifting their base and decrementing their length. This is
  // required in order to skip the data that was successfully written via
  // uv_try_write().
  int StreamWrap::DoTryWrite(uv_buf_t** bufs, size_t* count) {
    int err;
    size_t written;
    uv_buf_t* vbufs = *bufs;
    size_t vcount = *count;
  
    err = uv_try_write(stream(), vbufs, vcount);
    if (err == UV_ENOSYS || err == UV_EAGAIN)
      return 0;
    if (err < 0)
      return err;
  
    // Slice off the buffers: skip all written buffers and slice the one that
    // was partially written.
    written = err;
    for (; written != 0 && vcount > 0; vbufs++, vcount--) {
      // Slice
      if (vbufs[0].len > written) {
        vbufs[0].base += written;
        vbufs[0].len -= written;
        written = 0;
        break;
  
      // Discard
      } else {
        written -= vbufs[0].len;
      }
    }
  
    *bufs = vbufs;
    *count = vcount;
  
    return 0;
  }
  
  
  int StreamWrap::DoWrite(WriteWrap* w,
                          uv_buf_t* bufs,
                          size_t count,
                          uv_stream_t* send_handle) {
    int r;
    if (send_handle == nullptr) {
      r = uv_write(&w->req_, stream(), bufs, count, AfterWrite);
    } else {
      r = uv_write2(&w->req_, stream(), bufs, count, send_handle, AfterWrite);
    }
  
    if (!r) {
      size_t bytes = 0;
      for (size_t i = 0; i < count; i++)
        bytes += bufs[i].len;
      if (stream()->type == UV_TCP) {
        NODE_COUNT_NET_BYTES_SENT(bytes);
      } else if (stream()->type == UV_NAMED_PIPE) {
        NODE_COUNT_PIPE_BYTES_SENT(bytes);
      }
    }
  
    UpdateWriteQueueSize();
  
    return r;
  }
  
  
  void StreamWrap::AfterWrite(uv_write_t* req, int status) {
    WriteWrap* req_wrap = ContainerOf(&WriteWrap::req_, req);
    HandleScope scope(req_wrap->env()->isolate());
    Context::Scope context_scope(req_wrap->env()->context());
    req_wrap->Done(status);
  }
  
  
  void StreamWrap::OnAfterWriteImpl(WriteWrap* w, void* ctx) {
    StreamWrap* wrap = static_cast<StreamWrap*>(ctx);
    wrap->UpdateWriteQueueSize();
  }
  
  }  // namespace node
  
  NODE_MODULE_CONTEXT_AWARE_BUILTIN(stream_wrap, node::StreamWrap::Initialize)
  