/*-------------------------------------------------------------------------*\
* "Disconnects" a DGRAM socket
\*-------------------------------------------------------------------------*/
const char *inet_trydisconnect(p_socket ps, int family, p_timeout tm)
{
    switch (family) {
        case PF_INET: {
            struct sockaddr_in sin;
            memset((char *) &sin, 0, sizeof(sin));
            sin.sin_family = AF_UNSPEC;
            sin.sin_addr.s_addr = INADDR_ANY;
            return socket_strerror(socket_connect(ps, (SA *) &sin, 
                sizeof(sin), tm));
        }
        case PF_INET6: {
            struct sockaddr_in6 sin6;
            struct in6_addr addrany = IN6ADDR_ANY_INIT; 
            memset((char *) &sin6, 0, sizeof(sin6));
            sin6.sin6_family = AF_UNSPEC;
fprintf(stderr, "disconnecting\n");
            sin6.sin6_addr = addrany;
            return socket_strerror(socket_connect(ps, (SA *) &sin6, 
                sizeof(sin6), tm));
        }
    }
    return NULL;
}

/*-------------------------------------------------------------------------*\
* Puts the sockt in listen mode
\*-------------------------------------------------------------------------*/
static int meth_listen(lua_State *L)
{
    p_tcp tcp = (p_tcp) auxiliar_checkclass(L, "tcp{master}", 1);
    int backlog = (int) luaL_optnumber(L, 2, 32);
    int err = socket_listen(&tcp->sock, backlog);
    if (err != IO_DONE) {
        lua_pushnil(L);
        lua_pushstring(L, socket_strerror(err));
        return 2;
    }
    /* turn master object into a server object */
    auxiliar_setclass(L, "tcp{server}", 1);
    lua_pushnumber(L, 1);
    return 1;
}

/*-------------------------------------------------------------------------*\
* Waits for and returns a client object attempting connection to the
* server object
\*-------------------------------------------------------------------------*/
static int meth_acceptfd(lua_State *L)
{
    p_tcp server = (p_tcp) auxiliar_checkclass(L, "tcp{server}", 1);
    p_timeout tm = timeout_markstart(&server->tm);
    t_socket sock;
    int err = socket_accept(&server->sock, &sock, NULL, NULL, tm);
    /* if successful, push client socket */
    if (err == IO_DONE) {
        lua_pushnumber(L, sock);
        return 1;
    } else {
        lua_pushnil(L);
        lua_pushstring(L, socket_strerror(err));
        return 2;
    }
}

/*-------------------------------------------------------------------------*\
* Tries to connect to remote address (address, port)
\*-------------------------------------------------------------------------*/
const char *inet_tryconnect(p_socket ps, int *family, const char *address,
        const char *serv, p_timeout tm, struct addrinfo *connecthints)
{
#ifdef LUASOCKET_SECURITY_SANDBOX
    if (luasocket_ip_allowed(address))
        return "connect restricted";
#endif // LUASOCKET_SECURITY_SANDBOX
    struct addrinfo *iterator = NULL, *resolved = NULL;
    const char *err = NULL;
    int current_family = *family;
    /* try resolving */
    err = socket_gaistrerror(getaddrinfo(address, serv,
                connecthints, &resolved));
    if (err != NULL) {
        if (resolved) freeaddrinfo(resolved);
        return err;
    }
    for (iterator = resolved; iterator; iterator = iterator->ai_next) {
        timeout_markstart(tm);
        /* create new socket if necessary. if there was no
         * bind, we need to create one for every new family
         * that shows up while iterating. if there was a
         * bind, all families will be the same and we will
         * not enter this branch. */
        if (current_family != iterator->ai_family || *ps == SOCKET_INVALID) {
            socket_destroy(ps);
            err = inet_trycreate(ps, iterator->ai_family,
                iterator->ai_socktype, iterator->ai_protocol);
            if (err) continue;
            current_family = iterator->ai_family;
            /* set non-blocking before connect */
            socket_setnonblocking(ps);
        }
        /* try connecting to remote address */
        err = socket_strerror(socket_connect(ps, (SA *) iterator->ai_addr,
            (socklen_t) iterator->ai_addrlen, tm));
        /* if success or timeout is zero, break out of loop */
        if (err == NULL || timeout_iszero(tm)) {
            *family = current_family;
            break;
        }
    }
    freeaddrinfo(resolved);
    /* here, if err is set, we failed */
    return err;
}

static ssize_t
rd_kafka_transport_socket_recvmsg (rd_kafka_transport_t *rktrans,
				   struct msghdr *msg,
				   char *errstr, size_t errstr_size) {
#ifndef _MSC_VER
	ssize_t r;
#ifdef sun
	/* SunOS doesn't seem to set errno when recvmsg() fails
	 * due to no data and MSG_DONTWAIT is set. */
	socket_errno = EAGAIN;
#endif
	r = recvmsg(rktrans->rktrans_s, msg, MSG_DONTWAIT);
	if (r == -1 && socket_errno == EAGAIN)
		return 0;
	else if (r == 0) {
		/* Receive 0 after POLLIN event means connection closed. */
		rd_snprintf(errstr, errstr_size, "Disconnected");
		return -1;
	} else if (r == -1)
		rd_snprintf(errstr, errstr_size, "%s", rd_strerror(errno));

	return r;
#else
	ssize_t sum = 0;
	int i;

	for (i = 0; i < msg->msg_iovlen; i++) {
		ssize_t r;

		r = recv(rktrans->rktrans_s,
			 msg->msg_iov[i].iov_base, (int) msg->msg_iov[i].iov_len, 0);
		if (r == SOCKET_ERROR) {
			if (WSAGetLastError() == WSAEWOULDBLOCK)
				break;
			rd_snprintf(errstr, errstr_size, "%s", socket_strerror(WSAGetLastError()));
			return -1;
		}
		sum += r;
		if ((size_t)r < msg->msg_iov[i].iov_len)
			break;
	}
	return sum;
#endif
}

/**
 * Map error code into string.
 */
static const char *ssl_ioerror(void *ctx, int err)
{
  if (err == IO_SSL) {
    p_ssl ssl = (p_ssl) ctx;
    switch(ssl->error) {
    case SSL_ERROR_NONE: return "No error";
    case SSL_ERROR_ZERO_RETURN: return "closed";
    case SSL_ERROR_WANT_READ: return "wantread";
    case SSL_ERROR_WANT_WRITE: return "wantwrite";
    case SSL_ERROR_WANT_CONNECT: return "'connect' not completed";
    case SSL_ERROR_WANT_ACCEPT: return "'accept' not completed";
    case SSL_ERROR_WANT_X509_LOOKUP: return "Waiting for callback";
    case SSL_ERROR_SYSCALL: return "System error";
    case SSL_ERROR_SSL: return ERR_reason_error_string(ERR_get_error());
    default: return "Unknown SSL error";
    }
  }
  return socket_strerror(err);
}

/*-------------------------------------------------------------------------*\
* Binds an object to an address 
\*-------------------------------------------------------------------------*/
static const char *unix_trybind(p_unix un, const char *path) {
    struct sockaddr_un local;
    size_t len = strlen(path);
    int err;
    if (len >= sizeof(local.sun_path)) return "path too long";
    memset(&local, 0, sizeof(local));
    strcpy(local.sun_path, path);
    local.sun_family = AF_UNIX;
#ifdef UNIX_HAS_SUN_LEN
    local.sun_len = sizeof(local.sun_family) + sizeof(local.sun_len) 
        + len + 1;
    err = socket_bind(&un->sock, (SA *) &local, local.sun_len);

#else 
    err = socket_bind(&un->sock, (SA *) &local, 
            sizeof(local.sun_family) + len);
#endif
    if (err != IO_DONE) socket_destroy(&un->sock);
    return socket_strerror(err); 
}

int rd_kafka_transport_poll(rd_kafka_transport_t *rktrans, int tmout) {
#ifndef _MSC_VER
	int r;

	r = poll(&rktrans->rktrans_pfd, 1, tmout);
	if (r <= 0)
		return r;
	return rktrans->rktrans_pfd.revents;
#else
	int r;

	r = WSAPoll(&rktrans->rktrans_pfd, 1, tmout);
	if (r == 0) {
		/* Workaround for broken WSAPoll() while connecting:
		 * failed connection attempts are not indicated at all by WSAPoll()
		 * so we need to check the socket error when Poll returns 0.
		 * Issue #525 */
		r = ECONNRESET;
		if (unlikely(rktrans->rktrans_rkb->rkb_state ==
			     RD_KAFKA_BROKER_STATE_CONNECT &&
			     (rd_kafka_transport_get_socket_error(rktrans,
								  &r) == -1 ||
			      r != 0))) {
			char errstr[512];
			errno = r;
			rd_snprintf(errstr, sizeof(errstr),
				    "Connect to %s failed: %s",
				    rd_sockaddr2str(rktrans->rktrans_rkb->
						    rkb_addr_last,
						    RD_SOCKADDR2STR_F_PORT |
                                                    RD_SOCKADDR2STR_F_FAMILY),
                                    socket_strerror(r));
			rd_kafka_transport_connect_done(rktrans, errstr);
			return -1;
		} else
			return 0;
	} else if (r == SOCKET_ERROR)
		return -1;
	return rktrans->rktrans_pfd.revents;
#endif
}

/*-------------------------------------------------------------------------*\
* Tries to connect to remote address (address, port)
\*-------------------------------------------------------------------------*/
const char *inet_tryconnect(p_socket ps, const char *address,
                            unsigned short port, p_timeout tm)
{
    struct sockaddr_in remote;
    int err;
    memset(&remote, 0, sizeof(remote));
    remote.sin_family = AF_INET;
    remote.sin_port = htons(port);
    if (strcmp(address, "*")) {
        if (!inet_aton(address, &remote.sin_addr)) {
            struct hostent *hp = NULL;
            struct in_addr **addr;
            err = socket_gethostbyname(address, &hp);
            if (err != IO_DONE) return socket_hoststrerror(err);
            addr = (struct in_addr **) hp->h_addr_list;
            memcpy(&remote.sin_addr, *addr, sizeof(struct in_addr));
        }
    } else remote.sin_family = AF_UNSPEC;
    err = socket_connect(ps, (SA *) &remote, sizeof(remote), tm);
    return socket_strerror(err);
}

int main(int ac, char **av) {
  int rc, i;

#ifdef WIN32
  win_init();
#endif

  for (i = 0; TestCases[i] != NULL; i++) {
    const struct test_case *current = TestCases[i];
    const char *name = get_test_name(current);

    printf("%-48s", name);
    fflush(stdout);
    rc = test_case_run(current);
    if (rc) {
      printf(TEST_FAILED " (%s)\n", socket_strerror(-rc));
      break;
    }
    printf(TEST_OK "\n");
  }
  return rc;
}

/*-------------------------------------------------------------------------*\
* Waits for and returns a client object attempting connection to the 
* server object 
\*-------------------------------------------------------------------------*/
static int meth_accept(lua_State *L) {
    p_unix server = (p_unix) auxiliar_checkclass(L, "unix{server}", 1);
    p_timeout tm = timeout_markstart(&server->tm);
    t_socket sock;
    int err = socket_accept(&server->sock, &sock, NULL, NULL, tm);
    /* if successful, push client socket */
    if (err == IO_DONE) {
        p_unix clnt = (p_unix) lua_newuserdata(L, sizeof(t_unix));
        auxiliar_setclass(L, "unix{client}", -1);
        /* initialize structure fields */
        socket_setnonblocking(&sock);
        clnt->sock = sock;
        io_init(&clnt->io, (p_send)socket_send, (p_recv)socket_recv, 
                (p_error) socket_ioerror, &clnt->sock);
        timeout_init(&clnt->tm, -1, -1);
        buffer_init(&clnt->buf, &clnt->io, &clnt->tm);
        return 1;
    } else {
        lua_pushnil(L); 
        lua_pushstring(L, socket_strerror(err));
        return 2;
    }
}

/*-------------------------------------------------------------------------*\
* Creates a master unix object 
\*-------------------------------------------------------------------------*/
static int global_create(lua_State *L) {
    t_socket sock;
    int err = socket_create(&sock, AF_UNIX, SOCK_STREAM, 0);
    /* try to allocate a system socket */
    if (err == IO_DONE) { 
        /* allocate unix object */
        p_unix un = (p_unix) lua_newuserdata(L, sizeof(t_unix));
        /* set its type as master object */
        auxiliar_setclass(L, "unix{master}", -1);
        /* initialize remaining structure fields */
        socket_setnonblocking(&sock);
        un->sock = sock;
        io_init(&un->io, (p_send) socket_send, (p_recv) socket_recv, 
                (p_error) socket_ioerror, &un->sock);
        timeout_init(&un->tm, -1, -1);
        buffer_init(&un->buf, &un->io, &un->tm);
        return 1;
    } else {
        lua_pushnil(L);
        lua_pushstring(L, socket_strerror(err));
        return 2;
    }
}

/*-------------------------------------------------------------------------*\
* Tries to connect to remote address (address, port)
\*-------------------------------------------------------------------------*/
const char *inet_tryconnect(p_socket ps, const char *address,
        const char *serv, p_timeout tm, struct addrinfo *connecthints)
{
    struct addrinfo *iterator = NULL, *resolved = NULL;
    const char *err = NULL;
    /* try resolving */
    err = socket_gaistrerror(getaddrinfo(address, serv,
                connecthints, &resolved));
    if (err != NULL) {
        if (resolved) freeaddrinfo(resolved);
        return err;
    }
    for (iterator = resolved; iterator; iterator = iterator->ai_next) {
        timeout_markstart(tm);
        /* try connecting to remote address */
        err = socket_strerror(socket_connect(ps, (SA *) iterator->ai_addr, 
            (socklen_t) iterator->ai_addrlen, tm));
        /* if success, break out of loop */
        if (err == NULL) break;
    }
    freeaddrinfo(resolved);
    /* here, if err is set, we failed */
    return err;
}

/* handle_connect_results assumes that select or poll have already shown the
 * descriptor to be active */
void handle_connect_result(struct npool *ms, struct nevent *nse, enum nse_status status) {
  int optval;
  socklen_t optlen = sizeof(int);
  struct niod *iod = nse->iod;
#if HAVE_OPENSSL
  int sslerr;
  int rc = 0;
  int sslconnect_inprogress = nse->type == NSE_TYPE_CONNECT_SSL && nse->iod &&
    (nse->sslinfo.ssl_desire == SSL_ERROR_WANT_READ ||
     nse->sslinfo.ssl_desire == SSL_ERROR_WANT_WRITE);
#else
  int sslconnect_inprogress = 0;
#endif

  if (status == NSE_STATUS_TIMEOUT || status == NSE_STATUS_CANCELLED) {
    nse->status = status;
    nse->event_done = 1;
  } else if (sslconnect_inprogress) {
    /* Do nothing */
  } else if (status == NSE_STATUS_SUCCESS) {
    /* First we want to determine whether the socket really is connected */
    if (getsockopt(iod->sd, SOL_SOCKET, SO_ERROR, (char *)&optval, &optlen) != 0)
      optval = socket_errno(); /* Stupid Solaris */

    switch (optval) {
      case 0:
        nse->status = NSE_STATUS_SUCCESS;
        break;
      /* EACCES can be caused by ICMPv6 dest-unreach-admin, or when a port is
         blocked by Windows Firewall (WSAEACCES). */
      case EACCES:
      case ECONNREFUSED:
      case EHOSTUNREACH:
      case ENETDOWN:
      case ENETUNREACH:
      case ENETRESET:
      case ECONNABORTED:
      case ETIMEDOUT:
      case EHOSTDOWN:
      case ECONNRESET:
#ifdef WIN32
      case WSAEADDRINUSE:
      case WSAEADDRNOTAVAIL:
#endif
#ifndef WIN32
      case EPIPE: /* Has been seen after connect on Linux. */
      case ENOPROTOOPT: /* Also seen on Linux, perhaps in response to protocol unreachable. */
#endif
        nse->status = NSE_STATUS_ERROR;
        nse->errnum = optval;
        break;

      default:
        /* I'd like for someone to report it */
        fatal("Strange connect error from %s (%d): %s",
              inet_ntop_ez(&iod->peer, iod->peerlen), optval,
              socket_strerror(optval));
    }

    /* Now special code for the SSL case where the TCP connection was successful. */
    if (nse->type == NSE_TYPE_CONNECT_SSL &&
        nse->status == NSE_STATUS_SUCCESS) {
#if HAVE_OPENSSL
      assert(ms->sslctx != NULL);
      /* Reuse iod->ssl if present. If set, this is the second try at connection
         without the SSL_OP_NO_SSLv2 option set. */
      if (iod->ssl == NULL) {
        iod->ssl = SSL_new(ms->sslctx);
        if (!iod->ssl)
          fatal("SSL_new failed: %s", ERR_error_string(ERR_get_error(), NULL));
      }

#if HAVE_SSL_SET_TLSEXT_HOST_NAME
      if (iod->hostname != NULL) {
        if (SSL_set_tlsext_host_name(iod->ssl, iod->hostname) != 1)
          fatal("SSL_set_tlsext_host_name failed: %s", ERR_error_string(ERR_get_error(), NULL));
      }
#endif

      /* Associate our new SSL with the connected socket.  It will inherit the
       * non-blocking nature of the sd */
      if (SSL_set_fd(iod->ssl, iod->sd) != 1)
        fatal("SSL_set_fd failed: %s", ERR_error_string(ERR_get_error(), NULL));

      /* Event not done -- need to do SSL connect below */
      nse->sslinfo.ssl_desire = SSL_ERROR_WANT_CONNECT;
#endif
    } else {
      /* This is not an SSL connect (in which case we are always done), or the
       * TCP connect() underlying the SSL failed (in which case we are also done */
      nse->event_done = 1;
    }
  } else {
    fatal("Unknown status (%d)", status);
  }

  /* At this point the TCP connection is done, whether successful or not.
   * Therefore decrease the read/write listen counts that were incremented in
//.........这里部分代码省略.........

/* An event has been completed and the handler is about to be called. This
 * function writes out tracing data about the event if necessary */
void nsock_trace_handler_callback(struct npool *ms, struct nevent *nse) {
  struct niod *nsi;
  char *str;
  int strlength = 0;
  char displaystr[256];
  char errstr[256];

  if (NsockLogLevel > NSOCK_LOG_INFO)
    return;

  nsi = nse->iod;

  if (nse->status == NSE_STATUS_ERROR)
    Snprintf(errstr, sizeof(errstr), "[%s (%d)] ", socket_strerror(nse->errnum),
             nse->errnum);
  else
    errstr[0] = '\0';

  /* Some types have special tracing treatment */
  switch (nse->type) {
    case NSE_TYPE_CONNECT:
    case NSE_TYPE_CONNECT_SSL:
      nsock_log_info("Callback: %s %s %sfor EID %li [%s]",
                     nse_type2str(nse->type), nse_status2str(nse->status),
                     errstr, nse->id, get_peeraddr_string(nsi));
      break;

    case NSE_TYPE_READ:
      if (nse->status != NSE_STATUS_SUCCESS) {
        nsock_log_info("Callback: %s %s %sfor EID %li [%s]",
                       nse_type2str(nse->type), nse_status2str(nse->status),
                       errstr, nse->id, get_peeraddr_string(nsi));
      } else {
        str = nse_readbuf(nse, &strlength);
        if (strlength < 80) {
          memcpy(displaystr, ": ", 2);
          memcpy(displaystr + 2, str, strlength);
          displaystr[2 + strlength] = '\0';
          replacenonprintable(displaystr + 2, strlength, '.');
        } else {
          displaystr[0] = '\0';
        }
        nsock_log_info("Callback: %s %s for EID %li [%s] %s(%d bytes)%s",
                       nse_type2str(nse->type), nse_status2str(nse->status),
                       nse->id,
                       get_peeraddr_string(nsi),
                       nse_eof(nse) ? "[EOF]" : "", strlength, displaystr);
      }
      break;

    case NSE_TYPE_WRITE:
      nsock_log_info("Callback: %s %s %sfor EID %li [%s]",
                     nse_type2str(nse->type), nse_status2str(nse->status),
                     errstr, nse->id, get_peeraddr_string(nsi));
      break;

    case NSE_TYPE_TIMER:
      nsock_log_info("Callback: %s %s %sfor EID %li",
                     nse_type2str(nse->type), nse_status2str(nse->status),
                     errstr, nse->id);
      break;

#if HAVE_PCAP
    case NSE_TYPE_PCAP_READ:
      nsock_log_info("Callback: %s %s %sfor EID %li ",
                     nse_type2str(nse->type), nse_status2str(nse->status),
                     errstr, nse->id);
      break;
#endif

    default:
      fatal("Invalid nsock event type (%d)", nse->type);
  }
}

/* An event has been completed and the handler is about to be called. This
 * function writes out tracing data about the event if necessary */
void nsock_trace_handler_callback(mspool *ms, msevent *nse) {
  msiod *nsi;
  char *str;
  int strlength = 0;
  char displaystr[256];
  char errstr[256];

  if (ms->tracelevel == 0)
    return;

  nsi = nse->iod;

  if (nse->status == NSE_STATUS_ERROR)
    Snprintf(errstr, sizeof(errstr), "[%s (%d)] ", socket_strerror(nse->errnum), nse->errnum);
  else
    errstr[0] = '\0';

  /* Some types have special tracing treatment */
  switch(nse->type) {
    case NSE_TYPE_CONNECT:
    case NSE_TYPE_CONNECT_SSL:
      nsock_trace(ms, "Callback: %s %s %sfor EID %li [%s]",
                  nse_type2str(nse->type), nse_status2str(nse->status),
                  errstr, nse->id, get_peeraddr_string(nsi));
      break;

    case NSE_TYPE_READ:
      if (nse->status != NSE_STATUS_SUCCESS) {
        if (nsi->peerlen > 0) {
          nsock_trace(ms, "Callback: %s %s %sfor EID %li [%s]",
                      nse_type2str(nse->type), nse_status2str(nse->status),
                      errstr, nse->id, get_peeraddr_string(nsi));
        } else {
          nsock_trace(ms, "Callback: %s %s %sfor EID %li (peer unspecified)",
                      nse_type2str(nse->type), nse_status2str(nse->status),
                      errstr, nse->id);
        }
      } else {
        str = nse_readbuf(nse, &strlength);
        if (ms->tracelevel > 1 && strlength < 80) {
          memcpy(displaystr, ": ", 2);
          memcpy(displaystr + 2, str, strlength);
          displaystr[2 + strlength] = '\0';
          replacenonprintable(displaystr + 2, strlength, '.');
        } else {
          displaystr[0] = '\0';
        }

        if (nsi->peerlen > 0) {
          nsock_trace(ms, "Callback: %s %s for EID %li [%s] %s(%d bytes)%s",
                      nse_type2str(nse->type), nse_status2str(nse->status),
                      nse->id, get_peeraddr_string(nsi),
                      nse_eof(nse)? "[EOF]" : "", strlength, displaystr);
        } else {
          nsock_trace(ms, "Callback %s %s for EID %li (peer unspecified) %s(%d bytes)%s",
                      nse_type2str(nse->type), nse_status2str(nse->status),
                      nse->id, nse_eof(nse)? "[EOF]" : "", strlength, displaystr);
        }
      }
      break;

    case NSE_TYPE_WRITE:
      nsock_trace(ms, "Callback: %s %s %sfor EID %li [%s]",
                  nse_type2str(nse->type), nse_status2str(nse->status), errstr,
                  nse->id, get_peeraddr_string(nsi));
      break;

    case NSE_TYPE_TIMER:
      nsock_trace(ms, "Callback: %s %s %sfor EID %li",
                  nse_type2str(nse->type), nse_status2str(nse->status), errstr,
                  nse->id);
      break;

#if HAVE_PCAP
    case NSE_TYPE_PCAP_READ:
      nsock_trace(ms, "Callback: %s %s %sfor EID %li ",
                  nse_type2str(nse->type), nse_status2str(nse->status),
                  errstr, nse->id);
      break;
#endif

    default:
      assert(0);
      break;
  }
}

/*=========================================================================*\
* Lua methods
\*=========================================================================*/
const char *udp_strerror(int err) {
    /* a 'closed' error on an unconnected means the target address was not
     * accepted by the transport layer */
    if (err == IO_CLOSED) return "refused";
    else return socket_strerror(err);
}

//.........这里部分代码省略.........
    if(!TEST_NEQ(data, NULL)) {
        TEST_RETURN();
    }
    // Tell the host test to try and connect
    TEST_PRINT(">>> EC,%s,%d\r\n", local_addr, port);

    bool quit = false;
    // For several iterations
    while (!quit) {
        timedout = false;
        server_event_done = false;
        incoming = false;
        to.attach(onTimeout, SOCKET_TEST_SERVER_TIMEOUT);
        // Listen for incoming connections
        err = api->start_listen(&s, 0);
        if (!TEST_EQ(err, SOCKET_ERROR_NONE)) {
            TEST_EXIT();
        }
        // Reset the state of client_rx_done
        client_rx_done = false;
        // Wait for a connect event
        while (!timedout && !incoming) {
            __WFI();
        }
        if (TEST_EQ(timedout,0)) {
            to.detach();
        } else {
            TEST_EXIT();
        }
        if(!TEST_EQ(server_event.event, SOCKET_EVENT_ACCEPT)) {
            TEST_PRINT("Event: %d\r\n",(int)client_event.event);
            continue;
        }
        // Stop listening
        server_event_done = false;
        // err = api->stop_listen(&s);
        // TEST_EQ(err, SOCKET_ERROR_NONE);
        // Accept an incoming connection
        cs.impl = server_event.i.a.newimpl;
        cs.family = s.family;
        cs.stack  = s.stack;
        err = api->accept(&cs, client_cb);
        if(!TEST_EQ(err, SOCKET_ERROR_NONE)) {
            continue;
        }
        to.attach(onTimeout, SOCKET_TEST_SERVER_TIMEOUT);

                    // Client should test for successive connections being rejected
        // Until Client disconnects
        while (client_event.event != SOCKET_EVENT_ERROR && client_event.event != SOCKET_EVENT_DISCONNECT) {
            // Wait for a read event
            while (!client_event_done && !client_rx_done && !timedout) {
                __WFI();
            }
            if (!TEST_EQ(client_event_done, false)) {
                client_event_done = false;
                continue;
            }
            // Reset the state of client_rx_done
            client_rx_done = false;

            // Receive some data
            size_t len = SOCKET_SENDBUF_MAXSIZE;
            err = api->recv(&cs, data, &len);
            if (!TEST_NEQ(err, SOCKET_ERROR_WOULD_BLOCK)) {
            	TEST_PRINT("Rx Would Block\r\n");
            	continue;
            }
            if (!TEST_EQ(err, SOCKET_ERROR_NONE)) {
            	TEST_PRINT("err: (%d) %s\r\n", err, socket_strerror(err));
                break;
            }

            // Check if the server should halt
            if (strncmp((const char *)data, "quit", 4) == 0) {
                quit = true;
                break;
            }
            // Send some data
            err = api->send(&cs, data, len);
            if (!TEST_EQ(err, SOCKET_ERROR_NONE)) {
                break;
            }
        }
        to.detach();
        TEST_NEQ(timedout, true);

        // Close client socket
        err = api->close(&cs);
        TEST_EQ(err, SOCKET_ERROR_NONE);
    }
    err = api->stop_listen(&s);
    TEST_EQ(err, SOCKET_ERROR_NONE);
test_exit:
    ticker.detach();
    TEST_PRINT(">>> KILL,EC\r\n");
    free(data);
    // Destroy server socket
    TEST_RETURN();
}

int kqueue_loop(mspool *nsp, int msec_timeout) {
  int results_left = 0;
  int event_msecs; /* msecs before an event goes off */
  int combined_msecs;
  struct timespec ts, *ts_p;
  int sock_err = 0;
  struct kqueue_engine_info *kinfo = (struct kqueue_engine_info *)nsp->engine_data;

  assert(msec_timeout >= -1);

  if (nsp->events_pending == 0)
    return 0; /* No need to wait on 0 events ... */


  if (gh_list_count(&nsp->active_iods) > kinfo->evlen) {
    kinfo->evlen = gh_list_count(&nsp->active_iods) * 2;
    kinfo->events = (struct kevent *)safe_realloc(kinfo->events, kinfo->evlen * sizeof(struct kevent));
  }

  do {
    msevent *nse;

    nsock_log_debug_all(nsp, "wait for events");

    nse = next_expirable_event(nsp);
    if (!nse)
      event_msecs = -1; /* None of the events specified a timeout */
    else
      event_msecs = MAX(0, TIMEVAL_MSEC_SUBTRACT(nse->timeout, nsock_tod));

#if HAVE_PCAP
#ifndef PCAP_CAN_DO_SELECT
    /* Force a low timeout when capturing packets on systems where
     * the pcap descriptor is not select()able. */
    if (gh_list_count(&nsp->pcap_read_events) > 0)
      if (event_msecs > PCAP_POLL_INTERVAL)
        event_msecs = PCAP_POLL_INTERVAL;
#endif
#endif

    /* We cast to unsigned because we want -1 to be very high (since it means no
     * timeout) */
    combined_msecs = MIN((unsigned)event_msecs, (unsigned)msec_timeout);

    /* Set up the timeval pointer we will give to kevent() */
    memset(&ts, 0, sizeof(struct timespec));
    if (combined_msecs >= 0) {
      ts.tv_sec = combined_msecs / 1000;
      ts.tv_nsec = (combined_msecs % 1000) * 1000000L;
      ts_p = &ts;
    } else {
      ts_p = NULL;
    }

#if HAVE_PCAP
#ifndef PCAP_CAN_DO_SELECT
    /* do non-blocking read on pcap devices that doesn't support select()
     * If there is anything read, just leave this loop. */
    if (pcap_read_on_nonselect(nsp)) {
      /* okay, something was read. */
    } else
#endif
#endif
    {
      results_left = kevent(kinfo->kqfd, NULL, 0, kinfo->events, kinfo->evlen, ts_p);
      if (results_left == -1)
        sock_err = socket_errno();
    }

    gettimeofday(&nsock_tod, NULL); /* Due to kevent delay */
  } while (results_left == -1 && sock_err == EINTR); /* repeat only if signal occurred */

  if (results_left == -1 && sock_err != EINTR) {
    nsock_log_error(nsp, "nsock_loop error %d: %s", sock_err, socket_strerror(sock_err));
    nsp->errnum = sock_err;
    return -1;
  }

  iterate_through_event_lists(nsp, results_left);

  return 1;
}

static int ncat_listen_stream(int proto)
{
    int rc, i, fds_ready;
    fd_set listen_fds;
    struct timeval tv;
    struct timeval *tvp = NULL;
    unsigned int num_sockets;

    /* clear out structs */
    FD_ZERO(&master_readfds);
    FD_ZERO(&master_writefds);
    FD_ZERO(&master_broadcastfds);
    FD_ZERO(&listen_fds);
#ifdef HAVE_OPENSSL
    FD_ZERO(&sslpending_fds);
#endif
    zmem(&client_fdlist, sizeof(client_fdlist));
    zmem(&broadcast_fdlist, sizeof(broadcast_fdlist));

#ifdef WIN32
    set_pseudo_sigchld_handler(decrease_conn_count);
#else
    /* Reap on SIGCHLD */
    Signal(SIGCHLD, sigchld_handler);
    /* Ignore the SIGPIPE that occurs when a client disconnects suddenly and we
       send data to it before noticing. */
    Signal(SIGPIPE, SIG_IGN);
#endif

#ifdef HAVE_OPENSSL
    if (o.ssl)
        setup_ssl_listen();
#endif

/* Not sure if this problem exists on Windows, but fcntl and /dev/null don't */
#ifndef WIN32
    /* Check whether stdin is closed. Because we treat this fd specially, we
     * can't risk it being reopened for an incoming connection, so we'll hold
     * it open instead. */
    if (fcntl(STDIN_FILENO, F_GETFD) == -1 && errno == EBADF) {
      logdebug("stdin is closed, attempting to reserve STDIN_FILENO\n");
      rc = open("/dev/null", O_RDONLY);
      if (rc >= 0 && rc != STDIN_FILENO) {
        /* Oh well, we tried */
        logdebug("Couldn't reserve STDIN_FILENO\n");
        close(rc);
      }
    }
#endif

    /* We need a list of fds to keep current fdmax. The second parameter is a
       number added to the supplied connection limit, that will compensate
       maxfds for the added by default listen and stdin sockets. */
    init_fdlist(&client_fdlist, sadd(o.conn_limit, num_listenaddrs + 1));

    for (i = 0; i < NUM_LISTEN_ADDRS; i++)
        listen_socket[i] = -1;

    num_sockets = 0;
    for (i = 0; i < num_listenaddrs; i++) {
        /* setup the main listening socket */
        listen_socket[num_sockets] = do_listen(SOCK_STREAM, proto, &listenaddrs[i]);
        if (listen_socket[num_sockets] == -1) {
            if (o.debug > 0)
                logdebug("do_listen(\"%s\"): %s\n", inet_ntop_ez(&listenaddrs[i].storage, sizeof(listenaddrs[i].storage)), socket_strerror(socket_errno()));
            continue;
        }

        /* Make our listening socket non-blocking because there are timing issues
         * which could cause us to block on accept() even though select() says it's
         * readable.  See UNPv1 2nd ed, p422 for more.
         */
        unblock_socket(listen_socket[num_sockets]);

        /* setup select sets and max fd */
        FD_SET(listen_socket[num_sockets], &master_readfds);
        add_fd(&client_fdlist, listen_socket[num_sockets]);

        FD_SET(listen_socket[num_sockets], &listen_fds);

        num_sockets++;
    }
    if (num_sockets == 0) {
        if (num_listenaddrs == 1)
            bye("Unable to open listening socket on %s: %s", inet_ntop_ez(&listenaddrs[0].storage, sizeof(listenaddrs[0].storage)), socket_strerror(socket_errno()));
        else
            bye("Unable to open any listening sockets.");
    }

    add_fd(&client_fdlist, STDIN_FILENO);

    init_fdlist(&broadcast_fdlist, o.conn_limit);

    if (o.idletimeout > 0)
        tvp = &tv;

    while (1) {
        /* We pass these temporary descriptor sets to fselect, since fselect
           modifies the sets it receives. */
        fd_set readfds = master_readfds, writefds = master_writefds;
//.........这里部分代码省略.........