int main(int argv,char **args)
{
	int ret,epfd,nfds,server_socket_fd,i,index;
	struct epoll_event events[256];
	int count;	

	server_socket_fd = server_start();	
	if(server_socket_fd<0)
	{
		return 0;
	}
	set_non_blocking(server_socket_fd);

	epfd = epoll_create(1024);
	set_non_blocking(epfd);
	epoll_add(epfd,server_socket_fd);

	struct http_request_r *r=NULL,*p=NULL;

        while(1)
        {
                nfds=epoll_wait(epfd,events,20,500);
                for(i=0;i<nfds;i++)
                {
                        if(events[i].data.fd==server_socket_fd)
                        {
				onAccept(epfd,events[i],&r);
			}
                        else if(events[i].events&EPOLLIN)
                        {
				onData(epfd,events[i],&r);
                        }
                }
        }
}

// connect to remote server timeout call.
bool connect_timeout(int sockfd, const struct sockaddr *addr, socklen_t addrlen, struct timeval* tv_to)
{
    bool result = false;
    set_non_blocking(sockfd, true);
    int res = connect(sockfd, addr, addrlen);
    if ( res < 0 )
    {
        if ( errno == EINPROGRESS )
        {
            fd_set set;
            FD_ZERO(&set);
            FD_SET(sockfd, &set);
            if ( select(sockfd + 1, NULL, &set, NULL, tv_to) > 0 )
            {
                int valopt;
                int lon = sizeof(int);
                getsockopt(sockfd, SOL_SOCKET, SO_ERROR, (void*)(&valopt), &lon);
                if ( valopt )
                {
                    logPrint("connect_timeout error in connect(%d): %s", sockfd, strerror(errno));
                }
            }
            else
            {
                logPrint("connect_timeout select(%d) timeout or error: %s", sockfd, strerror(errno));
            }
        }
        else
        {
            logPrint("connect_timeout connect(%d) error: %s", sockfd, strerror(errno));
        }
    }
    set_non_blocking(sockfd, false);
    return true;
}

int ssl_client_handshake(struct Client *cptr) {

  char *str;
  int err;

  cptr->ssl = (struct SSL*)SSL_new (ctx);                         CHK_NULL(cptr->ssl);    
  SSL_set_fd ((SSL*)cptr->ssl, cptr->fd);
  set_blocking(cptr->fd);
  err = SSL_connect ((SSL*)cptr->ssl);                     
  set_non_blocking(cptr->fd);
  if ((err)==-1) {
     irclog(L_NOTICE,"Could connect to %s:Error in SSL_connect()", 
                  get_client_name(cptr, TRUE));
     return 0;
     }
    
  /* Following two steps are optional and not required for
     data exchange to be successful. */
  
  /* Get the cipher - opt */
  set_blocking(cptr->fd);
  irclog (L_NOTICE,"SSL connection using %s", SSL_get_cipher ((SSL*)cptr->ssl));
  
  
  SetSecure(cptr);      
  /* Get server's certificate (note: beware of dynamic allocation) - opt */

  cptr->client_cert = (struct X509*)SSL_get_peer_certificate ((SSL *)cptr->ssl);
  set_non_blocking(cptr->fd);

  if (cptr->client_cert != NULL)
    {
      irclog (L_NOTICE,"Server certificate:");
  
      str = X509_NAME_oneline (X509_get_subject_name ((X509*)cptr->client_cert),0,0);
      CHK_NULL(str);
      irclog (L_NOTICE, "\t subject: %s", str);
      // Free (str);
      free (str);
  
      str = X509_NAME_oneline (X509_get_issuer_name  ((X509*)cptr->client_cert),0,0);
      CHK_NULL(str);
      irclog (L_NOTICE, "\t issuer: %s", str);
      // Free (str);
      free (str);
      /* We could do all sorts of certificate verification stuff here before
      deallocating the certificate. */

      X509_free ((X509*)cptr->client_cert);

    }
    else
      irclog (L_NOTICE, "Server does not have certificate.");
      
  
  return 1;
}

int
main(int argc, const char* const* argv)
{
    printf("GIMME YOUR INPUTS!\n");

    // set stdin/out to nonblocking
    set_non_blocking(STDIN_FILENO);
    set_non_blocking(STDOUT_FILENO);

    loop = uv_default_loop();
    uv_poll_init(loop, &stdin_watcher, STDIN_FILENO);
    uv_poll_init(loop, &stdout_watcher, STDOUT_FILENO);
    uv_poll_start(&stdin_watcher, UV_READABLE, stdin_cb);
    uv_run(loop, 0);
    return 0;
}

int main(void) 
{
    int sock_fd = -1;
    char buffer[1024];
    int bytes_recv = -1;
    
    log_set_level(LOG_LEVEL_DEBUG);

    sock_fd = udp_unix_socket_create(UNIX_DOMAIN_PATH);
    if (sock_fd < 0) {
        log_error("udp_unix_socket_create failed.");
        return 0;
    }

    set_non_blocking(sock_fd); 
    while(1) {
        bzero(buffer, sizeof(buffer));
        bytes_recv = udp_unix_socket_recv(sock_fd, buffer, sizeof(buffer) - 1);
        if (bytes_recv > 0) {
            log_debug("%d %s", bytes_recv, buffer);
        }
        else {
            log_info("no data received.");
        }
    }

    udp_unix_socket_close(sock_fd); 

    return 1;
}

int
_do_controller_direct_open_with_line(char* linebuf, size_t size, sc_controller* contr)
{
    sc_follow_context* cxt = NULL;
    char *atbl[51], **ap = atbl, *p, *val;

    for (p = linebuf; p != NULL; ) {
        while ((val = strsep(&p, " \t")) != NULL && *val == '\0');
            *ap++ = val;
    }
    *ap = 0;

    ap = atbl;
    while ((*ap)[0] == '-') {
    }

    contr->displayName = strdup(ap[0]);
    contr->f_direct = 1;

    fprintf(stderr, "OPEN: displayName=(%s)\n", p);

    cxt = sc_follow_context_new(ap[1], ap[0], 1, BUFSIZE, g_connection);
    g_context_list = az_list_add(g_context_list, cxt);

    set_non_blocking(contr->socket_fd);

    return 0;
}

VALUE io_spec_rb_io_wait_readable(VALUE self, VALUE io, VALUE read_p) {
  int fd = io_spec_get_fd(io);
  set_non_blocking(fd);
  char buf[RB_IO_WAIT_READABLE_BUF];
  wait_bool ret;

  if(RTEST(read_p)) {
    rb_ivar_set(self, rb_intern("@write_data"), Qtrue);
    if(read(fd, buf, RB_IO_WAIT_READABLE_BUF) != -1) {
      return Qnil;
    }
  }

  ret = rb_io_wait_readable(fd);

  if(RTEST(read_p)) {
    if(read(fd, buf, RB_IO_WAIT_READABLE_BUF) != 13) {
      return Qnil;
    }
    rb_ivar_set(self, rb_intern("@read_data"),
        rb_str_new(buf, RB_IO_WAIT_READABLE_BUF));
  }

  return wait_bool_to_ruby_bool(ret);
}

int new_connection_handler(event_loop* el, int fd, kivi_client* client) {
	log_info("new connection %d", fd);
	int client_fd = accept(fd, NULL, NULL);
	if (client_fd > 0) {
		if (set_non_blocking(client_fd)) {
			log_warn("couldn't make client socket nonblocking %d", client_fd);
			return FAILED;
		}
		
		file_event file_ev;
		file_ev.mask = READ_EVENT | WRITE_EVENT;
		file_ev.read_handler = socket_reader;
		file_ev.write_handler = socket_writer;
		if (add_to_el(el, client_fd, file_ev) == FAILED) {
			return FAILED;
		} else {
			client = create_new_client(client_fd);
			server.clients[client_fd] = client;
		}
	} else {
		return FAILED;
	}

	return OK;
}

bool TransportUDP::initializeSocket()
{
  ROS_ASSERT(sock_ != ROS_INVALID_SOCKET);

  if (!(flags_ & SYNCHRONOUS))
  {
	  int result = set_non_blocking(sock_);
	  if ( result != 0 ) {
	      ROS_ERROR("setting socket [%d] as non_blocking failed with error [%d]", sock_, result);
      close();
      return false;
    }
  }

  socklen_t len = sizeof(local_address_);
  getsockname(sock_, (sockaddr *)&local_address_, &len);
  local_port_ = ntohs(local_address_.sin_port);

  ROS_ASSERT(poll_set_ || (flags_ & SYNCHRONOUS));
  if (poll_set_)
  {
    poll_set_->addSocket(sock_, boost::bind(&TransportUDP::socketUpdate, this, _1), shared_from_this());
  }

  return true;
}

/*
 * It is possible for a race condition to exist; such that select()
 * indicates that a listen()ing socket is able to recieve a new connection
 * and that a later accept() call will still block because the connection
 * has been closed in the interim.  This wrapper for accept() attempts to
 * defeat this by making the accept() call nonblocking.
 */
int	my_accept (int s, struct sockaddr *addr, int *addrlen)
{
	int	retval;
	set_non_blocking(s);
	retval = accept(s, addr, addrlen);
	set_blocking(s);
	return retval;
}

static int run_script_select(FILE **sout, FILE **serr, struct cntr *cntr, int logfunc)
{
	int mfd=-1;
	fd_set fsr;
	struct timeval tval;
	int soutfd=fileno(*sout);
	int serrfd=fileno(*serr);
	setlinebuf(*sout);
	setlinebuf(*serr);
	set_non_blocking(soutfd);
	set_non_blocking(serrfd);

	while(1)
	{
		mfd=-1;
		FD_ZERO(&fsr);
		if(*sout) add_fd_to_sets(soutfd, &fsr, NULL, NULL, &mfd);
		if(*serr) add_fd_to_sets(serrfd, &fsr, NULL, NULL, &mfd);
		tval.tv_sec=1;
		tval.tv_usec=0;
		if(select(mfd+1, &fsr, NULL, NULL, &tval)<0)
		{
			if(errno!=EAGAIN && errno!=EINTR)
			{
				logp("run_script_select error: %s\n",
					strerror(errno));
				return -1;
			}
		}
		if(FD_ISSET(soutfd, &fsr))
			log_script_output(sout, NULL, logfunc);
		if(FD_ISSET(serrfd, &fsr))
			log_script_output(serr, cntr, logfunc);

		if(!*sout && !*serr && got_sigchld)
		{
			//fclose(*sout); *sout=NULL;
			//fclose(*serr); *serr=NULL;
			got_sigchld=0;
			return 0;
		}
	}

	// Never get here.
	return -1;
}

static enum cliret initial_comms(struct async *as,
	enum action *action, char **incexc, struct conf **confs)
{
	struct asfd *asfd;
	char *server_version=NULL;
	enum cliret ret=CLIENT_OK;
	asfd=as->asfd;

	if(authorise_client(asfd, &server_version,
	  get_string(confs[OPT_CNAME]),
	  get_string(confs[OPT_PASSWORD]),
	  get_cntr(confs)))
		goto error;

	if(server_version)
	{
		logp("Server version: %s\n", server_version);
		// Servers before 1.3.2 did not tell us their versions.
		// 1.3.2 and above can do the automatic CA stuff that
		// follows.
		switch(ca_client_setup(asfd, confs))
		{
			case 0:
				break; // All OK.
			case 1:
				// Certificate signed successfully.
				// Everything is OK, but we will reconnect now,
				// in order to use the new keys/certificates.
				goto reconnect;
			default:
				logp("Error with cert signing request\n");
				goto error;
		}
	}

	set_non_blocking(asfd->fd);

	if(ssl_check_cert(asfd->ssl, NULL, confs))
	{
		logp("check cert failed\n");
		goto error;
	}

	if(extra_comms(as, confs, action, incexc))
	{
		logp("extra comms failed\n");
		goto error;
	}

	ret=CLIENT_OK; goto end;
error:
	ret=CLIENT_ERROR; goto end;
reconnect:
	ret=CLIENT_RECONNECT; goto end;
end:
	free_w(&server_version);
	return ret;
}

static int
init(char *server_addr, char *server_port, char *event_sub_server_port, int *listen_fd, int *event_sub_fd) {
  int ret;

  *listen_fd = create_listening_fd(server_addr, server_port);
  if(*listen_fd == -1) {
    fprintf(stderr, "I cannot create the listening socket\n");
    return -1;
  }
  if(set_non_blocking(*listen_fd) == 0)
    return -1;

  *event_sub_fd = create_listening_fd(server_addr, event_sub_server_port);
  if(*event_sub_fd == -1) {
    fprintf(stderr, "I cannot create the event subscribe listening socket\n");
    return -1;
  }
  if(set_non_blocking(*event_sub_fd) == 0)
    return -1;

  /* I get my architecture */
  if( (my_arch = aconv_get_host_arch()) == ACONV_ARCH_ERROR ) {
    fprintf(stderr, "I cannot get my architecture\n");
    return -1;
  }
  /* I init the librsc module */ 
  ret = rscs_init(my_arch);
  if(ret == -1) {
    fprintf(stderr, "I cannot initialize the RSC module\n");
    return -1;
  }
  GDEBUG(1, "My architecture is %s\n", aconv_arch2str(my_arch));
  
  /* I set the list of ioctl request I support AFTER the initialization
   * of the rsc_module (otherwise segfault)*/
  init_ioctl_register_request();
  
  /* I Init the pollfd structure */
  if((pfdinfo = pollfd_init()) == NULL) {
    fprintf(stderr, "I cannot init the pollfd structure\n");
    return -1;
  }
  
  return 0;
}

static enum cliret initial_comms(struct async *as,
	enum action *action, char **incexc, long *name_max, 
	struct conf *conf)
{
	struct asfd *asfd;
	char *server_version=NULL;
	enum cliret ret=CLIENT_OK;
	asfd=as->asfd;

	if(authorise_client(asfd, conf, &server_version))
		goto error;

	if(server_version)
	{
		int ca_ret=0;
		logp("Server version: %s\n", server_version);
		// Servers before 1.3.2 did not tell us their versions.
		// 1.3.2 and above can do the automatic CA stuff that
		// follows.
		if((ca_ret=ca_client_setup(asfd, conf))<0)
		{
			// Error
			logp("Error with certificate signing request\n");
			goto error;
		}
		else if(ca_ret>0)
		{
			// Certificate signed successfully.
			// Everything is OK, but we will reconnect now, in
			// order to use the new keys/certificates.
			goto reconnect;
		}
	}

	set_non_blocking(asfd->fd);

	if(ssl_check_cert(asfd->ssl, conf))
	{
		logp("check cert failed\n");
		goto error;
	}

	if(extra_comms(as, conf, action, incexc, name_max))
	{
		logp("extra comms failed\n");
		goto error;
	}

	ret=CLIENT_OK; goto end;
error:
	ret=CLIENT_ERROR; goto end;
reconnect:
	ret=CLIENT_RECONNECT; goto end;
end:
	if(server_version) free(server_version);
	return ret;
}

int listening_socket( const char *port ) {
    if ( not port )
        return -1;

    // stolen from https://banu.com/blog/2/how-to-use-epoll-a-complete-example-in-c/
    addrinfo hints;
    memset( &hints, 0, sizeof( addrinfo ) );
    hints.ai_family   = AF_UNSPEC;   /* Return IPv4 and IPv6 choices */
    hints.ai_socktype = SOCK_STREAM; /* We want a TCP socket */
    hints.ai_flags    = AI_PASSIVE;  /* All interfaces */

    addrinfo *result;
    if ( int s = getaddrinfo( NULL, port, &hints, &result ) ) {
        fprintf( stderr, "getaddrinfo: %s\n", gai_strerror( s ) );
        return -1;
    }

    for ( addrinfo *rp = result; rp; rp = rp->ai_next ) {
        // socket ok ?
        int listen_fd = socket( rp->ai_family, rp->ai_socktype, rp->ai_protocol );
        if ( listen_fd < 0 )
            continue;

        // no "Address already in use" error message
        int yes = 1;
        if ( setsockopt( listen_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof( yes ) ) == -1 )
            perror( "setsockopt" );

        // TCP_CORK
        //if ( setsockopt( listen_fd, SOL_TCP, TCP_CORK, &yes, sizeof( yes ) ) == -1 )
        //    perror( "setsockopt" );

        // bind ok ?
        if ( bind( listen_fd, rp->ai_addr, rp->ai_addrlen ) == 0 ) {
            freeaddrinfo( result );

            if ( set_non_blocking( listen_fd ) == -1 ) {
                close( listen_fd );
                return -1;
            }

            if ( listen( listen_fd, SOMAXCONN ) == -1 ) {
                perror( "listen" );
                close( listen_fd );
                return -1;
            }

            return listen_fd;
        }

        close( listen_fd );
    }

    fprintf( stderr, "Could not bind\n" );
    freeaddrinfo( result );
    return -1;
}

void addfd(int epollfd, int fd)
{
	struct epoll_event _event;
	_event.data.fd = fd;
	_event.events = EPOLLIN | EPOLLET;
	epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &_event);
	set_non_blocking(fd);
	log_info("%d add to epoll fd %d\n", fd, epollfd);
}

/*
 * It is possible for a race condition to exist; such that select()
 * indicates that a listen()ing socket is able to recieve a new connection
 * and that a later accept() call will still block because the connection
 * has been closed in the interim.  This wrapper for accept() attempts to
 * defeat this by making the accept() call nonblocking.
 */
int	Accept (int s, SA *addr, int *addrlen)
{
	int	retval;

	set_non_blocking(s);
	retval = accept(s, addr, addrlen);
	set_blocking(s);
	return retval;
}

void TCPServer::connect(const SocketAddress& socket_address)
{
    if (!socket_address.is_server_valid())
        throw NetworkException(NetworkExceptionT::InvalidSocketAddress);

    struct addrinfo hints;
    std::memset(&hints, 0, sizeof(hints));
    hints.ai_family = AF_UNSPEC; // Both IPV6 and IPV4
    hints.ai_socktype = SOCK_STREAM; // TCP
    hints.ai_flags = AI_PASSIVE; // We want to bind on

    struct addrinfo* addrs = nullptr;

    //Create a lambda deleter
    auto deleter = [](struct addrinfo* ptr)
        { if (ptr != nullptr) freeaddrinfo(ptr); };

    //Will store addresses to allow RAII.
    std::unique_ptr<struct addrinfo, void(*)(struct addrinfo*)>
        resources_keeper(nullptr, deleter);

    retrieve_addresses(socket_address.m_name, socket_address.m_port,
                       hints, addrs,
                       resources_keeper, deleter);

    //Socket FileDescriptor
    FileDescriptor fd;
    struct addrinfo *addr = nullptr;
    for (addr = addrs; addr != nullptr; addr = addr->ai_next)
    {
        fd = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
        if (fd == -1)
            continue;

        int errcode = bind(fd, addr->ai_addr, addr->ai_addrlen);

        //We binded on this socket
        if (errcode == 0)
            break;

        //Failed, let's try again
        close (fd);
    }

    if (addr == nullptr)
        throw NetworkException(NetworkExceptionT::BindFailed);

    if (!set_non_blocking(fd))
        throw NetworkException(NetworkExceptionT::CantSetNonblocking);

    if (listen(fd, MAX_CONNECTIONS) < 0)
        throw NetworkException(NetworkExceptionT::ListenFailed);

    m_connected = true;
    m_fd = fd;
}

static int accept_cb(tcp_server_stream_t *listen_stream)
{
    struct sockaddr_in remote_addr;
    struct epoll_event event;
    socklen_t          addr_len;
    int                sock_fd;
    tcp_server_t      *server;
    tcp_server_stream_t      *stream;

    bzero(&remote_addr, sizeof(remote_addr));
    addr_len = sizeof(struct sockaddr_in); 

    sock_fd = accept(listen_stream->sock_fd, (struct sockaddr *)&remote_addr, &addr_len);
    if (sock_fd < 0) {
        if (EAGAIN != errno) {
            log_error("accept error, %s.", strerror(errno)); 
            return 0;
        }
        else {
            log_info("no connection for accept."); 
            return 1;
        }
    }

    log_debug("new connection comes, fd: %d.", sock_fd);
    if (!set_non_blocking(sock_fd)) {
        log_error("set sock_fd: %d failed.", sock_fd);
        close(sock_fd);
        return 0;
    }

    stream = get_idle_tcp_stream(server);
    if (NULL == stream) {
        log_error("no idle stream.");
        return 0;
    }

    tcp_stream_init(server, stream, sock_fd, &remote_addr); 
    add_stream_into_busy_list(server, stream);

    /* 将当前sock fd添加到epoll中 */
    bzero(&event, sizeof(event));
    event.data.fd  = sock_fd;
    event.data.ptr = stream;
    event.events   = EPOLLIN | EPOLLET;

    if (epoll_ctl(server->epoll_fd, EPOLL_CTL_ADD, sock_fd, &event) < 0) {
        log_error("accept_cb, EPOLL_CTL_ADD error, %s.", strerror(errno)); 
        tcp_stream_close(stream);
        return 0;
    }
    server->active_stream_num++;

    return 1;
}

int pkr_conn_accepted_cb(int sockfd, void* void_pkm)
{
  struct incoming_conn_state* ics;
  struct pk_manager* pkm = (struct pk_manager*) void_pkm;
  struct pk_backend_conn* pkb;
  socklen_t slen;
  char rbuf[128];
  char lbuf[128];

  PK_TRACE_FUNCTION;

  ics = malloc(sizeof(struct incoming_conn_state));
  ics->pkm = pkm;
  ics->pkb = NULL;
  ics->tunnel = NULL;
  ics->hostname = NULL;
  ics->parsed_as = PROTO_UNKNOWN;
  ics->parse_state = PARSE_UNDECIDED;
  ics->created = pk_time();
  ics->unparsed_data = NULL;

  slen = sizeof(ics->local_addr);
  getsockname(sockfd, (struct sockaddr*) &(ics->local_addr), &slen);

  slen = sizeof(ics->remote_addr);
  getpeername(sockfd, (struct sockaddr*) &(ics->remote_addr), &slen);

  pk_log(PK_LOG_TUNNEL_DATA,
         "Accepted; remote=%s; local=%s; fd=%d",
         in_addr_to_str((struct sockaddr*) &(ics->remote_addr), rbuf, 128),
         in_addr_to_str((struct sockaddr*) &(ics->local_addr), lbuf, 128),
         sockfd);

  /* Allocate a connection for this request or die... */
  sprintf(lbuf, "!NEW:%d", ics->remote_addr.sin_port);
  if (NULL == (pkb = pkm_alloc_be_conn(pkm, NULL, lbuf))) {
    _pkr_close(ics, PK_LOG_ERROR, "BE alloc failed");
    PKS_close(sockfd);
    return -1;
  }

  pkb->kite = NULL;
  pkb->conn.sockfd = sockfd;
  ics->pkb = pkb;

  set_non_blocking(sockfd);

  ev_io_init(&(pkb->conn.watch_r),
             pkr_new_conn_readable_cb,
             PKS_EV_FD(sockfd),
             EV_READ);
  pkb->conn.watch_r.data = (void *) ics;
  ev_io_start(pkm->loop, &(pkb->conn.watch_r));
  return 0;
}