process_pub_con(agooPub pub, agooConLoop loop) {
    agooUpgraded	up = pub->up;

    if (NULL != up && NULL != up->con && up->con->loop == loop) {
	int	pending;

	// TBD Change pending to be based on length of con queue
	if (1 == (pending = atomic_fetch_sub(&up->pending, 1))) {
	    if (NULL != up && agoo_server.ctx_nil_value != up->ctx && up->on_empty) {
		agooReq	req = agoo_req_create(0);

		req->up = up;
		req->method = AGOO_ON_EMPTY;
		req->hook = agoo_hook_create(AGOO_NONE, NULL, up->ctx, PUSH_HOOK, &agoo_server.eval_queue);
		agoo_upgraded_ref(up);
		agoo_queue_push(&agoo_server.eval_queue, (void*)req);
	    }
	}
    }
    switch (pub->kind) {
    case AGOO_PUB_CLOSE:
	// A close after already closed is used to decrement the reference
	// count on the upgraded so it can be destroyed in the con loop
	// threads.
	if (NULL != up->con && up->con->loop == loop) {
	    agooRes	res = agoo_res_create(up->con);

	    if (NULL != res) {
		agoo_con_res_append(up->con, res);
		res->con_kind = up->con->bind->kind;
		res->close = true;
	    }
	}
	break;
    case AGOO_PUB_WRITE: {
	if (NULL == up->con) {
	    agoo_log_cat(&agoo_warn_cat, "Connection already closed. WebSocket write failed.");
	} else if (up->con->loop == loop) {
	    agooRes	res = agoo_res_create(up->con);

	    if (NULL != res) {
		agoo_con_res_append(up->con, res);
		res->con_kind = AGOO_CON_ANY;
		agoo_res_message_push(res, pub->msg);
	    }
	}
	break;
    case AGOO_PUB_SUB:
	if (NULL != up && up->con->loop == loop) {
	    agoo_upgraded_add_subject(pub->up, pub->subject);
	    pub->subject = NULL;
	}
	break;
    case AGOO_PUB_UN:
	if (NULL != up && up->con->loop == loop) {
	    unsubscribe_pub(pub);
	}
	break;
    case AGOO_PUB_MSG:
	publish_pub(pub, loop);
	break;
    }
    default:
	break;
    }
    agoo_pub_destroy(pub);
}
process_pub_con(agooPub pub, agooConLoop loop) {
    agooUpgraded	up = pub->up;

    if (NULL != up && NULL != up->con && up->con->loop == loop) {
	int	pending;

	 
	if (1 == (pending = atomic_fetch_sub(&up->pending, 1))) {
	    if (NULL != up && agoo_server.ctx_nil_value != up->ctx && up->on_empty) {
		agooReq	req = agoo_req_create(0);

		req->up = up;
		req->method = AGOO_ON_EMPTY;
		req->hook = agoo_hook_create(AGOO_NONE, NULL, up->ctx, PUSH_HOOK, &agoo_server.eval_queue);
		agoo_upgraded_ref(up);
		agoo_queue_push(&agoo_server.eval_queue, (void*)req);
	    }
	}
    }
    switch (pub->kind) {
    case AGOO_PUB_CLOSE:
	 
	 
	 
	if (NULL != up->con && up->con->loop == loop) {
	    agooRes	res = agoo_res_create(up->con);

	    if (NULL != res) {
		agoo_con_res_append(up->con, res);
		res->con_kind = up->con->bind->kind;
		res->close = true;
	    }
	}
	break;
    case AGOO_PUB_WRITE: {
	if (NULL == up->con) {
	    agoo_log_cat(&agoo_warn_cat, "Connection already closed. WebSocket write failed.");
	} else if (up->con->loop == loop) {
	    agooRes	res = agoo_res_create(up->con);

	    if (NULL != res) {
		agoo_con_res_append(up->con, res);
		res->con_kind = AGOO_CON_ANY;
		agoo_res_message_push(res, pub->msg);
	    }
	}
	break;
    case AGOO_PUB_SUB:
	if (NULL != up && up->con->loop == loop) {
	    agoo_upgraded_add_subject(pub->up, pub->subject);
	    pub->subject = NULL;
	}
	break;
    case AGOO_PUB_UN:
	if (NULL != up && up->con->loop == loop) {
	    unsubscribe_pub(pub);
	}
	break;
    case AGOO_PUB_MSG:
	publish_pub(pub, loop);
	break;
    }
    default:
	break;
    }
    agoo_pub_destroy(pub);
}