void poll_events(pj_stun_config *stun_cfg, unsigned msec,
		 pj_bool_t first_event_only)
{
    pj_time_val stop_time;
    int count = 0;

    pj_gettimeofday(&stop_time);
    stop_time.msec += msec;
    pj_time_val_normalize(&stop_time);

    /* Process all events for the specified duration. */
    for (;;) {
	pj_time_val timeout = {0, 1}, now;
	int c;

	c = pj_timer_heap_poll( stun_cfg->timer_heap, NULL );
	if (c > 0)
	    count += c;

	//timeout.sec = timeout.msec = 0;
	c = pj_ioqueue_poll( stun_cfg->ioqueue, &timeout);
	if (c > 0)
	    count += c;

	pj_gettimeofday(&now);
	if (PJ_TIME_VAL_GTE(now, stop_time))
	    break;

	if (first_event_only && count >= 0)
	    break;
    }
}

    //
    // Handle events.
    //
    int handle_events(Pj_Time_Val *max_timeout)
    {
        Pj_Time_Val timeout(0, 0);
        int timer_count;

        timer_count = pj_timer_heap_poll( th_, &timeout );

        if (timeout.get_sec() < 0) 
            timeout.sec = PJ_MAXINT32;

        /* If caller specifies maximum time to wait, then compare the value 
         * with the timeout to wait from timer, and use the minimum value.
         */
        if (max_timeout && timeout >= *max_timeout) {
	    timeout = *max_timeout;
        }

        /* Poll events in ioqueue. */
        int ioqueue_count;

        ioqueue_count = pj_ioqueue_poll(ioq_, &timeout);
        if (ioqueue_count < 0)
	    return ioqueue_count;

        return ioqueue_count + timer_count;
    }

/*
 * udp_destroy()
 *
 * This function is called by transport manager (by transport->destroy()).
 */
static pj_status_t udp_destroy( pjsip_transport *transport )
{
    struct udp_transport *tp = (struct udp_transport*)transport;
    int i;

    /* Mark this transport as closing. */
    tp->is_closing = 1;

    /* Cancel all pending operations. */
    /* blp: NO NO NO...
     *      No need to post queued completion as we poll the ioqueue until
     *      we've got events anyway. Posting completion will only cause
     *      callback to be called twice with IOCP: one for the post completion
     *      and another one for closing the socket.
     *
    for (i=0; i<tp->rdata_cnt; ++i) {
	pj_ioqueue_post_completion(tp->key, 
				   &tp->rdata[i]->tp_info.op_key.op_key, -1);
    }
    */

    /* Unregister from ioqueue. */
    if (tp->key) {
	pj_ioqueue_unregister(tp->key);
	tp->key = NULL;
    } else {
	/* Close socket. */
	if (tp->sock && tp->sock != PJ_INVALID_SOCKET) {
	    pj_sock_close(tp->sock);
	    tp->sock = PJ_INVALID_SOCKET;
	}
    }

    /* Must poll ioqueue because IOCP calls the callback when socket
     * is closed. We poll the ioqueue until all pending callbacks 
     * have been called.
     */
    for (i=0; i<50 && tp->is_closing < 1+tp->rdata_cnt; ++i) {
	int cnt;
	pj_time_val timeout = {0, 1};

	cnt = pj_ioqueue_poll(pjsip_endpt_get_ioqueue(transport->endpt), 
			      &timeout);
	if (cnt == 0)
	    break;
    }

    if (tp->grp_lock) {
	pj_grp_lock_t *grp_lock = tp->grp_lock;
	tp->grp_lock = NULL;
	pj_grp_lock_dec_ref(grp_lock);
	/* Transport may have been deleted at this point */
    } else {
	udp_on_destroy(tp);
    }

    return PJ_SUCCESS;
}

/**
 * Worker thread proc.
 */
static int PJ_THREAD_FUNC worker_proc(void *arg)
{
    pjmedia_endpt *endpt = (pjmedia_endpt*) arg;

    while (!endpt->quit_flag) {
	pj_time_val timeout = { 0, 500 };
	pj_ioqueue_poll(endpt->ioqueue, &timeout);
    }

    return 0;
}

static void handle_events(pj_stun_config *cfg, unsigned msec_delay)
{
    pj_time_val delay;

    pj_timer_heap_poll(cfg->timer_heap, NULL);

    delay.sec = 0;
    delay.msec = msec_delay;
    pj_time_val_normalize(&delay);

    pj_ioqueue_poll(cfg->ioqueue, &delay);
}

/* 
 * Handle timer and network events 
 */
static void srv_handle_events(pj_turn_srv *srv, const pj_time_val *max_timeout)
{
    /* timeout is 'out' var. This just to make compiler happy. */
    pj_time_val timeout = { 0, 0};
    unsigned net_event_count = 0;
    int c;

    /* Poll the timer. The timer heap has its own mutex for better 
     * granularity, so we don't need to lock the server. 
     */
    timeout.sec = timeout.msec = 0;
    c = pj_timer_heap_poll( srv->core.timer_heap, &timeout );

    /* timer_heap_poll should never ever returns negative value, or otherwise
     * ioqueue_poll() will block forever!
     */
    pj_assert(timeout.sec >= 0 && timeout.msec >= 0);
    if (timeout.msec >= 1000) timeout.msec = 999;

    /* If caller specifies maximum time to wait, then compare the value with
     * the timeout to wait from timer, and use the minimum value.
     */
    if (max_timeout && PJ_TIME_VAL_GT(timeout, *max_timeout)) {
	timeout = *max_timeout;
    }

    /* Poll ioqueue. 
     * Repeat polling the ioqueue while we have immediate events, because
     * timer heap may process more than one events, so if we only process
     * one network events at a time (such as when IOCP backend is used),
     * the ioqueue may have trouble keeping up with the request rate.
     *
     * For example, for each send() request, one network event will be
     *   reported by ioqueue for the send() completion. If we don't poll
     *   the ioqueue often enough, the send() completion will not be
     *   reported in timely manner.
     */
    do {
	c = pj_ioqueue_poll( srv->core.ioqueue, &timeout);
	if (c < 0) {
	    pj_thread_sleep(PJ_TIME_VAL_MSEC(timeout));
	    return;
	} else if (c == 0) {
	    break;
	} else {
	    net_event_count += c;
	    timeout.sec = timeout.msec = 0;
	}
    } while (c > 0 && net_event_count < MAX_NET_EVENTS);

}

static int worker_thread_proc(void *p)
{
    struct stun_test_session *test_sess = (struct stun_test_session*)p;

    PJ_LOG(4,(THIS_FILE, "Worker thread running"));

    while (!test_sess->thread_quit_flag) {
	pj_time_val timeout = {0, 10};
	pj_timer_heap_poll(test_sess->stun_cfg.timer_heap, NULL);
	pj_ioqueue_poll(test_sess->stun_cfg.ioqueue, &timeout);
    }

    PJ_LOG(4,(THIS_FILE, "Worker thread quitting"));
    return 0;
}

static int worker_thread(void *unused)
{
    PJ_UNUSED_ARG(unused);

    while (!g.quit) {
	const pj_time_val delay = {0, 10};

	/* Poll ioqueue for the TURN client */
	pj_ioqueue_poll(g.stun_config.ioqueue, &delay);

	/* Poll the timer heap */
	pj_timer_heap_poll(g.stun_config.timer_heap, NULL);

    }

    return 0;
}

pj_status_t getURL(const char *curl)
{
    pj_str_t url;
    pj_http_req_callback hcb;
    pj_status_t status;

    pj_bzero(&hcb, sizeof(hcb));
    hcb.on_complete = &on_complete;
    hcb.on_data_read = &on_data_read;
    hcb.on_send_data = &on_send_data;
    hcb.on_response = &on_response;

    /* Create pool, timer, and ioqueue */
    pool = pj_pool_create(mem, NULL, 8192, 4096, NULL);
    if (pj_timer_heap_create(pool, 16, &timer_heap))
        return -31;
    if (pj_ioqueue_create(pool, 16, &ioqueue))
        return -32;

    pj_strdup2(pool, &url, curl);

    if ((status = pj_http_req_create(pool, &url, timer_heap, ioqueue, 
                           NULL, &hcb, &http_req)) != PJ_SUCCESS)
        return status;

    if ((status = pj_http_req_start(http_req)) != PJ_SUCCESS)
        return status;

    while (pj_http_req_is_running(http_req)) {
        pj_time_val delay = {0, 50};
	pj_ioqueue_poll(ioqueue, &delay);
	pj_timer_heap_poll(timer_heap, NULL);
    }

    pj_http_req_destroy(http_req);
    pj_ioqueue_destroy(ioqueue);
    pj_timer_heap_destroy(timer_heap);
    pj_pool_release(pool);

    return PJ_SUCCESS;
}

/* The worker thread. */
static int worker_thread(void *p)
{
    struct thread_arg *arg = (struct thread_arg*) p;
    const pj_time_val timeout = {0, 100};
    int rc;

    while (!thread_quit_flag) {

	++arg->counter;
        rc = pj_ioqueue_poll(arg->ioqueue, &timeout);
	//TRACE_((THIS_FILE, "     thread: poll returned rc=%d", rc));
        if (rc < 0) {
	    char errmsg[PJ_ERR_MSG_SIZE];
	    pj_strerror(-rc, errmsg, sizeof(errmsg));
            PJ_LOG(3, (THIS_FILE, 
		       "...error in pj_ioqueue_poll() in thread %d "
		       "after %d loop: %s [pj_status_t=%d]", 
		       arg->id, arg->counter, errmsg, -rc));
            //return -1;
        }
    }
    return 0;
}

static int worker_thread(void *arg)
{
    pj_ioqueue_t *ioqueue = (pj_ioqueue_t*) arg;
    struct op_key read_op, write_op;
    char recv_buf[512], send_buf[512];
    pj_ssize_t length;
    pj_status_t rc;

    read_op.peer = &write_op;
    read_op.is_pending = 0;
    read_op.last_err = 0;
    read_op.buffer = recv_buf;
    read_op.size = sizeof(recv_buf);
    read_op.addrlen = sizeof(read_op.addr);

    write_op.peer = &read_op;
    write_op.is_pending = 0;
    write_op.last_err = 0;
    write_op.buffer = send_buf;
    write_op.size = sizeof(send_buf);

    length = sizeof(recv_buf);
    rc = pj_ioqueue_recvfrom(key, &read_op.op_key_, recv_buf, &length, 0,
                             &read_op.addr, &read_op.addrlen);
    if (rc == PJ_SUCCESS) {
        read_op.is_pending = 1;
        on_read_complete(key, &read_op.op_key_, length);
    }
    
    while (!thread_quit_flag) {
        pj_time_val timeout;
        timeout.sec = 0; timeout.msec = 10;
        rc = pj_ioqueue_poll(ioqueue, &timeout);
    }
    return 0;
}

//.........这里部分代码省略.........
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    status = pj_sock_listen(sock, PJ_SOMAXCONN);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    asock_cb.on_accept_complete = &asock_on_accept_complete;
    status = pj_activesock_create(pool, sock, pj_SOCK_STREAM(), NULL, 
				  ioqueue, &asock_cb, &state_serv, &asock_serv);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    status = pj_activesock_start_accept(asock_serv, pool);
    if (status != PJ_SUCCESS)
	goto on_return;

    /* Update listener address */
    {
	int addr_len;

	addr_len = sizeof(listen_addr);
	pj_sock_getsockname(sock, (pj_sockaddr_t*)&listen_addr, &addr_len);
    }

    /* CLIENT */
    pj_ssl_sock_param_default(&param);
    param.cb.on_connect_complete = &ssl_on_connect_complete;
    param.cb.on_data_read = &ssl_on_data_read;
    param.cb.on_data_sent = &ssl_on_data_sent;
    param.ioqueue = ioqueue;
    param.timer_heap = timer;
    param.timeout.sec = 0;
    param.timeout.msec = ms_timeout;
    pj_time_val_normalize(&param.timeout);
    param.user_data = &state_cli;

    state_cli.pool = pool;
    state_cli.is_server = PJ_FALSE;
    state_cli.is_verbose = PJ_TRUE;

    status = pj_ssl_sock_create(pool, &param, &ssock_cli);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    /* Init default bind address */
    {
	pj_str_t tmp_st;
	pj_sockaddr_init(PJ_AF_INET, &addr, pj_strset2(&tmp_st, "127.0.0.1"), 0);
    }

    status = pj_ssl_sock_start_connect(ssock_cli, pool, 
				       (pj_sockaddr_t*)&addr, 
				       (pj_sockaddr_t*)&listen_addr, 
				       pj_sockaddr_get_len(&listen_addr));
    if (status != PJ_EPENDING) {
	goto on_return;
    }

    /* Wait until everything has been sent/received or error */
    while ((!state_serv.err && !state_serv.done) || (!state_cli.err && !state_cli.done))
    {
#ifdef PJ_SYMBIAN
	pj_symbianos_poll(-1, 1000);
#else
	pj_time_val delay = {0, 100};
	pj_ioqueue_poll(ioqueue, &delay);
	pj_timer_heap_poll(timer, &delay);
#endif
    }

    if (state_serv.err || state_cli.err) {
	if (state_cli.err != PJ_SUCCESS)
	    status = state_cli.err;
	else
	    status = state_serv.err;

	goto on_return;
    }

    PJ_LOG(3, ("", "...Done!"));

on_return:
    if (asock_serv)
	pj_activesock_close(asock_serv);
    if (ssock_cli && !state_cli.err && !state_cli.done)
	pj_ssl_sock_close(ssock_cli);
    if (timer)
	pj_timer_heap_destroy(timer);
    if (ioqueue)
	pj_ioqueue_destroy(ioqueue);
    if (pool)
	pj_pool_release(pool);

    return status;
}

//.........这里部分代码省略.........
    param.timeout.msec = ms_timeout;
    pj_time_val_normalize(&param.timeout);

    /* SERVER */
    param.user_data = &state_serv;
    state_serv.pool = pool;
    state_serv.is_server = PJ_TRUE;
    state_serv.is_verbose = PJ_TRUE;

    status = pj_ssl_sock_create(pool, &param, &ssock_serv);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    status = pj_ssl_sock_set_certificate(ssock_serv, pool, cert);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    /* Init bind address */
    {
	pj_str_t tmp_st;
	pj_sockaddr_init(PJ_AF_INET, &listen_addr, pj_strset2(&tmp_st, "127.0.0.1"), 0);
    }

    status = pj_ssl_sock_start_accept(ssock_serv, pool, &listen_addr, pj_sockaddr_get_len(&listen_addr));
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    /* Update listener address */
    {
	pj_ssl_sock_info info;

	pj_ssl_sock_get_info(ssock_serv, &info);
	pj_sockaddr_cp(&listen_addr, &info.local_addr);
    }

    /* CLIENT */
    state_cli.pool = pool;
    status = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, &sock);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    asock_cb.on_connect_complete = &asock_on_connect_complete;
    asock_cb.on_data_read = &asock_on_data_read;
    status = pj_activesock_create(pool, sock, pj_SOCK_STREAM(), NULL, 
				  ioqueue, &asock_cb, &state_cli, &asock_cli);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    status = pj_activesock_start_connect(asock_cli, pool, (pj_sockaddr_t*)&listen_addr, 
					 pj_sockaddr_get_len(&listen_addr));
    if (status == PJ_SUCCESS) {
	asock_on_connect_complete(asock_cli, PJ_SUCCESS);
    } else if (status == PJ_EPENDING) {
	status = PJ_SUCCESS;
    } else {
	goto on_return;
    }

    /* Wait until everything has been sent/received or error */
    while (!state_serv.err && !state_cli.err && !state_serv.done && !state_cli.done)
    {
#ifdef PJ_SYMBIAN
	pj_symbianos_poll(-1, 1000);
#else
	pj_time_val delay = {0, 100};
	pj_ioqueue_poll(ioqueue, &delay);
	pj_timer_heap_poll(timer, &delay);
#endif
    }

    if (state_serv.err || state_cli.err) {
	if (state_serv.err != PJ_SUCCESS)
	    status = state_serv.err;
	else
	    status = state_cli.err;

	goto on_return;
    }

    PJ_LOG(3, ("", "...Done!"));

on_return:
    if (ssock_serv)
	pj_ssl_sock_close(ssock_serv);
    if (asock_cli && !state_cli.err && !state_cli.done)
	pj_activesock_close(asock_cli);
    if (timer)
	pj_timer_heap_destroy(timer);
    if (ioqueue)
	pj_ioqueue_destroy(ioqueue);
    if (pool)
	pj_pool_release(pool);

    return status;
}

//.........这里部分代码省略.........
    param.proto = cli_proto;
    param.user_data = &state_cli;
    param.ciphers_num = (cli_cipher == -1)? 0 : 1;
    ciphers[0] = cli_cipher;

    state_cli.pool = pool;
    state_cli.check_echo = PJ_TRUE;
    state_cli.is_verbose = PJ_TRUE;

    {
	pj_time_val now;

	pj_gettimeofday(&now);
	pj_srand((unsigned)now.sec);
	state_cli.send_str_len = (pj_rand() % 5 + 1) * 1024 + pj_rand() % 1024;
    }
    state_cli.send_str = pj_pool_alloc(pool, state_cli.send_str_len);
    {
	unsigned i;
	for (i = 0; i < state_cli.send_str_len; ++i)
	    state_cli.send_str[i] = (char)(pj_rand() % 256);
    }

    status = pj_ssl_sock_create(pool, &param, &ssock_cli);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    /* Set cert for client */
    {

	if (!client_provide_cert) {
	    pj_str_t tmp1, tmp2;

	    pj_strset2(&tmp1, (char*)CERT_CA_FILE);
	    pj_strset2(&tmp2, NULL);
	    status = pj_ssl_cert_load_from_files(pool, 
						 &tmp1, &tmp2, &tmp2, &tmp2,
						 &cert);
	    if (status != PJ_SUCCESS) {
		goto on_return;
	    }
	}

	status = pj_ssl_sock_set_certificate(ssock_cli, pool, cert);
	if (status != PJ_SUCCESS) {
	    goto on_return;
	}
    }

    status = pj_ssl_sock_start_connect(ssock_cli, pool, &addr, &listen_addr, pj_sockaddr_get_len(&addr));
    if (status == PJ_SUCCESS) {
	ssl_on_connect_complete(ssock_cli, PJ_SUCCESS);
    } else if (status == PJ_EPENDING) {
	status = PJ_SUCCESS;
    } else {
	goto on_return;
    }

    /* Wait until everything has been sent/received or error */
    while (!state_serv.err && !state_cli.err && !state_serv.done && !state_cli.done)
    {
#ifdef PJ_SYMBIAN
	pj_symbianos_poll(-1, 1000);
#else
	pj_time_val delay = {0, 100};
	pj_ioqueue_poll(ioqueue, &delay);
#endif
    }

    /* Clean up sockets */
    {
	pj_time_val delay = {0, 100};
	while (pj_ioqueue_poll(ioqueue, &delay) > 0);
    }

    if (state_serv.err || state_cli.err) {
	if (state_serv.err != PJ_SUCCESS)
	    status = state_serv.err;
	else
	    status = state_cli.err;

	goto on_return;
    }

    PJ_LOG(3, ("", "...Done!"));
    PJ_LOG(3, ("", ".....Sent/recv: %d/%d bytes", state_cli.sent, state_cli.recv));

on_return:
    if (ssock_serv)
	pj_ssl_sock_close(ssock_serv);
    if (ssock_cli && !state_cli.err && !state_cli.done) 
	pj_ssl_sock_close(ssock_cli);
    if (ioqueue)
	pj_ioqueue_destroy(ioqueue);
    if (pool)
	pj_pool_release(pool);

    return status;
}

static int https_client_test(unsigned ms_timeout)
{
    pj_pool_t *pool = NULL;
    pj_ioqueue_t *ioqueue = NULL;
    pj_timer_heap_t *timer = NULL;
    pj_ssl_sock_t *ssock = NULL;
    pj_ssl_sock_param param;
    pj_status_t status;
    struct test_state state = {0};
    pj_sockaddr local_addr, rem_addr;
    pj_str_t tmp_st;

    pool = pj_pool_create(mem, "https_get", 256, 256, NULL);

    status = pj_ioqueue_create(pool, 4, &ioqueue);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    status = pj_timer_heap_create(pool, 4, &timer);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    state.pool = pool;
    state.send_str = HTTP_REQ;
    state.send_str_len = pj_ansi_strlen(state.send_str);
    state.is_verbose = PJ_TRUE;

    pj_ssl_sock_param_default(&param);
    param.cb.on_connect_complete = &ssl_on_connect_complete;
    param.cb.on_data_read = &ssl_on_data_read;
    param.cb.on_data_sent = &ssl_on_data_sent;
    param.ioqueue = ioqueue;
    param.user_data = &state;
    param.server_name = pj_str((char*)HTTP_SERVER_ADDR);
    param.timer_heap = timer;
    param.timeout.sec = 0;
    param.timeout.msec = ms_timeout;
    param.proto = PJ_SSL_SOCK_PROTO_SSL23;
    pj_time_val_normalize(&param.timeout);

    status = pj_ssl_sock_create(pool, &param, &ssock);
    if (status != PJ_SUCCESS) {
	goto on_return;
    }

    pj_sockaddr_init(PJ_AF_INET, &local_addr, pj_strset2(&tmp_st, "0.0.0.0"), 0);
    pj_sockaddr_init(PJ_AF_INET, &rem_addr, pj_strset2(&tmp_st, HTTP_SERVER_ADDR), HTTP_SERVER_PORT);
    status = pj_ssl_sock_start_connect(ssock, pool, &local_addr, &rem_addr, sizeof(rem_addr));
    if (status == PJ_SUCCESS) {
	ssl_on_connect_complete(ssock, PJ_SUCCESS);
    } else if (status == PJ_EPENDING) {
	status = PJ_SUCCESS;
    } else {
	goto on_return;
    }

    /* Wait until everything has been sent/received */
    while (state.err == PJ_SUCCESS && !state.done) {
#ifdef PJ_SYMBIAN
	pj_symbianos_poll(-1, 1000);
#else
	pj_time_val delay = {0, 100};
	pj_ioqueue_poll(ioqueue, &delay);
	pj_timer_heap_poll(timer, &delay);
#endif
    }

    if (state.err) {
	status = state.err;
	goto on_return;
    }

    PJ_LOG(3, ("", "...Done!"));
    PJ_LOG(3, ("", ".....Sent/recv: %d/%d bytes", state.sent, state.recv));

on_return:
    if (ssock && !state.err && !state.done) 
	pj_ssl_sock_close(ssock);
    if (ioqueue)
	pj_ioqueue_destroy(ioqueue);
    if (timer)
	pj_timer_heap_destroy(timer);
    if (pool)
	pj_pool_release(pool);

    return status;
}

//.........这里部分代码省略.........
	}

	status = pj_ssl_sock_create(pool, &param, &ssock_cli[i]);
	if (status != PJ_SUCCESS) {
	    app_perror("...ERROR pj_ssl_sock_create()", status);
	    cli_err++;
	    clients_num--;
	    continue;
	}

	status = pj_ssl_sock_start_connect(ssock_cli[i], pool, &addr, &listen_addr, pj_sockaddr_get_len(&addr));
	if (status == PJ_SUCCESS) {
	    ssl_on_connect_complete(ssock_cli[i], PJ_SUCCESS);
	} else if (status == PJ_EPENDING) {
	    status = PJ_SUCCESS;
	} else {
	    app_perror("...ERROR pj_ssl_sock_create()", status);
	    pj_ssl_sock_close(ssock_cli[i]);
	    ssock_cli[i] = NULL;
	    clients_num--;
	    cli_err++;
	    continue;
	}

	/* Give chance to server to accept this client */
	{
	    unsigned n = 5;

#ifdef PJ_SYMBIAN
	    while(n && pj_symbianos_poll(-1, 1000))
		n--;
#else
	    pj_time_val delay = {0, 100};
	    while(n && pj_ioqueue_poll(ioqueue, &delay) > 0)
		n--;
#endif
	}
    }

    /* Wait until everything has been sent/received or error */
    while (clients_num)
    {
#ifdef PJ_SYMBIAN
	pj_symbianos_poll(-1, 1000);
#else
	pj_time_val delay = {0, 100};
	pj_ioqueue_poll(ioqueue, &delay);
	pj_timer_heap_poll(timer, &delay);
#endif
    }

    /* Clean up sockets */
    {
	pj_time_val delay = {0, 500};
	while (pj_ioqueue_poll(ioqueue, &delay) > 0);
    }

    if (state_serv.err != PJ_SUCCESS) {
	status = state_serv.err;
	goto on_return;
    }

    PJ_LOG(3, ("", "...Done!"));

    /* SSL setup and data transfer duration */
    {

//.........这里部分代码省略.........
	status=-30; goto on_error;
    }
    if (status==PJ_EPENDING) {
	++pending_op;
    }

    // Client socket connect()
    status = pj_ioqueue_connect(ckey1, &addr, sizeof(addr));
    if (status!=PJ_SUCCESS && status != PJ_EPENDING) {
        app_perror("...ERROR in pj_ioqueue_connect()", rc);
	status=-40; goto on_error;
    }
    if (status==PJ_EPENDING) {
	++pending_op;
    }

    // Poll until connected
    callback_read_size = callback_write_size = 0;
    callback_accept_status = callback_connect_status = -2;
    callback_call_count = 0;

    callback_read_key = callback_write_key = 
        callback_accept_key = callback_connect_key = NULL;
    callback_accept_op = callback_read_op = callback_write_op = NULL;

    while (pending_op) {
	pj_time_val timeout = {1, 0};

#ifdef PJ_SYMBIAN
	callback_call_count = 0;
	pj_symbianos_poll(-1, 1000);
	status = callback_call_count;
#else
	status = pj_ioqueue_poll(ioque, &timeout);
#endif
	if (status > 0) {
            if (callback_accept_status != -2) {
                if (callback_accept_status != 0) {
                    status=-41; goto on_error;
                }
                if (callback_accept_key != skey) {
                    status=-42; goto on_error;
                }
                if (callback_accept_op != &accept_op) {
                    status=-43; goto on_error;
                }
                callback_accept_status = -2;
            }

            if (callback_connect_status != -2) {
                if (callback_connect_status != 0) {
                    status=-50; goto on_error;
                }
                if (callback_connect_key != ckey1) {
                    status=-51; goto on_error;
                }
                callback_connect_status = -2;
            }

	    if (status > pending_op) {
		PJ_LOG(3,(THIS_FILE,
			  "...error: pj_ioqueue_poll() returned %d "
			  "(only expecting %d)",
			  status, pending_op));
		return -52;
	    }

//.........这里部分代码省略.........
    TRACE_("start sendto...");
    bytes = bufsize;
    rc = pj_ioqueue_sendto(ckey, &write_op, send_buf, &bytes, 0, &dst_addr, 
			   sizeof(dst_addr));
    if (rc != PJ_SUCCESS && rc != PJ_EPENDING) {
        app_perror("...error: pj_ioqueue_sendto", rc);
	status=-30; goto on_error;
    } else if (rc == PJ_EPENDING) {
	send_pending = 1;
	PJ_LOG(3, (THIS_FILE, 
		   "......ok: sendto returned pending"));
    } else {
	send_pending = 0;
	PJ_LOG(3, (THIS_FILE, 
		   "......ok: sendto returned immediate success"));
    }

    // reset callback variables.
    callback_read_size = callback_write_size = 0;
    callback_accept_status = callback_connect_status = -2;
    callback_read_key = callback_write_key = 
        callback_accept_key = callback_connect_key = NULL;
    callback_read_op = callback_write_op = NULL;

    // Poll if pending.
    while (send_pending || recv_pending) {
	int rc;
	pj_time_val timeout = { 5, 0 };

	TRACE_("poll...");
#ifdef PJ_SYMBIAN
	rc = pj_symbianos_poll(-1, PJ_TIME_VAL_MSEC(timeout));
#else
	rc = pj_ioqueue_poll(ioque, &timeout);
#endif

	if (rc == 0) {
	    PJ_LOG(1,(THIS_FILE, "...ERROR: timed out..."));
	    status=-45; goto on_error;
        } else if (rc < 0) {
            app_perror("...ERROR in ioqueue_poll()", -rc);
	    status=-50; goto on_error;
	}

	if (callback_read_key != NULL) {
            if (callback_read_size != bufsize) {
                status=-61; goto on_error;
            }
            if (callback_read_key != skey) {
                status=-65; goto on_error;
            }
            if (callback_read_op != &read_op) {
                status=-66; goto on_error;
            }

	    if (pj_memcmp(send_buf, recv_buf, bufsize) != 0) {
		status=-67; goto on_error;
	    }
	    if (addrlen != sizeof(pj_sockaddr_in)) {
		status=-68; goto on_error;
	    }
	    if (addr.sin_family != pj_AF_INET()) {
		status=-69; goto on_error;
	    }

//.........这里部分代码省略.........

    /* Init operation key. */
    pj_ioqueue_op_key_init(&opkey, sizeof(opkey));

    /* Start reading. */
    bytes = sizeof(recvbuf);
    status = pj_ioqueue_recv( key, &opkey, recvbuf, &bytes, 0);
    if (status != PJ_EPENDING) {
	app_perror("Expecting PJ_EPENDING, but got this", status);
	return -150;
    }

    /* Init destination address. */
    addrlen = sizeof(addr);
    status = pj_sock_getsockname(rsock, &addr, &addrlen);
    if (status != PJ_SUCCESS) {
	app_perror("getsockname error", status);
	return -160;
    }

    /* Override address with 127.0.0.1, since getsockname will return
     * zero in the address field.
     */
    addr.sin_addr = pj_inet_addr2("127.0.0.1");

    /* Init buffer to send */
    pj_ansi_strcpy(sendbuf, "Hello0123");

    /* Send one packet. */
    bytes = sizeof(sendbuf);
    status = pj_sock_sendto(ssock, sendbuf, &bytes, 0,
			    &addr, sizeof(addr));

    if (status != PJ_SUCCESS) {
	app_perror("sendto error", status);
	return -170;
    }

    /* Check if packet is received. */
    timeout.sec = 1; timeout.msec = 0;
#ifdef PJ_SYMBIAN
    pj_symbianos_poll(-1, 1000);
#else
    pj_ioqueue_poll(ioqueue, &timeout);
#endif

    if (packet_cnt != 1) {
	return -180;
    }

    /* Just to make sure things are settled.. */
    pj_thread_sleep(100);

    /* Start reading again. */
    bytes = sizeof(recvbuf);
    status = pj_ioqueue_recv( key, &opkey, recvbuf, &bytes, 0);
    if (status != PJ_EPENDING) {
	app_perror("Expecting PJ_EPENDING, but got this", status);
	return -190;
    }

    /* Reset packet counter */
    packet_cnt = 0;

    /* Send one packet. */
    bytes = sizeof(sendbuf);
    status = pj_sock_sendto(ssock, sendbuf, &bytes, 0,
			    &addr, sizeof(addr));

    if (status != PJ_SUCCESS) {
	app_perror("sendto error", status);
	return -200;
    }

    /* Now unregister and close socket. */
    pj_ioqueue_unregister(key);

    /* Poll ioqueue. */
#ifdef PJ_SYMBIAN
    pj_symbianos_poll(-1, 1000);
#else
    timeout.sec = 1; timeout.msec = 0;
    pj_ioqueue_poll(ioqueue, &timeout);
#endif

    /* Must NOT receive any packets after socket is closed! */
    if (packet_cnt > 0) {
	PJ_LOG(3,(THIS_FILE, "....errror: not expecting to receive packet "
			     "after socket has been closed"));
	return -210;
    }

    /* Success */
    pj_sock_close(ssock);
    pj_ioqueue_destroy(ioqueue);

    pj_pool_release(pool);

    return 0;
}

int http_client_test_delete()
{
    pj_str_t url;
    pj_http_req_callback hcb;
    pj_http_req_param param;
    char urlbuf[80];

    pj_bzero(&hcb, sizeof(hcb));
    hcb.on_complete = &on_complete;
    hcb.on_response = &on_response;

    /* Create pool, timer, and ioqueue */
    pool = pj_pool_create(mem, NULL, 8192, 4096, NULL);
    if (pj_timer_heap_create(pool, 16, &timer_heap))
        return -61;
    if (pj_ioqueue_create(pool, 16, &ioqueue))
        return -62;

#ifdef USE_LOCAL_SERVER
    thread_quit = PJ_FALSE;
    g_server.action = ACTION_REPLY;
    g_server.send_content_length = PJ_TRUE;
    g_server.data_size = 0;
    g_server.buf_size = 1024;

    sstatus = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, 
                             &g_server.sock);
    if (sstatus != PJ_SUCCESS)
        return -41;

    pj_sockaddr_in_init(&addr, NULL, 0);

    sstatus = pj_sock_bind(g_server.sock, &addr, sizeof(addr));
    if (sstatus != PJ_SUCCESS)
        return -43;

    {
	pj_sockaddr_in addr;
	int addr_len = sizeof(addr);
	sstatus = pj_sock_getsockname(g_server.sock, &addr, &addr_len);
	if (sstatus != PJ_SUCCESS)
	    return -44;
	g_server.port = pj_sockaddr_in_get_port(&addr);
	pj_ansi_snprintf(urlbuf, sizeof(urlbuf),
			 "http://127.0.0.1:%d/test/test2.txt",
			 g_server.port);
	url = pj_str(urlbuf);
    }

    sstatus = pj_sock_listen(g_server.sock, 8);
    if (sstatus != PJ_SUCCESS)
        return -45;

    sstatus = pj_thread_create(pool, NULL, &server_thread, &g_server,
                               0, 0, &g_server.thread);
    if (sstatus != PJ_SUCCESS)
        return -47;

#else
    pj_cstr(&url, "http://127.0.0.1:280/test/test2.txt");
#endif

    pj_http_req_param_default(&param);
    pj_strset2(&param.method, (char*)"DELETE");
    if (pj_http_req_create(pool, &url, timer_heap, ioqueue, 
                           &param, &hcb, &http_req))
        return -63;

    if (pj_http_req_start(http_req))
        return -65;

    while (pj_http_req_is_running(http_req)) {
        pj_time_val delay = {0, 50};
	pj_ioqueue_poll(ioqueue, &delay);
	pj_timer_heap_poll(timer_heap, NULL);
    }

#ifdef USE_LOCAL_SERVER
    thread_quit = PJ_TRUE;
    pj_thread_join(g_server.thread);
    pj_sock_close(g_server.sock);
#endif

    pj_http_req_destroy(http_req);
    pj_ioqueue_destroy(ioqueue);
    pj_timer_heap_destroy(timer_heap);
    pj_pool_release(pool);

    return PJ_SUCCESS;
}