int do_connect(int type)
{
    int sock = 0;

    if (type != SOCK_STREAM && type != SOCK_DGRAM)
        return -1;

    /* We need a socket that can be inherited by child processes in
       ncat_exec_win.c, for --exec and --sh-exec. inheritable_socket is from
       nbase. */
    sock = inheritable_socket(targetss.storage.ss_family, type, 0);

    if (srcaddr.storage.ss_family != AF_UNSPEC) {
        size_t sa_len;

#ifdef HAVE_SOCKADDR_SA_LEN
        sa_len = srcaddr.sockaddr.sa_len;
#else
        sa_len = sizeof(srcaddr);
#endif
        if (bind(sock, &srcaddr.sockaddr, sa_len) < 0) {
            bye("bind to %s:%hu: %s.", inet_socktop(&srcaddr),
                inet_port(&srcaddr), socket_strerror(socket_errno()));
        }
    }

    if (sock != -1) {
        if (connect(sock, &targetss.sockaddr, (int) targetsslen) != -1)
            return sock;
        else if (socket_errno() == EINPROGRESS || socket_errno() == EAGAIN)
            return sock;
    }
    return -1;
}

void setup_environment(struct fdinfo *info)
{
    union sockaddr_u su;
    char ip[INET6_ADDRSTRLEN];
    char port[16];
    socklen_t alen = sizeof(su);

    if (getpeername(info->fd, &su.sockaddr, &alen) != 0) {
        bye("getpeername failed: %s", socket_strerror(socket_errno()));
    }
#ifdef HAVE_SYS_UN_H
    if (su.sockaddr.sa_family == AF_UNIX) {
        /* say localhost to keep it backwards compatible */
        setenv_portable("NCAT_REMOTE_ADDR", "localhost");
        setenv_portable("NCAT_REMOTE_PORT", "");
    } else
#endif
    if (getnameinfo((struct sockaddr *)&su, alen, ip, sizeof(ip),
            port, sizeof(port), NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
        setenv_portable("NCAT_REMOTE_ADDR", ip);
        setenv_portable("NCAT_REMOTE_PORT", port);
    } else {
        bye("getnameinfo failed: %s", socket_strerror(socket_errno()));
    }

    if (getsockname(info->fd, (struct sockaddr *)&su, &alen) < 0) {
        bye("getsockname failed: %s", socket_strerror(socket_errno()));
    }
#ifdef HAVE_SYS_UN_H
    if (su.sockaddr.sa_family == AF_UNIX) {
        /* say localhost to keep it backwards compatible, else su.un.sun_path */
        setenv_portable("NCAT_LOCAL_ADDR", "localhost");
        setenv_portable("NCAT_LOCAL_PORT", "");
    } else
#endif
    if (getnameinfo((struct sockaddr *)&su, alen, ip, sizeof(ip),
            port, sizeof(port), NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
        setenv_portable("NCAT_LOCAL_ADDR", ip);
        setenv_portable("NCAT_LOCAL_PORT", port);
    } else {
        bye("getnameinfo failed: %s", socket_strerror(socket_errno()));
    }

    switch(o.proto) {
        case IPPROTO_TCP:
            setenv_portable("NCAT_PROTO", "TCP");
            break;
        case IPPROTO_SCTP:
            setenv_portable("NCAT_PROTO", "SCTP");
            break;
        case IPPROTO_UDP:
            setenv_portable("NCAT_PROTO", "UDP");
            break;
    }
}

static int
xmlNanoHTTPSend(xmlNanoHTTPCtxtPtr ctxt, const char * xmt_ptr, int outlen) {

    int 	total_sent = 0;

    if ( (ctxt->state & XML_NANO_HTTP_WRITE) && (xmt_ptr != NULL ) ) {
        while (total_sent < outlen) {
            int nsent = send(ctxt->fd, xmt_ptr + total_sent,
                                      outlen - total_sent, 0);
            if (nsent>0)
                total_sent += nsent;
	    else if ( ( nsent == -1 ) && 
#if defined(EAGAIN) && EAGAIN != EWOULDBLOCK
	    	      ( socket_errno( ) != EAGAIN ) &&
#endif
		        ( socket_errno( ) != EWOULDBLOCK ) ) {
		__xmlIOErr(XML_FROM_HTTP, 0, "send failed\n");
		if ( total_sent == 0 )
		    total_sent = -1;
		break;
	    }
	    else {
	        /*
		**  No data sent
		**  Since non-blocking sockets are used, wait for 
		**  socket to be writable or default timeout prior
		**  to retrying.
		*/

		struct timeval	tv;
		fd_set		wfd;

		tv.tv_sec = timeout;
		tv.tv_usec = 0;
		FD_ZERO( &wfd );
#ifdef _MSC_VER
#pragma warning(push)
#pragma warning(disable: 4018)
#endif
		FD_SET( ctxt->fd, &wfd );
#ifdef _MSC_VER
#pragma warning(pop)
#endif
		(void)select( ctxt->fd + 1, NULL, &wfd, NULL, &tv );
	    }
	}
    }

    return total_sent;
}

int get_authenticated_socket( const char * host, int port)
{
    int remote = 0 ;
    int ret;

    dest_host = strdup(host);
    dest_port = port;
    
    if (0 < connect_timeout)
        set_timeout (connect_timeout);
    //if (check_direct(dest_host))
    //   relay_method = METHOD_DIRECT;
  
    if ( relay_method == METHOD_DIRECT ) {
        remote = open_connection (dest_host, dest_port);
        if ( remote == SOCKET_ERROR )
            g_error( "Unable to connect to destination host, errno=%d\n",
                     socket_errno());
    } else {
        remote = open_connection (relay_host, relay_port);
        if ( remote == SOCKET_ERROR )
            g_error( "Unable to connect to relay host, errno=%d\n",
                     socket_errno());
    }

    if (socks_ns.sin_addr.s_addr != 0)
        switch_ns (&socks_ns);

    switch ( relay_method ) {
        case METHOD_HTTP:
            ret = begin_http_relay(remote);
            switch (ret) {
                case START_ERROR:
                    close (remote);
                    g_error("failed to begin relaying via HTTP.\n");
                case START_OK:
                    break;
                case START_RETRY:
                    
                    close (remote);
            }
            break;
        case METHOD_DIRECT:
            g_debug("Did not using proxy bypass ...(%s,%d)",__FILE__, __LINE__);
            break;
    }
    return remote;
}

/* Create the actual socket (nse->iod->sd) underlying the iod. This unblocks the
 * socket, binds to the localaddr address, sets IP options, and sets the
 * broadcast flag. Trying to change these functions after making this call will
 * not have an effect. This function needs to be called before you try to read
 * or write on the iod. */
static int nsock_make_socket(struct npool *ms, struct niod *iod, int family, int type, int proto) {

  /* inheritable_socket is from nbase */
  iod->sd = (int)inheritable_socket(family, type, proto);
  if (iod->sd == -1) {
    nsock_log_error("Socket trouble: %s", socket_strerror(socket_errno()));
    return -1;
  }

  unblock_socket(iod->sd);

  iod->lastproto = proto;

  if (iod->locallen)
    mksock_bind_addr(ms, iod);

  if (iod->ipoptslen && family == AF_INET)
    mksock_set_ipopts(ms, iod);

  if (ms->device)
    mksock_bind_device(ms, iod);

  if (ms->broadcast && type != SOCK_STREAM)
    mksock_set_broadcast(ms, iod);

  /* mksock_* functions can raise warnings/errors
   * but we don't let them stop us for now. */
  return iod->sd;
}

/* Broadcast a message to all the descriptors in fds. Returns -1 if any of the
   sends failed. */
int ncat_broadcast(fd_set *fds, const fd_list_t *fdlist, const char *msg, size_t size)
{
    struct fdinfo *fdn;
    int i, ret;

    if (o.recvonly)
        return size;

    ret = 0;
    for (i = 0; i <= fdlist->fdmax; i++) {
        if (!FD_ISSET(i, fds))
            continue;

        fdn = get_fdinfo(fdlist, i);
        ncat_assert(fdn != NULL);
        if (blocking_fdinfo_send(fdn, msg, size) <= 0) {
            if (o.debug > 1)
                logdebug("Error sending to fd %d: %s.\n", i, socket_strerror(socket_errno()));
            ret = -1;
        }
    }

    ncat_log_send(msg, size);

    return ret;
}

ssize_t R_SockRead(int sockp, void *buf, size_t len, int blocking, int timeout)
{
    ssize_t res;

    if(blocking && R_SocketWait(sockp, 0, timeout) != 0) return 0;
    res = recv(sockp, buf, len, 0);
    return (res >= 0) ? res : -socket_errno();
}

ssize_t R_SockRead(int sockp, void *buf, size_t len, int blocking, int timeout)
{
    ssize_t res;

    if(blocking && (res = R_SocketWait(sockp, 0, timeout)) != 0)
        return res < 0 ? res : 0; /* socket error or timeout */
    res = recv(sockp, buf, len, 0);
    return (res >= 0) ? res : -socket_errno();
}

obj_t bigloo_socket_read_returns_data(int fd) {
  char buf;

  if ((0 == recv(fd, &buf, 1, MSG_PEEK)) && 
      (socket_errno() != EAGAIN)) {
    return BFALSE;
  } else {
    return BTRUE;
  }
}

void setup_environment(struct fdinfo *info)
{
    union sockaddr_u su;
    char ip[INET6_ADDRSTRLEN];
    char port[16];
    socklen_t alen = sizeof(su);

    if (getpeername(info->fd, &su.sockaddr, &alen) != 0) {
        bye("getpeername failed: %s", socket_strerror(socket_errno()));
    }
    if (getnameinfo((struct sockaddr *)&su, alen, ip, sizeof(ip),
            port, sizeof(port), NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
        setenv_portable("NCAT_REMOTE_ADDR", ip);
        setenv_portable("NCAT_REMOTE_PORT", port);
    } else {
        bye("getnameinfo failed: %s", socket_strerror(socket_errno()));
    }

    if (getsockname(info->fd, (struct sockaddr *)&su, &alen) < 0) {
        bye("getsockname failed: %s", socket_strerror(socket_errno()));
    }
    if (getnameinfo((struct sockaddr *)&su, alen, ip, sizeof(ip),
            port, sizeof(port), NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
        setenv_portable("NCAT_LOCAL_ADDR", ip);
        setenv_portable("NCAT_LOCAL_PORT", port);
    } else {
        bye("getnameinfo failed: %s", socket_strerror(socket_errno()));
    }

    switch(o.proto) {
        case IPPROTO_TCP:
            setenv_portable("NCAT_PROTO", "TCP");
            break;
        case IPPROTO_SCTP:
            setenv_portable("NCAT_PROTO", "SCTP");
            break;
        case IPPROTO_UDP:
            setenv_portable("NCAT_PROTO", "UDP");
            break;
    }
}

int
htsp_tcp_read(socket_t fd, void *buf, size_t len)
{
  int x = recv(fd, buf, len, MSG_WAITALL);

  if(x == -1)
    return socket_errno();
  if(x != len)
    return ECONNRESET;
  return 0;

}

static int mksock_bind_addr(struct npool *ms, struct niod *iod) {
  int rc;
  int one = 1;

  rc = setsockopt(iod->sd, SOL_SOCKET, SO_REUSEADDR, (const char *)&one, sizeof(one));
  if (rc == -1) {
    int err = socket_errno();

    nsock_log_error("Setting of SO_REUSEADDR failed (#%li): %s (%d)", iod->id,
                    socket_strerror(err), err);
  }

  nsock_log_info("Binding to %s (IOD #%li)", get_localaddr_string(iod), iod->id);
  rc = bind(iod->sd, (struct sockaddr *)&iod->local, (int) iod->locallen);
  if (rc == -1) {
    int err = socket_errno();

    nsock_log_error("Bind to %s failed (IOD #%li): %s (%d)",
                    get_localaddr_string(iod), iod->id,
                    socket_strerror(err), err);
  }
  return 0;
}

static int mksock_set_broadcast(struct npool *ms, struct niod *iod) {
  int rc;
  int one = 1;

  rc = setsockopt(iod->sd, SOL_SOCKET, SO_BROADCAST,
                  (const char *)&one, sizeof(one));
  if (rc == -1) {
    int err = socket_errno();

    nsock_log_error("Setting of SO_BROADCAST failed (IOD #%li): %s (%d)",
                    iod->id, socket_strerror(err), err);
  }
  return 0;
}

ssize_t R_SockWrite(int sockp, const void *buf, size_t len, int timeout)
{
    ssize_t res, out = 0;

    /* Rprintf("socket %d writing |%s|\n", sockp, buf); */
    /* This function is not passed a `blocking' argument so the code
       here is equivalent to blocking == TRUE; it's not clear
       non-blocking writes make much sense with the current connection
       interface since there is no way to tell how much, if anything,
       has been written.  LT */
    do {
	if(R_SocketWait(sockp, 1, timeout) != 0) return out;
	res = send(sockp, buf, len, 0);
	if (res < 0 && socket_errno() != EWOULDBLOCK)
	    return -socket_errno();
	else {
	    { const char *cbuf = buf; cbuf += res; buf = cbuf; }
	    len -= res;
	    out += res;
	}
    } while (/* ! blocking && */len > 0);
    return out;
}

static int mksock_set_ipopts(struct npool *ms, struct niod *iod) {
  int rc;

  errno = 0;
  rc = setsockopt(iod->sd, IPPROTO_IP, IP_OPTIONS, (const char *)iod->ipopts,
                  iod->ipoptslen);
  if (rc == -1) {
    int err = socket_errno();

    nsock_log_error("Setting of IP options failed (IOD #%li): %s (%d)",
                    iod->id, socket_strerror(err), err);
  }
  return 0;
}

int sendf(SOCKET s, const char *fmt,...)
{
    static char buf[10240];                   
    
    va_list args;
    va_start( args, fmt );
    vsnprintf( buf, sizeof(buf), fmt, args );
    va_end( args );

    //report_text(">>>", buf);
    if ( send(s, buf, strlen(buf), 0) == SOCKET_ERROR ) {
        g_debug("failed to send http request. errno=%d\n", socket_errno());
        return -1;
    }
    return 0;
}

static int mksock_bind_device(struct npool *ms, struct niod *iod) {
  int rc;

  rc = socket_bindtodevice(iod->sd, ms->device);
  if (!rc) {
    int err = socket_errno();

    if (err != EPERM)
      nsock_log_error("Setting of SO_BINDTODEVICE failed (IOD #%li): %s (%d)",
                      iod->id, socket_strerror(err), err);
    else
      nsock_log_debug_all("Setting of SO_BINDTODEVICE failed (IOD #%li): %s (%d)",
                          iod->id, socket_strerror(err), err);
  }
  return 0;
}

/*
 * socktoa - return a numeric host name from a sockaddr_storage structure
 */
const char *
socktoa(
	const sockaddr_u *sock
	)
{
	int		saved_errno;
	char *		res;
	char *		addr;
	u_long		scope;

	saved_errno = socket_errno();
	LIB_GETBUF(res);

	if (NULL == sock) {
		strlcpy(res, "(null)", LIB_BUFLENGTH);
	} else {
		switch(AF(sock)) {

		case AF_INET:
		case AF_UNSPEC:
			inet_ntop(AF_INET, PSOCK_ADDR4(sock), res,
				  LIB_BUFLENGTH);
			break;

		case AF_INET6:
			inet_ntop(AF_INET6, PSOCK_ADDR6(sock), res,
				  LIB_BUFLENGTH);
			scope = SCOPE_VAR(sock);
			if (0 != scope && !strchr(res, '%')) {
				addr = res;
				LIB_GETBUF(res);
				snprintf(res, LIB_BUFLENGTH, "%s%%%lu",
					 addr, scope);
				res[LIB_BUFLENGTH - 1] = '\0';
			}
			break;

		default:
			snprintf(res, LIB_BUFLENGTH, 
				 "(socktoa unknown family %d)", 
				 AF(sock));
		}
	}
	errno = saved_errno;

	return res;
}

int atomic_out( SOCKET s, char *buf, int size )
{
    int ret, len;

    assert( buf != NULL );
    assert( 0<=size );
  
    ret = 0;
    while ( 0 < size ) {
        len = send( s, buf+ret, size, 0 );
        if ( len == -1 )
            g_error("atomic_out() failed to send(), %d\n", socket_errno());
        ret += len;
        size -= len;
    }
    g_debug("atomic_out()  [%d bytes]\n", ret);
    return ret;
}

int
htsp_tcp_read_timeout(socket_t fd, char *buf, size_t len, int timeout)
{
  int x, tot = 0, val, err;
  fd_set fd_read;
  struct timeval tv;

  assert(timeout > 0);

  while(tot != len) {

    tv.tv_sec  =         timeout / 1000;
    tv.tv_usec = 1000 * (timeout % 1000);

    FD_ZERO(&fd_read);
    FD_SET(fd, &fd_read);

    x = select((int)fd+1, &fd_read, NULL, NULL, &tv);

    if(x == 0)
      return ETIMEDOUT;

    val = 1;
    ioctlsocket(fd, FIONBIO, &val);

    x   = recv(fd, buf + tot, len - tot, 0);
    err = socket_errno();

    val = 0;
    ioctlsocket(fd, FIONBIO, &val);

    if(x == 0)
      return ECONNRESET;
    else if(x == -1)
    {
      if(err == EAGAIN)
        continue;
      return err;
    }

    tot += x;
  }
  return 0;
}