//CALLBACK ON CLIENT ACCEPT
static void accept_cb(struct ev_loop *loop, struct ev_io *server, int revents)
{
    int client_socket = accept(server->fd, 0, 0);
    set_nonblock(client_socket);
    clients.push_front(client_socket);

    struct ev_io *watcher = new my_io;
    ev_init(watcher, read_cb);
    ev_io_set(watcher, client_socket, EV_READ);
    ev_io_start(loop, watcher);

    std::cout << "accepted connection\n";
}

int main(int argc, const char *argv[])
{

    set_nonblock(STDIN_FILENO);

    char buf[100];
    int ret = read(STDIN_FILENO, buf, 100);
    if(ret == -1)
        ERR_EXIT("read");

    
    return 0;
}

int 
swill_sock_set_nonblock(int sock)
{
   int old = -1;

   old = set_nonblock(sock);
#if defined(GONZO_DEBUG) && (GONZO_DEBUG > 0)
   fprintf(stderr, "GONZO: Set nonblock on socket %d, old value was %d\n",
	   sock, old);
#endif

   return old;
}  

void network_t::update() {
    int num = epoll_wait(epfd, events, max_conn, -1);
	
    #pragma omp parallel for
    for (int i=0; i<num; ++i) {
        epoll_event* ev = events+i;
        mydata_t* md = (mydata_t*)ev->data.ptr;

        if (md->fd == acceptor) {
       	    sockaddr_in client_addr;
	    socklen_t sinsize = sizeof(client_addr);

	    int newfd = accept(acceptor, (sockaddr*)&client_addr, &sinsize);
	    while (newfd >= 0) {
		set_nonblock(newfd);
		set_linger(newfd, 0);
		set_nodelay(newfd);
		auto md = deal_event(epfd, EPOLL_CTL_ADD, newfd, EPOLLIN|EPOLLET, new mydata_t);
		
		logon(md);

	        newfd = accept(acceptor, (sockaddr*)&client_addr, &sinsize);
	    }

	    if (errno != EWOULDBLOCK && errno != EAGAIN) {
	       perror("accept newfd");
	       continue;
	    }
        } else {
	    if (ev->events & (EPOLLERR | EPOLLHUP)) {
	        md->close();
	    } else {
	        if (ev->events & EPOLLIN) {
	            md->deal_read();
		}
		if (ev->events & EPOLLOUT) {
		    md->deal_write();
		}
            }

	    if (md->closed.load()) {
		logoff(md);

		shutdown(md->fd, SHUT_RDWR);
		close(md->fd);
		delete md;
	    }
	}
    }
	
}

int
evcom_server_listen (evcom_server *server, struct sockaddr *address, int backlog)
{
  assert(!LISTENING(server));

  int fd = socket(address->sa_family, SOCK_STREAM, 0);
  if (fd < 0) {
    server->errorno = errno;
    evcom_perror("socket()", errno);
    return -1;
  }

  server->fd = fd;
  ev_io_set(&server->watcher, server->fd, EV_READ);

  if (set_nonblock(fd) != 0) {
    server->errorno = errno;
    evcom_perror("set_nonblock()", errno);
    close(fd);
    return -1;
  }

  int flags = 1;
  setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags));
  setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags));

  /* XXX: Sending single byte chunks in a response body? Perhaps there is a
   * need to enable the Nagel algorithm dynamically. For now disabling.
   */
  //setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags));

  if (bind(fd, address, address_length(address)) < 0) {
    server->errorno = errno;
    evcom_perror("bind()", errno);
    close(fd);
    return -1;
  }

  if (listen(fd, backlog) < 0) {
    server->errorno = errno;
    evcom_perror("listen()", errno);
    close(fd);
    return -1;
  }

  server->flags |= EVCOM_LISTENING;
  server->action = accept_connections;

  return 0;
}

int wait_for_connection(int master_sock) {
    static socklen_t len = sizeof(struct sockaddr);
    struct sockaddr_in peer;

    int newsock = accept(master_sock, (struct sockaddr*)&peer, &len);
    if (newsock < 0) {
        if (errno != EINTR) {
            perror("accept");
        }
    }

    set_nonblock(newsock);
    return newsock;
}

static void test_fd (flux_reactor_t *reactor)
{
    int fd[2];
    flux_watcher_t *r, *w;

    ok (socketpair (PF_LOCAL, SOCK_STREAM, 0, fd) == 0
        && set_nonblock (fd[0]) == 0 && set_nonblock (fd[1]) == 0,
        "fd: successfully created non-blocking socketpair");
    r = flux_fd_watcher_create (reactor, fd[0], FLUX_POLLIN, fdreader, NULL);
    w = flux_fd_watcher_create (reactor, fd[1], FLUX_POLLOUT, fdwriter, NULL);
    ok (r != NULL && w != NULL,
        "fd: reader and writer created");
    flux_watcher_start (r);
    flux_watcher_start (w);
    ok (flux_reactor_run (reactor, 0) == 0,
        "fd: reactor ran to completion after %lu bytes", fdwriter_bufsize);
    flux_watcher_stop (r);
    flux_watcher_stop (w);
    flux_watcher_destroy (r);
    flux_watcher_destroy (w);
    close (fd[0]);
    close (fd[1]);
}

int timed_connect(int fd,
    const struct sockaddr * addr,
    socklen_t addrlen, int timeout)
{
  struct pollfd pfd;
  int ret;

  if (set_nonblock(fd)==-1)
    return -1;

  do ret = connect(fd, addr, addrlen);
  while (ret==-1 && errno==EINTR);
  if (ret==0)
    return 0;

  if (ret==-1 && errno!=EINPROGRESS)
    return -1;

  pfd.fd = fd;
  pfd.events = POLLIN | POLLOUT;

  do ret = poll(&pfd, 1, timeout);
  while (ret==-1 && errno==EINTR);

  if (ret==-1)
  {
    return -1;
  }
  else if (ret==0)
  {
    errno = ETIMEDOUT;
    return -1;
  }
  else
  {
    int err;
    socklen_t error_len = sizeof(err);
    if (getsockopt(fd, SOL_SOCKET, SO_ERROR, (char *)&err, &error_len)==-1)
      return -1;

    if (err!=0)
    {
      errno = err;
      return -1;
    }

    return 0;
  }
}

int accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen)
{
    int ret;
    int connfd = 0;

    while ((connfd = g_sys_accept(sockfd, addr, addrlen)) < 0)
    {
        if (EINTR == errno)
            continue;

        if (!fd_not_ready())
            return -1;

        ret = add_fd_event(sockfd, EVENT_READABLE, event_conn_callback, current_coro());
        if (ret)
            return -2;

        schedule_timeout(ACCEPT_TIMEOUT);
        del_fd_event(sockfd, EVENT_READABLE);
        if (is_wakeup_by_timeout())
        {
            errno = ETIME;
            return -3;
        }
    }

    ret = set_nonblock(connfd);
    if (ret)
    {
        close(connfd);
        return -4;
    }

    ret = enable_tcp_no_delay(connfd);
    if (ret)
    {
        close(connfd);
        return -5;
    }

    ret = set_keep_alive(connfd, KEEP_ALIVE);
    if (ret)
    {
        close(connfd);
        return -6;
    }

    return connfd;
}

static struct coupling*
create_coupling (int fd, int is_pull)
{
    int pipefd[2];

    struct coupling *c = malloc(sizeof(struct coupling));
    if (!c) return NULL;

    int r = pipe(pipefd);
    if (r < 0) return NULL;

    r = set_nonblock(pipefd[0]);
    if (r < 0) return NULL;
    assert(pipefd[0] >= 0);

    r = set_nonblock(pipefd[1]);
    if (r < 0) return NULL;
    assert(pipefd[1] >= 0);

    if (is_pull) {
        c->is_pull = 1;
        c->pullfd = fd;
        c->pushfd = pipefd[1];
        c->exposedfd = pipefd[0];
    } else {
        c->is_pull = 0;
        c->pushfd = fd;
        c->pullfd = pipefd[0];
        c->exposedfd = pipefd[1];
    }

    r = pthread_create(&c->tid, NULL, pump_thread, c);
    if (r < 0) return NULL;

    return c;
}

int create_listening_socket(int port) {
  int listenfd, optval = 1, res;
  struct sockaddr_in servaddr;

  memset(&servaddr, 0, sizeof(servaddr));
  TRY_OR_EXIT(listenfd, socket(AF_INET, SOCK_STREAM, 0), "socket");
  setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
  servaddr.sin_family      = AF_INET;
  servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
  servaddr.sin_port        = htons(port);
  TRY_OR_EXIT(res, bind(listenfd, (struct sockaddr *) &servaddr, sizeof(servaddr)), "bind");
  TRY_OR_EXIT(res, listen(listenfd, 511), "listen");
  set_nonblock(listenfd);
  return listenfd;
}

/** Bind a server on the given port returning the created socket descriptor. */
int server_bind(struct server *server, char const *port)
{
	struct addrinfo hints;
	memset(&hints, 0, sizeof hints);
	hints.ai_family = AF_UNSPEC;
	hints.ai_socktype = SOCK_STREAM;
	hints.ai_flags = AI_PASSIVE; // use my IP

	struct addrinfo *servinfo;
	int rv;
	if ((rv = getaddrinfo(NULL, port, &hints, &servinfo)) != 0) {
		err(errno, "getaddrinfo: %s", gai_strerror(rv));
	}

	// loop through all the results and bind to the first we can
	int yes = 1;
	struct addrinfo *p;
	for (p = servinfo; p != NULL; p = p->ai_next) {
		if ((server->fd = socket(p->ai_family, p->ai_socktype,
		                         p->ai_protocol)) == -1) {
			perror("server: socket");
			continue;
		}
		set_nonblock(server->fd);

		if (setsockopt(server->fd, SOL_SOCKET, SO_REUSEADDR, &yes,
		               sizeof(int)) == -1) {
			perror("setsockopt");
			exit(1);
		}

		if (bind(server->fd, p->ai_addr, p->ai_addrlen) == -1) {
			close(server->fd);
			perror("server: bind");
			continue;
		}

		break;
	}

	if (p == NULL) {
		perror("server: failed to bind");
		return -2;
	}

	freeaddrinfo(servinfo); // all done with this structure
	return server->fd;
}

int vnode_listen(const char *name)
{
  int fd;
  struct sockaddr_un addr;

#ifdef DEBUG
  WARNX("opening '%s'", name);
#endif

  if (strlen(name) > sizeof(addr.sun_path) - 1)
  {
    WARNX("name too long: '%s'", name);
    return -1;
  }

  if ((fd = socket(AF_UNIX, SOCK_SEQPACKET, 0)) < 0)
  {
    WARN("socket() failed");
    return -1;
  }

  unlink(name);
  addr.sun_family = AF_UNIX;
  strcpy(addr.sun_path, name);

  if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0)
  {
    WARN("bind() failed for '%s'", name);
    close(fd);
    return -1;
  }

  /* to override umask */
  if (chmod(name, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH))
    WARN("fchmod() failed for '%s'", name);

  if (listen(fd, 5) < 0)
  {
    WARN("listen() failed");
    close(fd);
    return -1;
  }

  if (set_nonblock(fd))
    WARN("set_nonblock() failed for fd %d", fd);

  return fd;
}

/* Retruns evcom_stream if a connection could be accepted.
 * The returned stream is not yet attached to the event loop.
 * Otherwise NULL
 */
static evcom_stream*
accept_connection (evcom_server *server)
{
  struct sockaddr address; /* connector's address information */
  socklen_t addr_len = sizeof(address);

  int fd = accept(server->fd, &address, &addr_len);
  if (fd < 0) {
    switch (errno) {
      case EMFILE:
      case ENFILE:
        too_many_connections = 1;
        server->flags |= EVCOM_TOO_MANY_CONN;
        evcom_server_detach(server);
        return NULL;

      case EINTR:
      case EAGAIN:
        return NULL;

      default:
        evcom_perror("accept()", errno);
        return NULL;
    }
    assert(0 && "no reach");
  }

  evcom_stream *stream = NULL;

  if (server->on_connection) {
    stream = server->on_connection(server, &address);
  }

  if (stream == NULL) {
    close(fd);
    return NULL;
  }

  if (set_nonblock(fd) != 0) {
    evcom_perror("set_nonblock()", errno);
    return NULL;
  }

  stream->server = server;
  evcom_stream_assign_fds(stream, fd, fd);

  return stream;
}

int tcp_connect(const char *ip, unsigned short port,
		const char *bind_ip)
{
	struct sockaddr_in addr;
	int fd, tried = 5;
	if ((fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == -1) {
		perror("socket");
		goto out;
	}

again:
	if (bind_ip && tcp_bind(fd, bind_ip, 0) != 0) {
		if (-- tried > 0) {
			goto again;
		}
		goto err;
	}

	if (set_nonblock(fd) < 0) {
		perror("set_nonblock");
		goto err;
	}

	if (set_socket_linger(fd) != 0) {
		perror("setsockopt");
		return -1;
	}

	memset(&addr, '\0', sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	if (inet_pton(AF_INET, ip, &addr.sin_addr) <= 0) {
		perror("inet_pton");
		goto err;
	}

	if (connect(fd, (struct sockaddr *) &addr, sizeof(addr)) == -1) {
		if (errno != EINPROGRESS) {
			//fprintf(stderr, "errno = %d, str = %s\n", errno, strerror(errno));
			goto err;
		}
	}
	return fd;
err:
	close(fd);
out:
	return -1;
}

int CFtp::AVFtpConnect()
{
	struct sockaddr_in sin;    
    int on = 1;
    int iRetCode = -1;
    const int optval = 1;
	struct hostent* host;
	char recv_buf[1500];
    int state = -1;

	m_Ctrlsocket = socket(AF_INET, SOCK_STREAM, 0) ;
	FTP_ASSERT_RETURN( (m_Ctrlsocket > 0), FTP_ERROR);

	iRetCode = setsockopt(m_Ctrlsocket,SOL_SOCKET,SO_REUSEADDR,(const char *)&on,sizeof(on));
    FTP_ASSERT_RETURN( (iRetCode >= 0), FTP_ERROR);

	if (setsockopt(m_Ctrlsocket, IPPROTO_TCP, TCP_NODELAY,(char *)&optval, sizeof(optval)) < 0)
    {
        printf("error setsockopt nodelay");
        FTP_ASSERT_RETURN( 0, FTP_ERROR );
    }

	memset(&sin,0,sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_port = htons(m_ServerPort);

	host = gethostbyname( (char*)m_ServerIP.c_str() );
	if(host == NULL)
	{
		return -1;
	}
	sin.sin_addr.s_addr = *(unsigned long *)host->h_addr_list[0];

	iRetCode = connect(m_Ctrlsocket,(struct sockaddr *)(&sin),sizeof(sin));
	if (iRetCode < 0)
	{
		printf("FTP:Connet[%s][%d] Fail\n",m_ServerIP.c_str(),m_ServerPort);
		FTP_close(m_Ctrlsocket);
		return FTP_CONNECT_FAILED;
	}

	set_nonblock(TRUE, m_Ctrlsocket);
	iRetCode = AVFtpSendAndGet("", "", recv_buf, state);
    FTP_ASSERT_RETURN( (iRetCode >= FTP_OK), iRetCode);
    FTP_ASSERT_RETURN( (state == 220), FTP_ERROR);
	
    return FTP_OK;
}

static client_t * client_create (ctx_t *ctx, int fd)
{
    client_t *c;
    socklen_t crlen = sizeof (c->ucred);
    flux_t h = ctx->h;

    c = xzmalloc (sizeof (*c));
    c->fd = fd;
    if (!(c->uuid = zuuid_new ()))
        oom ();
    c->ctx = ctx;
    if (!(c->disconnect_notify = zhash_new ()))
        oom ();
    if (!(c->subscriptions = zlist_new ()))
        oom ();
    if (!(c->outqueue = zlist_new ()))
        oom ();
    if (getsockopt (fd, SOL_SOCKET, SO_PEERCRED, &c->ucred, &crlen) < 0) {
        flux_log (h, LOG_ERR, "getsockopt SO_PEERCRED: %s", strerror (errno));
        goto error;
    }
    assert (crlen == sizeof (c->ucred));
    /* Deny connections by uid other than session owner for now.
     */
    if (c->ucred.uid != ctx->session_owner) {
        flux_log (h, LOG_ERR, "connect by uid=%d pid=%d denied",
                  c->ucred.uid, (int)c->ucred.pid);
        goto error;
    }
    c->inw = flux_fd_watcher_create (fd, FLUX_POLLIN, client_read_cb, c);
    c->outw = flux_fd_watcher_create (fd, FLUX_POLLOUT, client_write_cb, c);
    if (!c->inw || !c->outw) {
        flux_log (h, LOG_ERR, "flux_fd_watcher_create: %s", strerror (errno));
        goto error;
    }
    flux_fd_watcher_start (h, c->inw);
    flux_msg_iobuf_init (&c->inbuf);
    flux_msg_iobuf_init (&c->outbuf);
    if (set_nonblock (c->fd, true) < 0) {
        flux_log (h, LOG_ERR, "set_nonblock: %s", strerror (errno));
        goto error;
    }

    return (c);
error:
    client_destroy (c);
    return NULL;
}

void Connection::add_new_user(){
	int Slave = accept(Master, 0,0);
	if (Slave == -1) {
		throw std::system_error(errno, std::system_category());
	}
	struct epoll_event event;
	event.data.fd = Slave;
	event.events = EPOLLIN | EPOLLET;
	epoll_ctl(epfd, EPOLL_CTL_ADD, Slave, &event);
	fds.insert(Slave);
	set_nonblock(Slave);
	msg[Slave] = std::string("");
	send(Slave, welcome, 24, MSG_NOSIGNAL);
	log.say("accepted connection\n");

}

static void 
listen_sock(int mc_sock, addr_t *mc_addr) {
	{
		// set sock option
		int flag = 1;
		setsockopt(mc_sock, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag));
		long fl = fcntl(mc_sock, F_GETFL);
		if (fl == -1) perror_fatal("can't get fcntl sock option");

		set_nonblock(mc_sock,1);
	}
	
	
	if ( bind(mc_sock, &mc_addr->name, mc_addr->namelen) < 0) perror_fatal("bind");	
	if (listen(mc_sock, 10) < 0) perror_fatal("publisher listen");	
}  

static int sock_init(edpnet_sock_t sock){
    struct edpnet_sock	*s = sock;
    
    if(set_nonblock(s->es_sock) < 0){
	close(s->es_sock);
	return -1;
    }

    INIT_LIST_HEAD(&s->es_node);
    INIT_LIST_HEAD(&s->es_iowrites);

    spi_spin_init(&s->es_lock);
    s->es_status |= kEDPNET_SOCK_STATUS_INIT;

    return 0;
}