agoo_con_loop(void *x) {
    agooConLoop		loop = (agooConLoop)x;
    struct _agooErr	err = AGOO_ERR_INIT;
    agooReady		ready = agoo_ready_create(&err);
    agooPub		pub;
    agooCon		c;
    int			con_queue_fd = agoo_queue_listen(&agoo_server.con_queue);
    int			pub_queue_fd = agoo_queue_listen(&loop->pub_queue);

    if (NULL == ready) {
	agoo_log_cat(&agoo_error_cat, "Failed to create connection manager. %s", err.msg);
	exit(EXIT_FAILURE);
	return NULL;
    }
    if (AGOO_ERR_OK != agoo_ready_add(&err, ready, con_queue_fd, &con_queue_handler, loop) ||
	AGOO_ERR_OK != agoo_ready_add(&err, ready, pub_queue_fd, &pub_queue_handler, loop)) {
	agoo_log_cat(&agoo_error_cat, "Failed to add queue connection to manager. %s", err.msg);
	exit(EXIT_FAILURE);

	return NULL;
    }
    atomic_fetch_add(&agoo_server.running, 1);

    while (agoo_server.active) {
	while (NULL != (c = (agooCon)agoo_queue_pop(&agoo_server.con_queue, 0.0))) {
	    c->loop = loop;
	    if (AGOO_ERR_OK != agoo_ready_add(&err, ready, c->sock, &con_handler, c)) {
		agoo_log_cat(&agoo_error_cat, "Failed to add connection to manager. %s", err.msg);
		agoo_err_clear(&err);
	    }
	    if (AGOO_CON_HTTPS == c->bind->kind) {
		con_ssl_setup(c);
	    }
	}
	while (NULL != (pub = (agooPub)agoo_queue_pop(&loop->pub_queue, 0.0))) {
	    process_pub_con(pub, loop);
	}
	if (AGOO_ERR_OK != agoo_ready_go(&err, ready)) {
	    agoo_log_cat(&agoo_error_cat, "IO error. %s", err.msg);
	    agoo_err_clear(&err);
	}
    }
    agoo_ready_destroy(ready);
    atomic_fetch_sub(&agoo_server.running, 1);

    return NULL;
}
agoo_con_loop(void *x) {
    agooConLoop		loop = (agooConLoop)x;
    struct _agooErr	err = AGOO_ERR_INIT;
    agooReady		ready = agoo_ready_create(&err);
    agooPub		pub;
    agooCon		c;
    int			con_queue_fd = agoo_queue_listen(&agoo_server.con_queue);
    int			pub_queue_fd = agoo_queue_listen(&loop->pub_queue);

    if (NULL == ready) {
	agoo_log_cat(&agoo_error_cat, "Failed to create connection manager. %s", err.msg);
	exit(EXIT_FAILURE);
	return NULL;
    }
    if (AGOO_ERR_OK != agoo_ready_add(&err, ready, con_queue_fd, &con_queue_handler, loop) ||
	AGOO_ERR_OK != agoo_ready_add(&err, ready, pub_queue_fd, &pub_queue_handler, loop)) {
	agoo_log_cat(&agoo_error_cat, "Failed to add queue connection to manager. %s", err.msg);
	exit(EXIT_FAILURE);

	return NULL;
    }
    atomic_fetch_add(&agoo_server.running, 1);

    while (agoo_server.active) {
	while (NULL != (c = (agooCon)agoo_queue_pop(&agoo_server.con_queue, 0.0))) {
	    c->loop = loop;
	    if (AGOO_ERR_OK != agoo_ready_add(&err, ready, c->sock, &con_handler, c)) {
		agoo_log_cat(&agoo_error_cat, "Failed to add connection to manager. %s", err.msg);
		agoo_err_clear(&err);
	    }
	    if (AGOO_CON_HTTPS == c->bind->kind) {
		con_ssl_setup(c);
	    }
	}
	while (NULL != (pub = (agooPub)agoo_queue_pop(&loop->pub_queue, 0.0))) {
	    process_pub_con(pub, loop);
	}
	if (AGOO_ERR_OK != agoo_ready_go(&err, ready)) {
	    agoo_log_cat(&agoo_error_cat, "IO error. %s", err.msg);
	    agoo_err_clear(&err);
	}
    }
    agoo_ready_destroy(ready);
    atomic_fetch_sub(&agoo_server.running, 1);

    return NULL;
}