/**
 * Create UDP stream transport.
 */
PJ_DEF(pj_status_t) pjmedia_transport_udp_create2(pjmedia_endpt *endpt,
						  const char *name,
						  const pj_str_t *addr,
						  int port,
						  unsigned options,
						  pjmedia_transport **p_tp)
{
    pjmedia_sock_info si;
    pj_status_t status;

    
    /* Sanity check */
    PJ_ASSERT_RETURN(endpt && port && p_tp, PJ_EINVAL);


    pj_bzero(&si, sizeof(pjmedia_sock_info));
    si.rtp_sock = si.rtcp_sock = PJ_INVALID_SOCKET;

    /* Create RTP socket */
    status = pj_sock_socket(PJ_AF_INET, PJ_SOCK_DGRAM, 0, &si.rtp_sock);
    if (status != PJ_SUCCESS)
	goto on_error;

    /* Bind RTP socket */
    pj_sockaddr_in_init(&si.rtp_addr_name, addr, (pj_uint16_t)port);
    status = pj_sock_bind(si.rtp_sock, &si.rtp_addr_name, 
			  sizeof(si.rtp_addr_name));
    if (status != PJ_SUCCESS)
	goto on_error;


    /* Create RTCP socket */
    status = pj_sock_socket(PJ_AF_INET, PJ_SOCK_DGRAM, 0, &si.rtcp_sock);
    if (status != PJ_SUCCESS)
	goto on_error;

    /* Bind RTCP socket */
    pj_sockaddr_in_init(&si.rtcp_addr_name, addr, (pj_uint16_t)(port+1));
    status = pj_sock_bind(si.rtcp_sock, &si.rtcp_addr_name,
			  sizeof(si.rtcp_addr_name));
    if (status != PJ_SUCCESS)
	goto on_error;

    
    /* Create UDP transport by attaching socket info */
    return pjmedia_transport_udp_attach( endpt, name, &si, options, p_tp);


on_error:
    if (si.rtp_sock != PJ_INVALID_SOCKET)
	pj_sock_close(si.rtp_sock);
    if (si.rtcp_sock != PJ_INVALID_SOCKET)
	pj_sock_close(si.rtcp_sock);
    return status;
}

int main() {
    pj_sock_t sock;
    pj_sockaddr_in to_addr;
    char *s = "Cong hoa xa hoi chu nghia VietNam";

    char buffer[100];
    pj_ssize_t len;

    struct timeval tv_begin, tv_end, tv_diff;

    pj_log_set_level(3);
    CHECK(__FILE__, pj_init());
    pj_bzero(buffer, sizeof(buffer));
    CHECK_R( __FILE__, pj_sock_socket(PJ_AF_INET, PJ_SOCK_DGRAM, 0, &sock) );
    //udp_socket(12345, &sock);
    setup_addr_with_host_and_port(&to_addr, "127.0.0.1", 33333);

    len = strlen(s);
    gettimeofday(&tv_begin, NULL);
    pj_sock_sendto(sock, s, &len, 0, &to_addr, sizeof(to_addr));
    
    PJ_LOG(3, (__FILE__, "Sent: %s", s));
    len = 100;
    pj_sock_recv(sock, buffer, &len, 0);
    gettimeofday(&tv_end, NULL);
    int diff = timeval_subtract(&tv_diff, &tv_end, &tv_begin);
    PJ_LOG(3, (__FILE__, "Received: %s %0.2f msec", buffer, diff*1.0/1000));
    pj_shutdown();
    return 0;
}

/* Create socket */
static pj_status_t create_socket(int af, const pj_sockaddr_t *local_a,
				 int addr_len, pj_sock_t *p_sock)
{
    pj_sock_t sock;
    pj_sockaddr_in tmp_addr;
    pj_status_t status;

    status = pj_sock_socket(af, pj_SOCK_DGRAM(), 0, &sock);
    if (status != PJ_SUCCESS)
	return status;

    if (local_a == NULL) {
      pj_sockaddr_in_init(&tmp_addr, NULL, 0);
      local_a = &tmp_addr;
      addr_len = sizeof(tmp_addr);
    }

    status = pj_sock_bind(sock, local_a, addr_len);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    *p_sock = sock;
    return PJ_SUCCESS;
}

/* Create socket */
static pj_status_t create_socket(int af, const pj_sockaddr_t *local_a,
				 int addr_len, pj_sock_t *p_sock)
{
    pj_sock_t sock;
    pj_sockaddr_in tmp_addr;
    pj_sockaddr_in6 tmp_addr6;
    pj_status_t status;

    status = pj_sock_socket(af, pj_SOCK_DGRAM(), 0, &sock);
    if (status != PJ_SUCCESS)
	return status;

    if (local_a == NULL) {
	if (af == pj_AF_INET6()) {
	    pj_bzero(&tmp_addr6, sizeof(tmp_addr6));
	    tmp_addr6.sin6_family = (pj_uint16_t)af;
	    local_a = &tmp_addr6;
	    addr_len = sizeof(tmp_addr6);
	} else {
	    pj_sockaddr_in_init(&tmp_addr, NULL, 0);
	    local_a = &tmp_addr;
	    addr_len = sizeof(tmp_addr);
	}
    }

    status = pj_sock_bind(sock, local_a, addr_len);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    *p_sock = sock;
    return PJ_SUCCESS;
}

static void open_udp_socket($UPROTO$_server_t *userver, char *addr, int port) {
    pj_sockaddr_in saddr;
    pj_str_t s;
    int optval = 1;
    SHOW_LOG(3, __FILE__":open_udp_socket: %s:%d\n", addr, port);   
    // create udp socket here
    CHECK(__FILE__, pj_sock_socket(PJ_AF_INET, PJ_SOCK_DGRAM, 0, &userver->fd));

#ifdef __ICS_INTEL__
    // Allow socket reuse
    //CHECK(__FILE__, pj_sock_setsockopt(userver->fd, PJ_SOL_SOCKET, 512, &optval, sizeof(optval)));
    CHECK(__FILE__, pj_sock_setsockopt(userver->fd, PJ_SOL_SOCKET, PJ_SO_REUSEADDR, &optval, sizeof(optval)));
#endif

    // bind the socket
    pj_bzero(&saddr, sizeof(saddr));
    saddr.sin_family = PJ_AF_INET;
    saddr.sin_port = pj_htons(port);
    saddr.sin_addr = pj_inet_addr(pj_cstr(&s,addr));

    /*pj_status_t status = pj_sock_bind(userver->fd, &saddr, sizeof(saddr));
    if( status != 0 ) {
        SHOW_LOG(1, __FILE__":open_udp_socket error(%d)\n", status);
        exit(-1);
    }*/
    CHECK(__FILE__, pj_sock_bind(userver->fd, &saddr, sizeof(saddr)));
}

static int simple_sock_test(void)
{
    int types[2];
    pj_sock_t sock;
    int i;
    pj_status_t rc = PJ_SUCCESS;

    types[0] = pj_SOCK_STREAM();
    types[1] = pj_SOCK_DGRAM();

    PJ_LOG(3,("test", "...simple_sock_test()"));

    for (i=0; i<(int)(sizeof(types)/sizeof(types[0])); ++i) {
	
	rc = pj_sock_socket(pj_AF_INET(), types[i], 0, &sock);
	if (rc != PJ_SUCCESS) {
	    app_perror("...error: unable to create socket", rc);
	    break;
	} else {
	    rc = pj_sock_close(sock);
	    if (rc != 0) {
		app_perror("...error: close socket", rc);
		break;
	    }
	}
    }
    return rc;
}

pj_status_t app_socket(int family, int type, int proto, int port,
                       pj_sock_t *ptr_sock)
{
    pj_sockaddr_in addr;
    pj_sock_t sock;
    pj_status_t rc;

    rc = pj_sock_socket(family, type, proto, &sock);
    if (rc != PJ_SUCCESS)
        return rc;

    pj_bzero(&addr, sizeof(addr));
    addr.sin_family = (pj_uint16_t)family;
    addr.sin_port = (short)(port!=-1 ? pj_htons((pj_uint16_t)port) : 0);
    rc = pj_sock_bind(sock, &addr, sizeof(addr));
    if (rc != PJ_SUCCESS)
        return rc;
    
#if PJ_HAS_TCP
    if (type == pj_SOCK_STREAM()) {
        rc = pj_sock_listen(sock, 5);
        if (rc != PJ_SUCCESS)
            return rc;
    }
#endif

    *ptr_sock = sock;
    return PJ_SUCCESS;
}

/* Get the default IP interface */
PJ_DEF(pj_status_t) pj_getdefaultipinterface(int af, pj_sockaddr *addr)
{
    pj_sock_t fd;
    pj_str_t cp;
    pj_sockaddr a;
    int len;
    pj_uint8_t zero[64];
    pj_status_t status;

    addr->addr.sa_family = (pj_uint16_t)af;

    status = pj_sock_socket(af, pj_SOCK_DGRAM(), 0, &fd);
    if (status != PJ_SUCCESS) {
	return status;
	}
	PJ_LOG(4, ("sock_common.c", "pj_getdefaultipinterface() pj_sock_socket."));

    if (af == PJ_AF_INET) {
	cp = pj_str("1.1.1.1");
    } else {
	cp = pj_str("1::1");
    }
    status = pj_sockaddr_init(af, &a, &cp, 53);
    if (status != PJ_SUCCESS) {
	pj_sock_close(fd);
	return status;
	}
	PJ_LOG(4, ("sock_common.c", "pj_getdefaultipinterface() pj_sockaddr_init."));

    status = pj_sock_connect(fd, &a, pj_sockaddr_get_len(&a));
    if (status != PJ_SUCCESS) {
	pj_sock_close(fd);
	return status;
	}
	PJ_LOG(4, ("sock_common.c", "pj_getdefaultipinterface() pj_sock_connect."));

    len = sizeof(a);
    status = pj_sock_getsockname(fd, &a, &len);
    if (status != PJ_SUCCESS) {
	pj_sock_close(fd);
	return status;
	}
	PJ_LOG(4, ("sock_common.c", "pj_getdefaultipinterface() pj_sock_getsockname."));

    pj_sock_close(fd);

    /* Check that the address returned is not zero */
    pj_bzero(zero, sizeof(zero));
    if (pj_memcmp(pj_sockaddr_get_addr(&a), zero,
		  pj_sockaddr_get_addr_len(&a))==0)
	{
		PJ_LOG(4, ("sock_common.c", "pj_getdefaultipinterface() interface not found."));
		return PJ_ENOTFOUND;
    }

    pj_sockaddr_copy_addr(addr, &a);

    /* Success */
    return PJ_SUCCESS;
}

static void open_udp_socket(response_client_t *uclient, char *server, int port) {
    pj_sockaddr_in *saddr;
    pj_str_t s;
    // open socket
    CHECK(__FILE__, pj_sock_socket(PJ_AF_INET, PJ_SOCK_STREAM, 0, &uclient->fd));

    uclient->connect_data = malloc(sizeof(pj_sockaddr_in));
    saddr = (pj_sockaddr_in *)uclient->connect_data;
    pj_bzero((void *)saddr, sizeof(pj_sockaddr_in));
    saddr->sin_family = PJ_AF_INET;
    saddr->sin_port = pj_htons(port);
    saddr->sin_addr = pj_inet_addr(pj_cstr(&s, server));
    
    CHECK(__FILE__,pj_sock_connect(uclient->fd, (const pj_sockaddr_t *)saddr, sizeof(pj_sockaddr_in)));
}

PJ_DEF(pj_status_t) pj_activesock_create_udp( pj_pool_t *pool,
					      const pj_sockaddr *addr,
					      const pj_activesock_cfg *opt,
					      pj_ioqueue_t *ioqueue,
					      const pj_activesock_cb *cb,
					      void *user_data,
					      pj_activesock_t **p_asock,
					      pj_sockaddr *bound_addr)
{
    pj_sock_t sock_fd;
    pj_sockaddr default_addr;
    pj_status_t status;

    if (addr == NULL) {
	pj_sockaddr_init(pj_AF_INET(), &default_addr, NULL, 0);
	addr = &default_addr;
    }

    status = pj_sock_socket(addr->addr.sa_family, pj_SOCK_DGRAM(), 0, 
			    &sock_fd);
    if (status != PJ_SUCCESS) {
	return status;
    }

    status = pj_sock_bind(sock_fd, addr, pj_sockaddr_get_len(addr));
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock_fd);
	return status;
    }

    status = pj_activesock_create(pool, sock_fd, pj_SOCK_DGRAM(), opt,
				  ioqueue, cb, user_data, p_asock);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock_fd);
	return status;
    }

    if (bound_addr) {
	int addr_len = sizeof(*bound_addr);
	status = pj_sock_getsockname(sock_fd, bound_addr, &addr_len);
	if (status != PJ_SUCCESS) {
	    pj_activesock_close(*p_asock);
	    return status;
	}
    }

    return PJ_SUCCESS;
}

int CPjSSLSocket::Connect(CPjSSLSocket_cb cb, void *key, 
			  const TInetAddr &local_addr, 
			  const TInetAddr &rem_addr,
			  const TDesC8 &servername,
			  const TDesC8 &ciphers)
{
    pj_status_t status;
    
    PJ_ASSERT_RETURN(state_ == SSL_STATE_NULL, PJ_EINVALIDOP);
    
    status = pj_sock_socket(rem_addr.Family(), pj_SOCK_STREAM(), 0, &sock_);
    if (status != PJ_SUCCESS)
	return status;

    // Apply QoS
    status = pj_sock_apply_qos2(sock_, qos_type_, &qos_params_, 
    				2,  THIS_FILE, NULL);
    
    RSocket &rSock = ((CPjSocket*)sock_)->Socket();

    local_addr_ = local_addr;
    
    if (!local_addr_.IsUnspecified()) {
	TInt err = rSock.Bind(local_addr_);
	if (err != KErrNone)
	    return PJ_RETURN_OS_ERROR(err);
    }
    
    cb_ = cb;
    key_ = key;
    rem_addr_ = rem_addr;
    
    /* Note: the following members only keep the pointer, not the data */
    servername_.Set(servername);
    ciphers_.Set(ciphers);

    rSock.Connect(rem_addr_, iStatus);
    SetActive();
    state_ = SSL_STATE_CONNECTING;
    
    rSock.LocalName(local_addr_);

    return PJ_EPENDING;
}

/* Initialize UDP socket */
static pj_status_t init_sock(pj_dns_resolver *resv)
{
    pj_ioqueue_callback socket_cb;
    pj_status_t status;

    /* Create the UDP socket */
    status = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &resv->udp_sock);
    if (status != PJ_SUCCESS)
	return status;

    /* Bind to any address/port */
    status = pj_sock_bind_in(resv->udp_sock, 0, 0);
    if (status != PJ_SUCCESS)
	return status;

    /* Register to ioqueue */
    pj_bzero(&socket_cb, sizeof(socket_cb));
    socket_cb.on_read_complete = &on_read_complete;
    status = pj_ioqueue_register_sock(resv->pool, resv->ioqueue,
				      resv->udp_sock, resv, &socket_cb,
				      &resv->udp_key);
    if (status != PJ_SUCCESS)
	return status;

    pj_ioqueue_op_key_init(&resv->udp_op_rx_key, sizeof(resv->udp_op_rx_key));
    pj_ioqueue_op_key_init(&resv->udp_op_tx_key, sizeof(resv->udp_op_tx_key));

    /* Start asynchronous read to the UDP socket */
    resv->udp_len = sizeof(resv->udp_rx_pkt);
    resv->udp_addr_len = sizeof(resv->udp_src_addr);
    status = pj_ioqueue_recvfrom(resv->udp_key, &resv->udp_op_rx_key,
				 resv->udp_rx_pkt, &resv->udp_len,
				 PJ_IOQUEUE_ALWAYS_ASYNC,
				 &resv->udp_src_addr, &resv->udp_addr_len);
    if (status != PJ_EPENDING)
	return status;

    return PJ_SUCCESS;
}

static pj_status_t get_local_interface(const pj_sockaddr *server,
				       pj_sockaddr *local_addr)
{
    pj_sock_t sock;
    pj_sockaddr tmp, local;
    int addr_len;
    pj_status_t status;

    status = pj_sock_socket(server->addr.sa_family, pj_SOCK_DGRAM(),
    			    0, &sock);
    if (status != PJ_SUCCESS)
	return status;

    addr_len = pj_sockaddr_get_len(server);
    pj_sockaddr_init(server->addr.sa_family, &local, NULL, 0);
    status = pj_sock_bind(sock, &local, addr_len);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    status = pj_sock_connect(sock, server, addr_len);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    status = pj_sock_getsockname(sock, &tmp, &addr_len);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    pj_sockaddr_cp(local_addr, &tmp);
    
    pj_sock_close(sock);
    return PJ_SUCCESS;
}

static pj_status_t get_local_interface(const pj_sockaddr_in *server,
				       pj_in_addr *local_addr)
{
    pj_sock_t sock;
    pj_sockaddr_in tmp;
    int addr_len;
    pj_status_t status;

    status = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &sock);
    if (status != PJ_SUCCESS)
	return status;

    status = pj_sock_bind_in(sock, 0, 0);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    status = pj_sock_connect(sock, server, sizeof(pj_sockaddr_in));
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    addr_len = sizeof(pj_sockaddr_in);
    status = pj_sock_getsockname(sock, &tmp, &addr_len);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    local_addr->s_addr = tmp.sin_addr.s_addr;
    
    pj_sock_close(sock);
    return PJ_SUCCESS;
}

/*
 * Repeated connect/accept on the same listener socket.
 */
static int compliance_test_2(pj_bool_t allow_concur)
{
#if defined(PJ_SYMBIAN) && PJ_SYMBIAN!=0
    enum { MAX_PAIR = 1, TEST_LOOP = 2 };
#else
    enum { MAX_PAIR = 4, TEST_LOOP = 2 };
#endif

    struct listener
    {
	pj_sock_t	     sock;
	pj_ioqueue_key_t    *key;
	pj_sockaddr_in	     addr;
	int		     addr_len;
    } listener;

    struct server
    {
	pj_sock_t	     sock;
	pj_ioqueue_key_t    *key;
	pj_sockaddr_in	     local_addr;
	pj_sockaddr_in	     rem_addr;
	int		     rem_addr_len;
	pj_ioqueue_op_key_t  accept_op;
    } server[MAX_PAIR];

    struct client
    {
	pj_sock_t	     sock;
	pj_ioqueue_key_t    *key;
    } client[MAX_PAIR];

    pj_pool_t *pool = NULL;
    char *send_buf, *recv_buf;
    pj_ioqueue_t *ioque = NULL;
    int i, bufsize = BUF_MIN_SIZE;
    pj_ssize_t status;
    int test_loop, pending_op = 0;
    pj_timestamp t_elapsed;
    pj_str_t s;
    pj_status_t rc;

    listener.sock = PJ_INVALID_SOCKET;
    listener.key = NULL;
    
    for (i=0; i<MAX_PAIR; ++i) {
    	server[i].sock = PJ_INVALID_SOCKET;
    	server[i].key = NULL;
    }
    
    for (i=0; i<MAX_PAIR; ++i) {
    	client[i].sock = PJ_INVALID_SOCKET;
    	client[i].key = NULL;	
    }
    
    // Create pool.
    pool = pj_pool_create(mem, NULL, POOL_SIZE, 4000, NULL);


    // Create I/O Queue.
    rc = pj_ioqueue_create(pool, PJ_IOQUEUE_MAX_HANDLES, &ioque);
    if (rc != PJ_SUCCESS) {
        app_perror("...ERROR in pj_ioqueue_create()", rc);
	return -10;
    }


    // Concurrency
    rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
    if (rc != PJ_SUCCESS) {
        app_perror("...ERROR in pj_ioqueue_set_default_concurrency()", rc);
	return -11;
    }

    // Allocate buffers for send and receive.
    send_buf = (char*)pj_pool_alloc(pool, bufsize);
    recv_buf = (char*)pj_pool_alloc(pool, bufsize);

    // Create listener socket
    rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, &listener.sock);
    if (rc != PJ_SUCCESS) {
        app_perror("...error creating socket", rc);
        status=-20; goto on_error;
    }

    // Bind listener socket.
    pj_sockaddr_in_init(&listener.addr, 0, 0);
    if ((rc=pj_sock_bind(listener.sock, &listener.addr, sizeof(listener.addr))) != 0 ) {
        app_perror("...bind error", rc);
	status=-30; goto on_error;
    }

    // Get listener address.
    listener.addr_len = sizeof(listener.addr);
    rc = pj_sock_getsockname(listener.sock, &listener.addr, &listener.addr_len);
    if (rc != PJ_SUCCESS) {
        app_perror("...ERROR in pj_sock_getsockname()", rc);
//.........这里部分代码省略.........

/*
 * Benchmarking IOQueue
 */
static int bench_test(pj_bool_t allow_concur, int bufsize, 
		      int inactive_sock_count)
{
    pj_sock_t ssock=-1, csock=-1;
    pj_sockaddr_in addr;
    pj_pool_t *pool = NULL;
    pj_sock_t *inactive_sock=NULL;
    pj_ioqueue_op_key_t *inactive_read_op;
    char *send_buf, *recv_buf;
    pj_ioqueue_t *ioque = NULL;
    pj_ioqueue_key_t *skey, *ckey, *keys[SOCK_INACTIVE_MAX+2];
    pj_timestamp t1, t2, t_elapsed;
    int rc=0, i;    /* i must be signed */
    pj_str_t temp;
    char errbuf[PJ_ERR_MSG_SIZE];

    TRACE__((THIS_FILE, "   bench test %d", inactive_sock_count));

    // Create pool.
    pool = pj_pool_create(mem, NULL, POOL_SIZE, 4000, NULL);

    // Allocate buffers for send and receive.
    send_buf = (char*)pj_pool_alloc(pool, bufsize);
    recv_buf = (char*)pj_pool_alloc(pool, bufsize);

    // Allocate sockets for sending and receiving.
    rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &ssock);
    if (rc == PJ_SUCCESS) {
        rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &csock);
    } else
        csock = PJ_INVALID_SOCKET;
    if (rc != PJ_SUCCESS) {
	app_perror("...error: pj_sock_socket()", rc);
	goto on_error;
    }

    // Bind server socket.
    pj_bzero(&addr, sizeof(addr));
    addr.sin_family = pj_AF_INET();
    addr.sin_port = pj_htons(PORT);
    if (pj_sock_bind(ssock, &addr, sizeof(addr)))
	goto on_error;

    pj_assert(inactive_sock_count+2 <= PJ_IOQUEUE_MAX_HANDLES);

    // Create I/O Queue.
    rc = pj_ioqueue_create(pool, PJ_IOQUEUE_MAX_HANDLES, &ioque);
    if (rc != PJ_SUCCESS) {
	app_perror("...error: pj_ioqueue_create()", rc);
	goto on_error;
    }

    // Set concurrency
    rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
    if (rc != PJ_SUCCESS) {
	app_perror("...error: pj_ioqueue_set_default_concurrency()", rc);
	goto on_error;
    }

    // Allocate inactive sockets, and bind them to some arbitrary address.
    // Then register them to the I/O queue, and start a read operation.
    inactive_sock = (pj_sock_t*)pj_pool_alloc(pool, 
				    inactive_sock_count*sizeof(pj_sock_t));
    inactive_read_op = (pj_ioqueue_op_key_t*)pj_pool_alloc(pool,
                              inactive_sock_count*sizeof(pj_ioqueue_op_key_t));
    pj_bzero(&addr, sizeof(addr));
    addr.sin_family = pj_AF_INET();
    for (i=0; i<inactive_sock_count; ++i) {
        pj_ssize_t bytes;

	rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &inactive_sock[i]);
	if (rc != PJ_SUCCESS || inactive_sock[i] < 0) {
	    app_perror("...error: pj_sock_socket()", rc);
	    goto on_error;
	}
	if ((rc=pj_sock_bind(inactive_sock[i], &addr, sizeof(addr))) != 0) {
	    pj_sock_close(inactive_sock[i]);
	    inactive_sock[i] = PJ_INVALID_SOCKET;
	    app_perror("...error: pj_sock_bind()", rc);
	    goto on_error;
	}
	rc = pj_ioqueue_register_sock(pool, ioque, inactive_sock[i], 
			              NULL, &test_cb, &keys[i]);
	if (rc != PJ_SUCCESS) {
	    pj_sock_close(inactive_sock[i]);
	    inactive_sock[i] = PJ_INVALID_SOCKET;
	    app_perror("...error(1): pj_ioqueue_register_sock()", rc);
	    PJ_LOG(3,(THIS_FILE, "....i=%d", i));
	    goto on_error;
	}
        bytes = bufsize;
	rc = pj_ioqueue_recv(keys[i], &inactive_read_op[i], recv_buf, &bytes, 0);
	if (rc != PJ_EPENDING) {
	    pj_sock_close(inactive_sock[i]);
	    inactive_sock[i] = PJ_INVALID_SOCKET;
	    app_perror("...error: pj_ioqueue_read()", rc);
	    goto on_error;
//.........这里部分代码省略.........

/*
 * Testing with many handles.
 * This will just test registering PJ_IOQUEUE_MAX_HANDLES count
 * of sockets to the ioqueue.
 */
static int many_handles_test(pj_bool_t allow_concur)
{
    enum { MAX = PJ_IOQUEUE_MAX_HANDLES };
    pj_pool_t *pool;
    pj_ioqueue_t *ioqueue;
    pj_sock_t *sock;
    pj_ioqueue_key_t **key;
    pj_status_t rc;
    int count, i; /* must be signed */

    PJ_LOG(3,(THIS_FILE,"...testing with so many handles"));

    pool = pj_pool_create(mem, NULL, 4000, 4000, NULL);
    if (!pool)
	return PJ_ENOMEM;

    key = (pj_ioqueue_key_t**) 
    	  pj_pool_alloc(pool, MAX*sizeof(pj_ioqueue_key_t*));
    sock = (pj_sock_t*) pj_pool_alloc(pool, MAX*sizeof(pj_sock_t));
    
    /* Create IOQueue */
    rc = pj_ioqueue_create(pool, MAX, &ioqueue);
    if (rc != PJ_SUCCESS || ioqueue == NULL) {
	app_perror("...error in pj_ioqueue_create", rc);
	return -10;
    }

    // Set concurrency
    rc = pj_ioqueue_set_default_concurrency(ioqueue, allow_concur);
    if (rc != PJ_SUCCESS) {
	return -11;
    }

    /* Register as many sockets. */
    for (count=0; count<MAX; ++count) {
	sock[count] = PJ_INVALID_SOCKET;
	rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &sock[count]);
	if (rc != PJ_SUCCESS || sock[count] == PJ_INVALID_SOCKET) {
	    PJ_LOG(3,(THIS_FILE, "....unable to create %d-th socket, rc=%d", 
				 count, rc));
	    break;
	}
	key[count] = NULL;
	rc = pj_ioqueue_register_sock(pool, ioqueue, sock[count],
				      NULL, &test_cb, &key[count]);
	if (rc != PJ_SUCCESS || key[count] == NULL) {
	    PJ_LOG(3,(THIS_FILE, "....unable to register %d-th socket, rc=%d", 
				 count, rc));
	    return -30;
	}
    }

    /* Test complete. */

    /* Now deregister and close all handles. */ 

    /* NOTE for RTEMS:
     *  It seems that the order of close(sock) is pretty important here.
     *  If we close the sockets with the same order as when they were created,
     *  RTEMS doesn't seem to reuse the sockets, thus next socket created
     *  will have descriptor higher than the last socket created.
     *  If we close the sockets in the reverse order, then the descriptor will
     *  get reused.
     *  This used to cause problem with select ioqueue, since the ioqueue
     *  always gives FD_SETSIZE for the first select() argument. This ioqueue
     *  behavior can be changed with setting PJ_SELECT_NEEDS_NFDS macro.
     */
    for (i=count-1; i>=0; --i) {
    ///for (i=0; i<count; ++i) {
	rc = pj_ioqueue_unregister(key[i]);
	if (rc != PJ_SUCCESS) {
	    app_perror("...error in pj_ioqueue_unregister", rc);
	}
    }

    rc = pj_ioqueue_destroy(ioqueue);
    if (rc != PJ_SUCCESS) {
	app_perror("...error in pj_ioqueue_destroy", rc);
    }
    
    pj_pool_release(pool);

    PJ_LOG(3,(THIS_FILE,"....many_handles_test() ok"));

    return 0;
}

/*
 * compliance_test()
 * To test that the basic IOQueue functionality works. It will just exchange
 * data between two sockets.
 */ 
static int compliance_test(pj_bool_t allow_concur)
{
    pj_sock_t ssock=-1, csock=-1;
    pj_sockaddr_in addr, dst_addr;
    int addrlen;
    pj_pool_t *pool = NULL;
    char *send_buf, *recv_buf;
    pj_ioqueue_t *ioque = NULL;
    pj_ioqueue_key_t *skey = NULL, *ckey = NULL;
    pj_ioqueue_op_key_t read_op, write_op;
    int bufsize = BUF_MIN_SIZE;
    pj_ssize_t bytes;
    int status = -1;
    pj_str_t temp;
    pj_bool_t send_pending, recv_pending;
    pj_status_t rc;

    pj_set_os_error(PJ_SUCCESS);

    // Create pool.
    pool = pj_pool_create(mem, NULL, POOL_SIZE, 4000, NULL);

    // Allocate buffers for send and receive.
    send_buf = (char*)pj_pool_alloc(pool, bufsize);
    recv_buf = (char*)pj_pool_alloc(pool, bufsize);

    // Allocate sockets for sending and receiving.
    TRACE_("creating sockets...");
    rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &ssock);
    if (rc==PJ_SUCCESS)
        rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &csock);
    else
        csock = PJ_INVALID_SOCKET;
    if (rc != PJ_SUCCESS) {
        app_perror("...ERROR in pj_sock_socket()", rc);
	status=-1; goto on_error;
    }

    // Bind server socket.
    TRACE_("bind socket...");
    pj_bzero(&addr, sizeof(addr));
    addr.sin_family = pj_AF_INET();
    addr.sin_port = pj_htons(PORT);
    if (pj_sock_bind(ssock, &addr, sizeof(addr))) {
	status=-10; goto on_error;
    }

    // Create I/O Queue.
    TRACE_("create ioqueue...");
    rc = pj_ioqueue_create(pool, PJ_IOQUEUE_MAX_HANDLES, &ioque);
    if (rc != PJ_SUCCESS) {
	status=-20; goto on_error;
    }

    // Set concurrency
    TRACE_("set concurrency...");
    rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
    if (rc != PJ_SUCCESS) {
	status=-21; goto on_error;
    }

    // Register server and client socket.
    // We put this after inactivity socket, hopefully this can represent the
    // worst waiting time.
    TRACE_("registering first sockets...");
    rc = pj_ioqueue_register_sock(pool, ioque, ssock, NULL, 
			          &test_cb, &skey);
    if (rc != PJ_SUCCESS) {
	app_perror("...error(10): ioqueue_register error", rc);
	status=-25; goto on_error;
    }
    TRACE_("registering second sockets...");
    rc = pj_ioqueue_register_sock( pool, ioque, csock, NULL, 
			           &test_cb, &ckey);
    if (rc != PJ_SUCCESS) {
	app_perror("...error(11): ioqueue_register error", rc);
	status=-26; goto on_error;
    }

    // Randomize send_buf.
    pj_create_random_string(send_buf, bufsize);

    // Register reading from ioqueue.
    TRACE_("start recvfrom...");
    pj_bzero(&addr, sizeof(addr));
    addrlen = sizeof(addr);
    bytes = bufsize;
    rc = pj_ioqueue_recvfrom(skey, &read_op, recv_buf, &bytes, 0,
			     &addr, &addrlen);
    if (rc != PJ_SUCCESS && rc != PJ_EPENDING) {
        app_perror("...error: pj_ioqueue_recvfrom", rc);
	status=-28; goto on_error;
    } else if (rc == PJ_EPENDING) {
	recv_pending = 1;
	PJ_LOG(3, (THIS_FILE, 
//.........这里部分代码省略.........

static int stun_destroy_test(void)
{
    enum { LOOP = 500 };
    struct stun_test_session test_sess;
    pj_sockaddr bind_addr;
    int addr_len;
    pj_caching_pool cp;
    pj_pool_t *pool;
    unsigned i;
    pj_status_t status;
    int rc = 0;

    PJ_LOG(3,(THIS_FILE, "  STUN destroy concurrency test"));

    pj_bzero(&test_sess, sizeof(test_sess));

    pj_caching_pool_init(&cp, NULL, 0);
    pool = pj_pool_create(&cp.factory, "testsess", 512, 512, NULL);

    pj_stun_config_init(&test_sess.stun_cfg, &cp.factory, 0, NULL, NULL);

    status = pj_timer_heap_create(pool, 1023, &test_sess.stun_cfg.timer_heap);
    pj_assert(status == PJ_SUCCESS);

    status = pj_lock_create_recursive_mutex(pool, NULL, &test_sess.lock);
    pj_assert(status == PJ_SUCCESS);

    pj_timer_heap_set_lock(test_sess.stun_cfg.timer_heap, test_sess.lock, PJ_TRUE);
    pj_assert(status == PJ_SUCCESS);

    status = pj_ioqueue_create(pool, 512, &test_sess.stun_cfg.ioqueue);
    pj_assert(status == PJ_SUCCESS);

    pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &test_sess.server_sock);
    pj_sockaddr_init(pj_AF_INET(), &bind_addr, NULL, 0);
    status = pj_sock_bind(test_sess.server_sock, &bind_addr, pj_sockaddr_get_len(&bind_addr));
    pj_assert(status == PJ_SUCCESS);

    addr_len = sizeof(bind_addr);
    status = pj_sock_getsockname(test_sess.server_sock, &bind_addr, &addr_len);
    pj_assert(status == PJ_SUCCESS);

    test_sess.server_port = pj_sockaddr_get_port(&bind_addr);

    status = pj_event_create(pool, NULL, PJ_TRUE, PJ_FALSE, &test_sess.server_event);
    pj_assert(status == PJ_SUCCESS);

    for (i=0; i<SERVER_THREAD_CNT; ++i) {
	status = pj_thread_create(pool, NULL,
	                          &server_thread_proc, &test_sess,
	                          0, 0, &test_sess.server_threads[i]);
	pj_assert(status == PJ_SUCCESS);
    }

    for (i=0; i<WORKER_THREAD_CNT; ++i) {
	status = pj_thread_create(pool, NULL,
	                          &worker_thread_proc, &test_sess,
	                          0, 0, &test_sess.worker_threads[i]);
	pj_assert(status == PJ_SUCCESS);
    }

    /* Test 1: Main thread calls destroy while callback is processing response */
    PJ_LOG(3,(THIS_FILE, "    Destroy in main thread while callback is running"));
    for (i=0; i<LOOP; ++i) {
	int sleep = pj_rand() % 5;

	PJ_LOG(3,(THIS_FILE, "      Try %-3d of %d", i+1, LOOP));

	/* Test 1: destroy at the same time when receiving response */
	pj_bzero(&test_sess.param, sizeof(test_sess.param));
	test_sess.param.client_sleep_after_start = 20;
	test_sess.param.client_sleep_before_destroy = sleep;
	test_sess.param.server_wait_for_event = PJ_TRUE;
	stun_destroy_test_session(&test_sess);
	PJ_LOG(3,(THIS_FILE,
		  "        stun test a: sleep delay:%d: clients with response: %d",
		  sleep, test_sess.param.client_got_response));

	/* Test 2: destroy at the same time with STUN retransmit timer */
	test_sess.param.server_drop_request = PJ_TRUE;
	test_sess.param.client_sleep_after_start = 0;
	test_sess.param.client_sleep_before_destroy = PJ_STUN_RTO_VALUE;
	test_sess.param.server_wait_for_event = PJ_FALSE;
	stun_destroy_test_session(&test_sess);
	PJ_LOG(3,(THIS_FILE, "        stun test b: retransmit concurrency"));

	/* Test 3: destroy at the same time with receiving response
	 * AND STUN retransmit timer */
	test_sess.param.client_got_response = 0;
	test_sess.param.server_drop_request = PJ_FALSE;
	test_sess.param.client_sleep_after_start = PJ_STUN_RTO_VALUE;
	test_sess.param.client_sleep_before_destroy = 0;
	test_sess.param.server_wait_for_event = PJ_TRUE;
	stun_destroy_test_session(&test_sess);
	PJ_LOG(3,(THIS_FILE,
		  "        stun test c: clients with response: %d",
		  test_sess.param.client_got_response));

	pj_thread_sleep(10);

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

static int run_client_test(const char *title,

			   pj_bool_t server_responding,
			   pj_stun_auth_type server_auth_type,

			   pj_stun_auth_type client_auth_type,
			   const char *realm,
			   const char *username,
			   const char *nonce,
			   const char *password,
			   pj_bool_t dummy_mi,

			   pj_bool_t expected_error,
			   pj_status_t expected_code,
			   const char *expected_realm,
			   const char *expected_nonce,
			   
			   int (*more_check)(void))
{
    pj_pool_t *pool;
    pj_stun_session_cb sess_cb;
    pj_stun_auth_cred cred;
    pj_stun_tx_data *tdata;
    pj_status_t status;
    int rc = 0;
    
    PJ_LOG(3,(THIS_FILE, "   %s test", title));

    /* Create client */
    pool = pj_pool_create(mem, "client", 1000, 1000, NULL);
    client = PJ_POOL_ZALLOC_T(pool, struct client);
    client->pool = pool;
    client->responding = PJ_TRUE;

    /* Create STUN session */
    pj_bzero(&sess_cb, sizeof(sess_cb));
    sess_cb.on_request_complete = &client_on_request_complete;
    sess_cb.on_send_msg = &client_send_msg;
    status = pj_stun_session_create(&stun_cfg, "client", &sess_cb, PJ_FALSE, NULL, &client->sess);
    if (status != PJ_SUCCESS) {
	destroy_client_server();
	return -200;
    }

    /* Create semaphore */
    status = pj_sem_create(pool, "client", 0, 1, &client->test_complete);
    if (status != PJ_SUCCESS) {
	destroy_client_server();
	return -205;
    }

    /* Create client socket */
    status = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &client->sock);
    if (status != PJ_SUCCESS) {
	destroy_client_server();
	return -210;
    }

    /* Bind client socket */
    status = pj_sock_bind_in(client->sock, 0, 0);
    if (status != PJ_SUCCESS) {
	destroy_client_server();
	return -220;
    }

    /* Create client thread */
    status = pj_thread_create(pool, "client", &client_thread, NULL, 0, 0, &client->thread);
    if (status != PJ_SUCCESS) {
	destroy_client_server();
	return -230;
    }

    /* Initialize credential */
    pj_bzero(&cred, sizeof(cred));
    cred.type = PJ_STUN_AUTH_CRED_STATIC;
    if (realm) cred.data.static_cred.realm = pj_str((char*)realm);
    if (username) cred.data.static_cred.username = pj_str((char*)username);
    if (nonce) cred.data.static_cred.nonce = pj_str((char*)nonce);
    if (password) cred.data.static_cred.data = pj_str((char*)password);
    cred.data.static_cred.data_type = PJ_STUN_PASSWD_PLAIN;
    status = pj_stun_session_set_credential(client->sess, client_auth_type, &cred);
    if (status != PJ_SUCCESS) {
	destroy_client_server();
	return -240;
    }

    /* Create the server */
    status = create_std_server(server_auth_type, server_responding);
    if (status != 0) {
	destroy_client_server();
	return status;
    }

    /* Create request */
    status = pj_stun_session_create_req(client->sess, PJ_STUN_BINDING_REQUEST, 
					PJ_STUN_MAGIC, NULL, &tdata);
    if (status != PJ_SUCCESS) {
	destroy_client_server();
	return -250;
    }
//.........这里部分代码省略.........