stream_wrap.cc

#include "stream_wrap.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) {
  env->SetProtoMethod(target, "setBlocking", SetBlocking);
  StreamBase::AddMethods<StreamWrap>(env, target);
}


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);
  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);

  // 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);
  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);
  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)