自学教程：C++ uv__stream_fd函数代码示例

int uv_fileno(const uv_handle_t* handle, uv_os_fd_t* fd) {  int fd_out;  switch (handle->type) {  case UV_TCP:  case UV_NAMED_PIPE:  case UV_TTY:    fd_out = uv__stream_fd((uv_stream_t*) handle);    break;  case UV_UDP:    fd_out = ((uv_udp_t *) handle)->io_watcher.fd;    break;  case UV_POLL:    fd_out = ((uv_poll_t *) handle)->io_watcher.fd;    break;  default:    return -EINVAL;  }  if (uv__is_closing(handle) || fd_out == -1)    return -EBADF;  *fd = fd_out;  return 0;}

int uv__socket_sockopt(uv_handle_t* handle, int optname, int* value) {  int r;  int fd;  socklen_t len;  if (handle == NULL || value == NULL)    return UV_EINVAL;  if (handle->type == UV_TCP || handle->type == UV_NAMED_PIPE)    fd = uv__stream_fd((uv_stream_t*) handle);  else if (handle->type == UV_UDP)    fd = ((uv_udp_t *) handle)->io_watcher.fd;  else    return UV_ENOTSUP;  len = sizeof(*value);  if (*value == 0)    r = getsockopt(fd, SOL_SOCKET, optname, value, &len);  else    r = setsockopt(fd, SOL_SOCKET, optname, (const void*) value, len);  if (r < 0)    return UV__ERR(errno);  return 0;}

int uv_tcp_nodelay(uv_tcp_t* handle, int on) {  int err;  if (uv__stream_fd(handle) != -1) {    err = uv__tcp_nodelay(uv__stream_fd(handle), on);    if (err)      return err;  }  if (on)    handle->flags |= UV_TCP_NODELAY;  else    handle->flags &= ~UV_TCP_NODELAY;  return 0;}

/* * Used for initializing stdio streams like options.stdin_stream. Returns * zero on success. See also the cleanup section in uv_spawn(). */static int uv__process_init_stdio(uv_stdio_container_t* container, int fds[2]) {  int mask;  int fd;  mask = UV_IGNORE | UV_CREATE_PIPE | UV_INHERIT_FD | UV_INHERIT_STREAM;  switch (container->flags & mask) {  case UV_IGNORE:    return 0;  case UV_CREATE_PIPE:    assert(container->data.stream != NULL);    if (container->data.stream->type != UV_NAMED_PIPE)      return -EINVAL;    else      return uv__make_socketpair(fds, 0);  case UV_INHERIT_FD:  case UV_INHERIT_STREAM:    if (container->flags & UV_INHERIT_FD)      fd = container->data.fd;    else      fd = uv__stream_fd(container->data.stream);    if (fd == -1)      return -EINVAL;    fds[1] = fd;    return 0;  default:    assert(0 && "Unexpected flags");    return -EINVAL;  }}

static void uv__drain(uv_stream_t* stream) {  uv_shutdown_t* req;  assert(ngx_queue_empty(&stream->write_queue));  assert(stream->write_queue_size == 0);  uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLOUT);  /* Shutdown? */  if ((stream->flags & UV_STREAM_SHUTTING) &&      !(stream->flags & UV_CLOSING) &&      !(stream->flags & UV_STREAM_SHUT)) {    assert(stream->shutdown_req);    req = stream->shutdown_req;    stream->shutdown_req = NULL;    uv__req_unregister(stream->loop, req);    if (shutdown(uv__stream_fd(stream), SHUT_WR)) {      /* Error. Report it. User should call uv_close(). */      uv__set_sys_error(stream->loop, errno);      if (req->cb) {        req->cb(req, -1);      }    } else {      uv__set_sys_error(stream->loop, 0);      ((uv_handle_t*) stream)->flags |= UV_STREAM_SHUT;      if (req->cb) {        req->cb(req, 0);      }    }  }}

/** * We get called here from directly following a call to connect(2). * In order to determine if we've errored out or succeeded must call * getsockopt. */static void uv__stream_connect(uv_stream_t* stream) {  int error;  uv_connect_t* req = stream->connect_req;  socklen_t errorsize = sizeof(int);  assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE);  assert(req);  if (stream->delayed_error) {    /* To smooth over the differences between unixes errors that     * were reported synchronously on the first connect can be delayed     * until the next tick--which is now.     */    error = stream->delayed_error;    stream->delayed_error = 0;  } else {    /* Normal situation: we need to get the socket error from the kernel. */    assert(uv__stream_fd(stream) >= 0);    tuvp_getsockopt(uv__stream_fd(stream),                    SOL_SOCKET,                    SO_ERROR,                    &error,                    &errorsize);    error = -error;  }  if (error == -EINPROGRESS)    return;  stream->connect_req = NULL;  uv__req_unregister(stream->loop, req);  if (error < 0 || QUEUE_EMPTY(&stream->write_queue)) {    uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLOUT);  }  if (req->cb)    req->cb(req, error);  if (uv__stream_fd(stream) == -1)    return;  if (error < 0) {    uv__stream_flush_write_queue(stream, -ECANCELED);    uv__write_callbacks(stream);  }}

static int maybe_new_socket(uv_tcp_t* handle, int domain, unsigned long flags) {  struct sockaddr_storage saddr;  socklen_t slen;  if (domain == AF_UNSPEC) {    handle->flags |= flags;    return 0;  }  if (uv__stream_fd(handle) != -1) {    if (flags & UV_HANDLE_BOUND) {      if (handle->flags & UV_HANDLE_BOUND) {        /* It is already bound to a port. */        handle->flags |= flags;        return 0;      }            /* Query to see if tcp socket is bound. */      slen = sizeof(saddr);      memset(&saddr, 0, sizeof(saddr));      if (getsockname(uv__stream_fd(handle), (struct sockaddr*) &saddr, &slen))        return -errno;      if ((saddr.ss_family == AF_INET6 &&          ((struct sockaddr_in6*) &saddr)->sin6_port != 0) ||          (saddr.ss_family == AF_INET &&          ((struct sockaddr_in*) &saddr)->sin_port != 0)) {        /* Handle is already bound to a port. */        handle->flags |= flags;        return 0;      }      /* Bind to arbitrary port */      if (bind(uv__stream_fd(handle), (struct sockaddr*) &saddr, slen))        return -errno;    }    handle->flags |= flags;    return 0;  }  return new_socket(handle, domain, flags);}

int uv_device_ioctl(uv_device_t* device, int cmd, uv_ioargs_t* args) {  int err;  err = ioctl(uv__stream_fd((uv_stream_t*) device), cmd, args->arg);  if (err < 0)    return -errno;    return err;}

int uv_device_ioctl(uv_device_t* device,                    unsigned long cmd,                    uv_ioargs_t* args) {    int err = ioctl(uv__stream_fd((uv_stream_t*) device), cmd, args->arg);    if (err < 0) {        return -errno;    }    return err;}

int uv__tcp_connect(uv_connect_t* req,                    uv_tcp_t* handle,                    const struct sockaddr* addr,                    unsigned int addrlen,                    uv_connect_cb cb) {  int err;  int r;  assert(handle->type == UV_TCP);  if (handle->connect_req != NULL)    return -EALREADY;  /* FIXME(bnoordhuis) -EINVAL or maybe -EBUSY. */  err = maybe_new_socket(handle,                         addr->sa_family,                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);  if (err)    return err;  handle->delayed_error = 0;  do {    errno = 0;    r = connect(uv__stream_fd(handle), addr, addrlen);  } while (r == -1 && errno == EINTR);  /* We not only check the return value, but also check the errno != 0.   * Because in rare cases connect() will return -1 but the errno   * is 0 (for example, on Android 4.3, OnePlus phone A0001_12_150227)   * and actually the tcp three-way handshake is completed.   */  if (r == -1 && errno != 0) {    if (errno == EINPROGRESS)      ; /* not an error */    else if (errno == ECONNREFUSED)    /* If we get a ECONNREFUSED wait until the next tick to report the     * error. Solaris wants to report immediately--other unixes want to     * wait.     */      handle->delayed_error = -errno;    else      return -errno;  }  uv__req_init(handle->loop, req, UV_CONNECT);  req->cb = cb;  req->handle = (uv_stream_t*) handle;  QUEUE_INIT(&req->queue);  handle->connect_req = req;  uv__io_start(handle->loop, &handle->io_watcher, POLLOUT);  if (handle->delayed_error)    uv__io_feed(handle->loop, &handle->io_watcher);  return 0;}

void uv__server_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {  uv_stream_t* stream;  int err;  stream = container_of(w, uv_stream_t, io_watcher);  assert(events == UV__POLLIN);  assert(stream->accepted_fd == -1);  assert(!(stream->flags & UV_CLOSING));  uv__io_start(stream->loop, &stream->io_watcher, UV__POLLIN);  /* connection_cb can close the server socket while we're   * in the loop so check it on each iteration.   */  while (uv__stream_fd(stream) != -1) {    assert(stream->accepted_fd == -1);    set_errno(0);    err = uv__accept(uv__stream_fd(stream));    if (err < 0) {      if (err == -EAGAIN || err == -EWOULDBLOCK)        return;  /* Not an error. */      if (err == -ECONNABORTED)        continue;  /* Ignore. Nothing we can do about that. */      stream->connection_cb(stream, err);      continue;    }    stream->accepted_fd = err;    stream->connection_cb(stream, 0);    if (stream->accepted_fd != -1) {      /* The user hasn't yet accepted called uv_accept() */      uv__io_stop(loop, &stream->io_watcher, UV__POLLIN);      return;    }    /* done accept for mbed */    break;  }}

int uv_pipe_bind(uv_pipe_t* handle, const char* name) {  struct sockaddr_un saddr;  const char* pipe_fname;  int sockfd;  int err;  size_t name_len;  pipe_fname = NULL;  sockfd = -1;  name_len = strlen(name);  if (name_len > sizeof(saddr.sun_path) - 1)    return -ENAMETOOLONG;  /* Already bound? */  if (uv__stream_fd(handle) >= 0)    return -EINVAL;  /* Make a copy of the file name, it outlives this function's scope. */  pipe_fname = uv__strdup(name);  if (pipe_fname == NULL)    return -ENOMEM;  /* We've got a copy, don't touch the original any more. */  name = NULL;  err = uv__socket(AF_UNIX, SOCK_STREAM, 0);  if (err < 0)    goto err_socket;  sockfd = err;  memset(&saddr, 0, sizeof saddr);  memcpy(saddr.sun_path, pipe_fname, name_len);  saddr.sun_family = AF_UNIX;  if (bind(sockfd, (struct sockaddr*)&saddr, sizeof saddr)) {    err = -errno;    /* Convert ENOENT to EACCES for compatibility with Windows. */    if (err == -ENOENT)      err = -EACCES;    uv__close(sockfd);    goto err_socket;  }  /* Success. */  handle->flags |= UV_HANDLE_BOUND;  handle->pipe_fname = pipe_fname; /* Is a strdup'ed copy. */  handle->io_watcher.fd = sockfd;  return 0;err_socket:  uv__free((void*)pipe_fname);  return err;}

int uv_tty_get_winsize(uv_tty_t* tty, int* width, int* height) {  struct winsize ws;  if (ioctl(uv__stream_fd(tty), TIOCGWINSZ, &ws))    return -errno;  *width = ws.ws_col;  *height = ws.ws_row;  return 0;}

int uv_tcp_v6only(uv_tcp_t* handle, int on) {  if (uv__stream_fd(handle) != -1)    return -1;  if (on)    handle->flags |= UV_TCP_V6ONLY;  else    handle->flags &= ~UV_TCP_V6ONLY;  return 0;}

int uv_tcp_getpeername(uv_tcp_t* handle,                       struct sockaddr* name,                       int* namelen) {  socklen_t socklen;  if (handle->delayed_error)    return handle->delayed_error;  if (uv__stream_fd(handle) < 0)    return -EINVAL;  /* FIXME(bnoordhuis) -EBADF */  /* sizeof(socklen_t) != sizeof(int) on some systems. */  socklen = (socklen_t) *namelen;  if (getpeername(uv__stream_fd(handle), name, &socklen))    return -errno;  *namelen = (int) socklen;  return 0;}

int uv_tcp_getpeername(uv_tcp_t* handle,                       struct sockaddr* name,                       int* namelen) {  socklen_t socklen;  if (handle->delayed_error)    return uv__set_sys_error(handle->loop, handle->delayed_error);  if (uv__stream_fd(handle) < 0)    return uv__set_sys_error(handle->loop, EINVAL);  /* sizeof(socklen_t) != sizeof(int) on some systems. */  socklen = (socklen_t) *namelen;  if (getpeername(uv__stream_fd(handle), name, &socklen) == -1)    return uv__set_sys_error(handle->loop, errno);  *namelen = (int) socklen;  return 0;}

int uv_pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {  if (uv__stream_fd(handle) == -1)    return -EINVAL;#if defined(__MVS__)  /* On zOS, backlog=0 has undefined behaviour */  if (backlog == 0)    backlog = 1;  else if (backlog < 0)    backlog = SOMAXCONN;#endif  if (listen(uv__stream_fd(handle), backlog))    return -errno;  handle->connection_cb = cb;  handle->io_watcher.cb = uv__server_io;  uv__io_start(handle->loop, &handle->io_watcher, POLLIN);  return 0;}

int uv_tcp_keepalive(uv_tcp_t* handle, int on, unsigned int delay) {  int err;  if (uv__stream_fd(handle) != -1) {    err =uv__tcp_keepalive(uv__stream_fd(handle), on, delay);    if (err)      return err;  }  if (on)    handle->flags |= UV_TCP_KEEPALIVE;  else    handle->flags &= ~UV_TCP_KEEPALIVE;  /* TODO Store delay if uv__stream_fd(handle) == -1 but don't want to enlarge   *      uv_tcp_t with an int that's almost never used...   */  return 0;}

int uv_tty_get_winsize(uv_tty_t* tty, int* width, int* height) {  struct winsize ws;  if (ioctl(uv__stream_fd(tty), TIOCGWINSZ, &ws) < 0) {    uv__set_sys_error(tty->loop, errno);    return -1;  }  *width = ws.ws_col;  *height = ws.ws_row;  return 0;}

static int uv__connect(uv_connect_t* req,                       uv_tcp_t* handle,                       struct sockaddr* addr,                       socklen_t addrlen,                       uv_connect_cb cb) {  int err;  int r;  assert(handle->type == UV_TCP);  if (handle->connect_req != NULL)    return -EALREADY;  /* FIXME(bnoordhuis) -EINVAL or maybe -EBUSY. */  err = maybe_new_socket(handle,                         addr->sa_family,                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);  if (err)    return err;  handle->delayed_error = 0;  do    r = connect(uv__stream_fd(handle), addr, addrlen);  while (r == -1 && errno == EINTR);  if (r == -1) {    if (errno == EINPROGRESS)      ; /* not an error */    else if (errno == ECONNREFUSED)    /* If we get a ECONNREFUSED wait until the next tick to report the     * error. Solaris wants to report immediately--other unixes want to     * wait.     */      handle->delayed_error = -errno;    else      return -errno;  }  uv__req_init(handle->loop, req, UV_CONNECT);  req->cb = cb;  req->handle = (uv_stream_t*) handle;  QUEUE_INIT(&req->queue);  handle->connect_req = req;  uv__io_start(handle->loop, &handle->io_watcher, UV__POLLOUT);  if (handle->delayed_error)    uv__io_feed(handle->loop, &handle->io_watcher);  return 0;}

int uv_tty_set_mode(uv_tty_t* tty, int mode) {  struct termios raw;  int fd;  fd = uv__stream_fd(tty);  if (mode && tty->mode == 0) {    /* on */    if (tcgetattr(fd, &tty->orig_termios)) {      goto fatal;    }    /* This is used for uv_tty_reset_mode() */    if (orig_termios_fd == -1) {      orig_termios = tty->orig_termios;      orig_termios_fd = fd;    }    raw = tty->orig_termios;    raw.c_iflag &= ~(BRKINT | ICRNL | INPCK | ISTRIP | IXON);    raw.c_oflag |= (ONLCR);    raw.c_cflag |= (CS8);    raw.c_lflag &= ~(ECHO | ICANON | IEXTEN | ISIG);    raw.c_cc[VMIN] = 1;    raw.c_cc[VTIME] = 0;    /* Put terminal in raw mode after draining */    if (tcsetattr(fd, TCSADRAIN, &raw)) {      goto fatal;    }    tty->mode = 1;    return 0;  } else if (mode == 0 && tty->mode) {    /* off */    /* Put terminal in original mode after flushing */    if (tcsetattr(fd, TCSAFLUSH, &tty->orig_termios)) {      goto fatal;    }    tty->mode = 0;    return 0;  }fatal:  uv__set_sys_error(tty->loop, errno);  return -1;}

/** * We get called here from directly following a call to connect(2). * In order to determine if we've errored out or succeeded must call * getsockopt. */static void uv__stream_connect(uv_stream_t* stream) {  int error;  uv_connect_t* req = stream->connect_req;  socklen_t errorsize = sizeof(int);  assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE);  assert(req);  if (stream->delayed_error) {    /* To smooth over the differences between unixes errors that     * were reported synchronously on the first connect can be delayed     * until the next tick--which is now.     */    error = stream->delayed_error;    stream->delayed_error = 0;  } else {    /* Normal situation: we need to get the socket error from the kernel. */    assert(uv__stream_fd(stream) >= 0);    getsockopt(uv__stream_fd(stream),               SOL_SOCKET,               SO_ERROR,               &error,               &errorsize);  }  if (error == EINPROGRESS)    return;  stream->connect_req = NULL;  uv__req_unregister(stream->loop, req);  if (req->cb) {    uv__set_sys_error(stream->loop, error);    req->cb(req, error ? -1 : 0);  }}

static int new_socket(uv_tcp_t* handle, int domain, unsigned long flags) {  struct sockaddr_storage saddr;  socklen_t slen;  int sockfd;  int err;  err = uv__socket(domain, SOCK_STREAM, 0);  if (err < 0)    return err;  sockfd = err;  err = uv__stream_open((uv_stream_t*) handle, sockfd, flags);  if (err) {    uv__close(sockfd);    return err;  }  if (flags & UV_HANDLE_BOUND) {    /* Bind this new socket to an arbitrary port */    slen = sizeof(saddr);    memset(&saddr, 0, sizeof(saddr));    err = getsockname(uv__stream_fd(handle), (struct sockaddr*) &saddr, &slen);    if (err) {      uv__close(sockfd);      return err;    }    err = bind(uv__stream_fd(handle), (struct sockaddr*) &saddr, slen);    if (err) {      uv__close(sockfd);      return err;    }  }  return 0;}

int uv_tty_get_winsize(uv_tty_t* tty, int* width, int* height) {  struct winsize ws;  int err;  do    err = ioctl(uv__stream_fd(tty), TIOCGWINSZ, &ws);  while (err == -1 && errno == EINTR);  if (err == -1)    return UV__ERR(errno);  *width = ws.ws_col;  *height = ws.ws_row;  return 0;}

int uv_tty_set_mode(uv_tty_t* tty, uv_tty_mode_t mode) {  struct termios tmp;  int fd;  if (tty->mode == (int) mode)    return 0;  fd = uv__stream_fd(tty);  if (tty->mode == UV_TTY_MODE_NORMAL && mode != UV_TTY_MODE_NORMAL) {    if (tcgetattr(fd, &tty->orig_termios))      return -errno;    /* This is used for uv_tty_reset_mode() */    uv_spinlock_lock(&termios_spinlock);    if (orig_termios_fd == -1) {      orig_termios = tty->orig_termios;      orig_termios_fd = fd;    }    uv_spinlock_unlock(&termios_spinlock);  }  tmp = tty->orig_termios;  switch (mode) {    case UV_TTY_MODE_NORMAL:      break;    case UV_TTY_MODE_RAW:      tmp.c_iflag &= ~(BRKINT | ICRNL | INPCK | ISTRIP | IXON);      tmp.c_oflag |= (ONLCR);      tmp.c_cflag |= (CS8);      tmp.c_lflag &= ~(ECHO | ICANON | IEXTEN | ISIG);      tmp.c_cc[VMIN] = 1;      tmp.c_cc[VTIME] = 0;      break;    case UV_TTY_MODE_IO:      uv__tty_make_raw(&tmp);      break;  }  /* Apply changes after draining */  if (tcsetattr(fd, TCSADRAIN, &tmp))    return -errno;  tty->mode = mode;  return 0;}

int uv_tty_set_mode(uv_tty_t* tty, int mode) {  struct termios raw;  int fd;  fd = uv__stream_fd(tty);  if (mode && tty->mode == 0) {  /* on */    if (tcgetattr(fd, &tty->orig_termios))      return -errno;    /* This is used for uv_tty_reset_mode() */    uv_spinlock_lock(&termios_spinlock);    if (orig_termios_fd == -1) {      orig_termios = tty->orig_termios;      orig_termios_fd = fd;    }    uv_spinlock_unlock(&termios_spinlock);    raw = tty->orig_termios;#ifdef _BSD_SOURCE    cfmakeraw(&raw);#else    raw.c_iflag &= ~(IGNBRK | BRKINT | PARMRK | ISTRIP | INLCR | IGNCR | ICRNL | IXON);    raw.c_oflag &= ~OPOST;    raw.c_lflag &= ~(ECHO | ECHONL | ICANON | ISIG | IEXTEN);    raw.c_cflag &= ~(CSIZE | PARENB);    raw.c_cflag |= CS8;#endif    raw.c_cc[VMIN] = 1;    raw.c_cc[VTIME] = 0;    /* Put terminal in raw mode after draining */    if (tcsetattr(fd, TCSADRAIN, &raw))      return -errno;    tty->mode = 1;  } else if (mode == 0 && tty->mode) {  /* off */    /* Put terminal in original mode after flushing */    if (tcsetattr(fd, TCSAFLUSH, &tty->orig_termios))      return -errno;    tty->mode = 0;  }  return 0;}

static int maybe_new_socket(uv_tcp_t* handle, int domain, int flags) {  int sockfd;  if (uv__stream_fd(handle) != -1)    return 0;  sockfd = uv__socket(domain, SOCK_STREAM, 0);  if (sockfd == -1)    return uv__set_sys_error(handle->loop, errno);  if (uv__stream_open((uv_stream_t*)handle, sockfd, flags)) {    close(sockfd);    return -1;  }  return 0;}

static int maybe_new_socket(uv_tcp_t* handle, int domain, int flags) {  int sockfd;  int err;  if (uv__stream_fd(handle) != -1)    return 0;  err = uv__socket(domain, SOCK_STREAM, 0);  if (err < 0)    return err;  sockfd = err;  err = uv__stream_open((uv_stream_t*) handle, sockfd, flags);  if (err) {    close(sockfd);    return err;  }  return 0;}

static int uv__pipe_getsockpeername(const uv_pipe_t* handle,                                    uv__peersockfunc func,                                    char* buffer,                                    size_t* size) {  struct sockaddr_un sa;  socklen_t addrlen;  int err;  addrlen = sizeof(sa);  memset(&sa, 0, addrlen);  err = func(uv__stream_fd(handle), (struct sockaddr*) &sa, &addrlen);  if (err < 0) {    *size = 0;    return -errno;  }#if defined(__linux__)  if (sa.sun_path[0] == 0)    /* Linux abstract namespace */    addrlen -= offsetof(struct sockaddr_un, sun_path);  else

static int uv__read_start_common(uv_stream_t* stream,                                 uv_alloc_cb alloc_cb,                                 uv_read_cb read_cb,                                 uv_read2_cb read2_cb) {  assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE ||      stream->type == UV_TTY);  if (stream->flags & UV_CLOSING) {    uv__set_sys_error(stream->loop, EINVAL);    return -1;  }  /* The UV_STREAM_READING flag is irrelevant of the state of the tcp - it just   * expresses the desired state of the user.   */  stream->flags |= UV_STREAM_READING;#if defined(__APPLE__)  /* Notify select() thread about state change */  if (stream->select != NULL)    uv__stream_osx_interrupt_select(stream);#endif /* defined(__APPLE__) */  /* TODO: try to do the read inline? */  /* TODO: keep track of tcp state. If we've gotten a EOF then we should   * not start the IO watcher.   */  assert(uv__stream_fd(stream) >= 0);  assert(alloc_cb);  stream->read_cb = read_cb;  stream->read2_cb = read2_cb;  stream->alloc_cb = alloc_cb;  uv__io_start(stream->loop, &stream->io_watcher, UV__POLLIN);  uv__handle_start(stream);  return 0;}