/* API */
PJ_DEF(pj_status_t) pjmedia_avi_dev_create_factory(
				    pj_pool_factory *pf,
				    unsigned max_dev,
				    pjmedia_vid_dev_factory **p_ret)
{
    struct avi_factory *cf;
    pj_pool_t *pool;
    pj_status_t status;

    pool = pj_pool_create(pf, "avidevfc%p", 512, 512, NULL);
    cf = PJ_POOL_ZALLOC_T(pool, struct avi_factory);
    cf->pf = pf;
    cf->pool = pool;
    cf->dev_count = max_dev;
    cf->base.op = &factory_op;

    cf->dev_info = (struct avi_dev_info*)
 		   pj_pool_calloc(cf->pool, cf->dev_count,
 				  sizeof(struct avi_dev_info));

    if (p_ret) {
	*p_ret = &cf->base;
    }

    status = pjmedia_vid_register_factory(NULL, &cf->base);
    if (status != PJ_SUCCESS)
	return status;

    PJ_LOG(4, (THIS_FILE, "AVI dev factory created with %d virtual device(s)",
	       cf->dev_count));

    return PJ_SUCCESS;
}

PJ_DEF(pj_hash_table_t*) pj_hash_create(pj_pool_t *pool, unsigned size)
{
    pj_hash_table_t *h;
    unsigned table_size;
    
    /* Check that PJ_HASH_ENTRY_BUF_SIZE is correct. */
    PJ_ASSERT_RETURN(sizeof(pj_hash_entry)<=PJ_HASH_ENTRY_BUF_SIZE, NULL);

    h = PJ_POOL_ALLOC_T(pool, pj_hash_table_t);
    h->count = 0;

    PJ_LOG( 6, ("hashtbl", "hash table %p created from pool %s", h, pj_pool_getobjname(pool)));

    /* size must be 2^n - 1.
       round-up the size to this rule, except when size is 2^n, then size
       will be round-down to 2^n-1.
     */
    table_size = 8;
    do {
	table_size <<= 1;    
    } while (table_size < size);
    table_size -= 1;
    
    h->rows = table_size;
    h->table = (pj_hash_entry**)
    	       pj_pool_calloc(pool, table_size+1, sizeof(pj_hash_entry*));
    return h;
}

/*
 * pj_thread_create(...)
 */
PJ_DEF(pj_status_t) pj_thread_create( pj_pool_t *pool, 
                                      const char *thread_name,
				      pj_thread_proc *proc, 
                                      void *arg,
				      pj_size_t stack_size, 
				      unsigned flags,
                                      pj_thread_t **thread_ptr)
{
    DWORD dwflags = 0;
    pj_thread_t *rec;

    PJ_CHECK_STACK();
    PJ_ASSERT_RETURN(pool && proc && thread_ptr, PJ_EINVAL);

    /* Set flags */
    if (flags & PJ_THREAD_SUSPENDED)
	dwflags |= CREATE_SUSPENDED;

    /* Create thread record and assign name for the thread */
    rec = (struct pj_thread_t*) pj_pool_calloc(pool, 1, sizeof(pj_thread_t));
    if (!rec)
	return PJ_ENOMEM;

    /* Set name. */
    if (!thread_name)
	thread_name = "thr%p";

    if (strchr(thread_name, '%')) {
	pj_ansi_snprintf(rec->obj_name, PJ_MAX_OBJ_NAME, thread_name, rec);
    } else {
	pj_ansi_strncpy(rec->obj_name, thread_name, PJ_MAX_OBJ_NAME);
	rec->obj_name[PJ_MAX_OBJ_NAME-1] = '\0';
    }

    PJ_LOG(6, (rec->obj_name, "Thread created"));

#if defined(PJ_OS_HAS_CHECK_STACK) && PJ_OS_HAS_CHECK_STACK!=0
    rec->stk_size = stack_size ? (pj_uint32_t)stack_size : 0xFFFFFFFFUL;
    rec->stk_max_usage = 0;
#endif

    /* Create the thread. */
    rec->proc = proc;
    rec->arg = arg;
    rec->hthread = CreateThread(NULL, stack_size, 
				thread_main, rec,
				dwflags, &rec->idthread);
    if (rec->hthread == NULL)
	return PJ_RETURN_OS_ERROR(GetLastError());

    /* Success! */
    *thread_ptr = rec;
    return PJ_SUCCESS;
}

/*
 * Notification on completion of DNS SRV resolution.
 */
static void dns_srv_resolver_cb(void *user_data,
                                pj_status_t status,
                                const pj_dns_srv_record *rec)
{
    pj_tcp_session *sess = (pj_tcp_session*) user_data;
    unsigned i, cnt, tot_cnt;

    /* Clear async resolver */
    sess->dns_async = NULL;

    /* Check failure */
    if (status != PJ_SUCCESS) {
        sess_shutdown(sess, status);
        return;
    }

    /* Calculate total number of server entries in the response */
    tot_cnt = 0;
    for (i=0; i<rec->count; ++i) {
        tot_cnt += rec->entry[i].server.addr_count;
    }

    if (tot_cnt > PJ_TCP_MAX_DNS_SRV_CNT)
        tot_cnt = PJ_TCP_MAX_DNS_SRV_CNT;

    /* Allocate server entries */
    sess->srv_addr_list = (pj_sockaddr*)
                          pj_pool_calloc(sess->pool, tot_cnt,
                                         sizeof(pj_sockaddr));

    /* Copy results to server entries */
    for (i=0, cnt=0; i<rec->count && cnt<PJ_TCP_MAX_DNS_SRV_CNT; ++i) {
        unsigned j;

        for (j=0; j<rec->entry[i].server.addr_count &&
                cnt<PJ_TCP_MAX_DNS_SRV_CNT; ++j)
        {
            pj_sockaddr_in *addr = &sess->srv_addr_list[cnt].ipv4;

            addr->sin_family = sess->af;
            addr->sin_port = pj_htons(rec->entry[i].port);
            addr->sin_addr.s_addr = rec->entry[i].server.addr[j].s_addr;

            ++cnt;
        }
    }
    sess->srv_addr_cnt = (pj_uint16_t)cnt;

    /* Set current server */
    sess->srv_addr = &sess->srv_addr_list[0];

    /* Set state to PJ_TCP_STATE_RESOLVED */
    pj_tcp_session_set_state(sess, PJ_TCP_STATE_RESOLVED);
}

PJ_DEF(pj_status_t) pj_atomic_create( pj_pool_t *pool, 
				      pj_atomic_value_t value,
				      pj_atomic_t **ptr_var)
{
    pj_atomic_t *t = pj_pool_calloc(pool, 1, sizeof(pj_atomic_t));
    if (!t) return PJ_ENOMEM;

    atomic_set(&t->atom, value);
    *ptr_var = t;
    
    return PJ_SUCCESS;
}

/* Util: initiate v4l2 streaming via mmap */
static pj_status_t vid4lin_stream_init_streaming(vid4lin_stream *stream)
{
    struct v4l2_requestbuffers req;
    unsigned i;
    pj_status_t status;

    pj_bzero(&req, sizeof(req));
    req.count = BUFFER_CNT;
    req.type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
    req.memory = V4L2_MEMORY_MMAP;
    status = xioctl(stream->fd, VIDIOC_REQBUFS, &req);
    if (status != PJ_SUCCESS)
	return status;

    stream->buffers = pj_pool_calloc(stream->pool, req.count,
				     sizeof(*stream->buffers));
    stream->buf_cnt = 0;

    for (i = 0; i < req.count; ++i) {
	struct v4l2_buffer buf;

	pj_bzero(&buf, sizeof(buf));

	buf.type        = V4L2_BUF_TYPE_VIDEO_CAPTURE;
	buf.memory      = V4L2_MEMORY_MMAP;
	buf.index       = i;

	status = xioctl(stream->fd, VIDIOC_QUERYBUF, &buf);
	if (status != PJ_SUCCESS)
	    goto on_error;

	stream->buffers[i].length = buf.length;
	stream->buffers[i].start = v4l2_mmap(NULL, buf.length,
					     PROT_READ | PROT_WRITE,
					     MAP_SHARED, stream->fd,
					     buf.m.offset);

	if (MAP_FAILED == stream->buffers[i].start) {
	    status = pj_get_os_error();
	    goto on_error;
	}

	stream->buf_cnt++;
    }

    PJ_LOG(5,(THIS_FILE, "  mmap streaming initialized"));

    stream->io_type = IO_TYPE_MMAP;
    return PJ_SUCCESS;

on_error:
    return status;
}

PJ_DEF(pj_atomic_t*) pj_atomic_create( pj_pool_t *pool, long value)
{
    pj_atomic_t *t = pj_pool_calloc(pool, 1, sizeof(pj_atomic_t));
    if (!t)
	return NULL;
#if PJ_HAS_THREADS
    t->mutex = pj_mutex_create(pool, "atm%p", PJ_MUTEX_SIMPLE);
    if (t->mutex == NULL)
	return NULL;
#endif
    t->value = value;
    return t;
}

/*
 * Duplicate SSL socket parameter.
 */
PJ_DEF(void) pj_ssl_sock_param_copy( pj_pool_t *pool, 
				     pj_ssl_sock_param *dst,
				     const pj_ssl_sock_param *src)
{
    /* Init secure socket param */
    pj_memcpy(dst, src, sizeof(*dst));
    if (src->ciphers_num > 0) {
	unsigned i;
	dst->ciphers = (pj_ssl_cipher*)
			pj_pool_calloc(pool, src->ciphers_num, 
				       sizeof(pj_ssl_cipher));
	for (i = 0; i < src->ciphers_num; ++i)
	    dst->ciphers[i] = src->ciphers[i];
    }

    if (src->curves_num > 0) {
	unsigned i;
    	dst->curves = (pj_ssl_curve *)pj_pool_calloc(pool, src->curves_num,
					   	     sizeof(pj_ssl_curve));
	for (i = 0; i < src->curves_num; ++i)
	    dst->curves[i] = src->curves[i];
    }

    if (src->server_name.slen) {
        /* Server name must be null-terminated */
        pj_strdup_with_null(pool, &dst->server_name, &src->server_name);
    }

    if (src->sigalgs.slen) {
    	/* Sigalgs name must be null-terminated */
    	pj_strdup_with_null(pool, &dst->sigalgs, &src->sigalgs);
    }

    if (src->entropy_path.slen) {
    	/* Path name must be null-terminated */
    	pj_strdup_with_null(pool, &dst->entropy_path, &src->entropy_path);
    }
}

PJ_DEF(pj_thread_t*) pj_thread_create(pj_pool_t *pool, const char *thread_name,
				      pj_thread_proc *proc, void *arg,
				      pj_size_t stack_size, void *stack, 
				      unsigned flags)
{
    DWORD dwflags = 0;
    pj_thread_t *rec;

    /* Stack argument must be NULL on WIN32 */
    if (stack != NULL)
	return NULL;

    /* Set flags */
    if (flags & PJ_THREAD_SUSPENDED)
	dwflags |= CREATE_SUSPENDED;

    /* Create thread record and assign name for the thread */
    rec = (struct pj_thread_t*) pj_pool_calloc(pool, 1, sizeof(pj_thread_t));
    if (!rec) {
	return NULL;
    }
    /* Set name. */
    if (!thread_name) {
	thread_name = "thr%p";
    }
    if (strchr(thread_name, '%')) {
	pj_snprintf(rec->obj_name, PJ_MAX_OBJ_NAME, thread_name, rec);
    } else {
	strncpy(rec->obj_name, thread_name, PJ_MAX_OBJ_NAME);
	rec->obj_name[PJ_MAX_OBJ_NAME-1] = '\0';
    }

    PJ_LOG(6, (rec->obj_name, "Thread created"));

    /* Create the thread. */
    rec->proc = proc;
    rec->arg = arg;
    rec->hthread = CreateThread(NULL, stack_size, 
				thread_main, rec,
				dwflags, &rec->idthread);
    if (rec->hthread == NULL) {
	return NULL;
    }
    return rec;
}

PJ_DEF(pj_status_t) pj_activesock_start_read(pj_activesock_t *asock,
					     pj_pool_t *pool,
					     unsigned buff_size,
					     pj_uint32_t flags)
{
    void **readbuf;
    unsigned i;

    PJ_ASSERT_RETURN(asock && pool && buff_size, PJ_EINVAL);

    readbuf = (void**) pj_pool_calloc(pool, asock->async_count, 
				      sizeof(void*));

    for (i=0; i<asock->async_count; ++i) {
	readbuf[i] = pj_pool_alloc(pool, buff_size);
    }

    return pj_activesock_start_read2(asock, pool, buff_size, readbuf, flags);
}

Jt_Station::Jt_Station(jt_station_config_t *cfg)
{
	PJ_LOG(3, ("station.cpp", "Jt_Station: build a station. id: %d, name: %s", cfg->id, cfg->name.ptr));
	_pool = pj_pool_create(&g_cp.factory, cfg->name.ptr, INIT_POOL_STATION_SIZE, INC_POOL_STATION_SIZE, NULL);
	pj_strdup_with_null(_pool, &_name, &cfg->name);
	_id  = cfg->id;
	_fin = cfg->fin;
	_rtinfo = pj_hash_create(_pool, MAX_SIGNAL_KIND_COUNTS);
	jt_neighbour_config_t *nb = (jt_neighbour_config_t*)pj_pool_calloc(_pool, cfg->link_cnts, sizeof(jt_neighbour_config_t));
	for(pj_uint32_t i=0; i<cfg->link_cnts; i++)
	{
		nb[i].event  = cfg->link_cfg[i]->event;
		nb[i].stt_id = cfg->link_cfg[i]->stt_id;
		pj_hash_set(_pool, _rtinfo, &(nb[i].event), sizeof(pj_uint32_t), 0, &(nb[i]));
	}

	_car_tbl = pj_hash_create(_pool, MAX_CAR_COUNTS_PER_STATION);
	_ops_tbl = pj_hash_create(_pool, MAX_SIGNAL_KIND_COUNTS);
}

pjsip_messaging_create_session( pjsip_endpoint *endpt, const pj_str_t *param_from,
			        const pj_str_t *param_to )
{
    pj_pool_t *pool;
    pjsip_messaging_session *ses;
    pj_str_t tmp, to;

    pool = pjsip_endpt_create_pool(endpt, "imsess", 1024, 1024);
    if (!pool)
	return NULL;

    ses = pj_pool_calloc(pool, 1, sizeof(pjsip_messaging_session));
    ses->pool = pool;
    ses->endpt = endpt;

    ses->call_id = pjsip_cid_hdr_create(pool);
    pj_create_unique_string(pool, &ses->call_id->id);

    ses->cseq = pjsip_cseq_hdr_create(pool);
    ses->cseq->cseq = pj_rand();
    ses->cseq->method = message_method;

    ses->from = pjsip_from_hdr_create(pool);
    pj_strdup_with_null(pool, &tmp, param_from);
    ses->from->uri = pjsip_parse_uri(pool, tmp.ptr, tmp.slen, PJSIP_PARSE_URI_AS_NAMEADDR);
    if (ses->from->uri == NULL) {
	pjsip_endpt_destroy_pool(endpt, pool);
	return NULL;
    }
    pj_create_unique_string(pool, &ses->from->tag);

    ses->to = pjsip_to_hdr_create(pool);
    pj_strdup_with_null(pool, &to, param_from);
    ses->to->uri = pjsip_parse_uri(pool, to.ptr, to.slen, PJSIP_PARSE_URI_AS_NAMEADDR);
    if (ses->to->uri == NULL) {
	pjsip_endpt_destroy_pool(endpt, pool);
	return NULL;
    }

    PJ_LOG(4,(THIS_FILE, "IM session created: recipient=%s", to.ptr));
    return ses;
}

/* Set local peer address */
PJ_DEF(void) pj_tcp_session_set_local_addr( pj_tcp_session *sess,
        const pj_sockaddr_t *addr)
{
    char buf[PJ_INET6_ADDRSTRLEN+20];

    if (!sess)
        return;

    if (!sess->local_addr) {
        sess->local_addr = (pj_sockaddr_t *)
                           pj_pool_calloc(sess->pool, 1,
                                          pj_sockaddr_get_len(addr));
    }
    sess->local_addr_len = pj_sockaddr_get_len(addr);
    pj_sockaddr_cp(sess->local_addr, addr);

    //pj_sockaddr_set_str_addr(pj_AF_INET(), (pj_sockaddr *)sess->peer_addr, &str_addr);

    PJ_LOG(4, (THIS_FILE, "pj_tcp_session_set_remote_peer_addr() %s",
               pj_sockaddr_print(sess->local_addr, buf, sizeof(buf), 3)));
}

pjsip_messaging_send_msg( pjsip_endpoint *endpt, pjsip_tx_data *tdata, 
			  void *token, pjsip_messaging_cb cb )
{
    pjsip_transaction *tsx;
    struct messaging_data *msg_data;

    /* Create transaction. */
    tsx = pjsip_endpt_create_tsx(endpt);
    if (!tsx) {
	pjsip_tx_data_dec_ref(tdata);
	return -1;
    }

    /* Save parameters to messaging data and attach to tsx. */
    msg_data = pj_pool_calloc(tsx->pool, 1, sizeof(struct messaging_data));
    msg_data->cb = cb;
    msg_data->token = token;

    /* Init transaction. */
    tsx->module_data[module_id] = msg_data;
    if (pjsip_tsx_init_uac(tsx, tdata) != 0) {
	pjsip_tx_data_dec_ref(tdata);
	pjsip_endpt_destroy_tsx(endpt, tsx);
	return -1;
    }

    pjsip_endpt_register_tsx(endpt, tsx);

    /* 
     * Instruct transaction to send message.
     * Further events will be received via transaction's event.
     */
    pjsip_tsx_on_tx_msg(tsx, tdata);

    /* Decrement reference counter. */
    pjsip_tx_data_dec_ref(tdata);
    return 0;
}

PJ_DEF(pj_status_t) pj_stun_create_bind_req( pj_pool_t *pool, 
					     void **msg, pj_size_t *len,
					     pj_uint32_t id_hi, 
					     pj_uint32_t id_lo)
{
    pj_stun_msg_hdr *hdr;
    
    PJ_LOG(5,(THIS_FILE, "pj_stun_create_bind_req"));

    hdr = pj_pool_calloc(pool, 1, sizeof(pj_stun_msg_hdr));
    if (!hdr) {
	PJ_LOG(5,(THIS_FILE, "Error allocating memory!"));
	return -1;
    }

    hdr->type = pj_htons(PJ_STUN_BINDING_REQUEST);
    hdr->tsx[2] = pj_htonl(id_hi);
    hdr->tsx[3] = pj_htonl(id_lo);
    *msg = hdr;
    *len = sizeof(pj_stun_msg_hdr);

    return 0;
}

PJ_DEF(pj_ioqueue_key_t*) pj_ioqueue_register( pj_pool_t *pool,
					       pj_ioqueue_t *ioque,
					       pj_oshandle_t sock,
					       void *user_data,
					       const pj_ioqueue_callback *cb)
{
    pj_ioqueue_key_t *key = NULL;
    pj_uint32_t value;
    
    pj_mutex_lock(ioque->mutex);

    if (ioque->count >= ioque->max)
	goto on_return;

    /* Set socket to nonblocking. */
    value = 1;
    if (pj_sock_ioctl((pj_sock_t)sock, PJ_FIONBIO, &value)) {
	PJ_PERROR(("ioqueue", "Error setting FIONBIO"));
	goto on_return;
    }

    /* Create key. */
    key = (pj_ioqueue_key_t*)pj_pool_calloc(pool, 1, sizeof(pj_ioqueue_key_t));
    key->fd = (pj_sock_t)sock;
    key->user_data = user_data;

    /* Save callback. */
    pj_memcpy(&key->cb, cb, sizeof(pj_ioqueue_callback));

    /* Register */
    pj_list_insert_before(&ioque->hlist, key);
    ++ioque->count;

on_return:
    pj_mutex_unlock(ioque->mutex);
    return key;
}

PJ_DEF(pj_status_t) pjstun_get_mapped_addr( pj_pool_factory *pf,
					    int sock_cnt, pj_sock_t sock[],
					    const pj_str_t *srv1, int port1,
					    const pj_str_t *srv2, int port2,
					    pj_sockaddr_in mapped_addr[])
{
    unsigned srv_cnt;
    pj_sockaddr_in srv_addr[2];
    int i, j, send_cnt = 0, nfds;
    pj_pool_t *pool;
    struct query_rec {
	struct {
	    pj_uint32_t	mapped_addr;
	    pj_uint32_t	mapped_port;
	} srv[2];
    } *rec;
    void       *out_msg;
    pj_size_t	out_msg_len;
    int wait_resp = 0;
    pj_status_t status;

    PJ_CHECK_STACK();

    TRACE_((THIS_FILE, "Entering pjstun_get_mapped_addr()"));

    /* Create pool. */
    pool = pj_pool_create(pf, "stun%p", 400, 400, NULL);
    if (!pool)
	return PJ_ENOMEM;


    /* Allocate client records */
    rec = (struct query_rec*) pj_pool_calloc(pool, sock_cnt, sizeof(*rec));
    if (!rec) {
	status = PJ_ENOMEM;
	goto on_error;
    }

    TRACE_((THIS_FILE, "  Memory allocated."));

    /* Create the outgoing BIND REQUEST message template */
    status = pjstun_create_bind_req( pool, &out_msg, &out_msg_len, 
				      pj_rand(), pj_rand());
    if (status != PJ_SUCCESS)
	goto on_error;

    TRACE_((THIS_FILE, "  Binding request created."));

    /* Resolve servers. */
    status = pj_sockaddr_in_init(&srv_addr[0], srv1, (pj_uint16_t)port1);
    if (status != PJ_SUCCESS)
		goto on_error;

    srv_cnt = 1;

    if (srv2 && port2) {
	status = pj_sockaddr_in_init(&srv_addr[1], srv2, (pj_uint16_t)port2);
	if (status != PJ_SUCCESS)
		goto on_error;

	if (srv_addr[1].sin_addr.s_addr != srv_addr[0].sin_addr.s_addr &&
	    srv_addr[1].sin_port != srv_addr[0].sin_port)
	{
	    srv_cnt++;
	}
    }

    TRACE_((THIS_FILE, "  Server initialized, using %d server(s)", srv_cnt));

    /* Init mapped addresses to zero */
    pj_memset(mapped_addr, 0, sock_cnt * sizeof(pj_sockaddr_in));

    /* We need these many responses */
    wait_resp = sock_cnt * srv_cnt;

    TRACE_((THIS_FILE, "  Done initialization."));

#if defined(PJ_SELECT_NEEDS_NFDS) && PJ_SELECT_NEEDS_NFDS!=0
    nfds = -1;
    for (i=0; i<sock_cnt; ++i) {
	if (sock[i] > nfds) {
	    nfds = sock[i];
	}
    }
#else
    nfds = FD_SETSIZE-1;
#endif

    /* Main retransmission loop. */
    for (send_cnt=0; send_cnt<MAX_REQUEST; ++send_cnt) {
	pj_time_val next_tx, now;
	pj_fd_set_t r;
	int select_rc;

	PJ_FD_ZERO(&r);

	/* Send messages to servers that has not given us response. */
	for (i=0; i<sock_cnt && status==PJ_SUCCESS; ++i) {
	    for (j=0; j<srv_cnt && status==PJ_SUCCESS; ++j) {
		pjstun_msg_hdr *msg_hdr = (pjstun_msg_hdr*) out_msg;
//.........这里部分代码省略.........

/*
 * Create a new listener on the specified port.
 */
PJ_DEF(pj_status_t) pj_turn_listener_create_udp( pj_turn_srv *srv,
					        int af,
					        const pj_str_t *bound_addr,
					        unsigned port,
						unsigned concurrency_cnt,
						unsigned flags,
						pj_turn_listener **p_listener)
{
    pj_pool_t *pool;
    struct udp_listener *udp;
    pj_ioqueue_callback ioqueue_cb;
    unsigned i;
    pj_status_t status;

    /* Create structure */
    pool = pj_pool_create(srv->core.pf, "udp%p", 1000, 1000, NULL);
    udp = PJ_POOL_ZALLOC_T(pool, struct udp_listener);
    udp->base.pool = pool;
    udp->base.obj_name = pool->obj_name;
    udp->base.server = srv;
    udp->base.tp_type = PJ_TURN_TP_UDP;
    udp->base.sock = PJ_INVALID_SOCKET;
    udp->base.destroy = &udp_destroy;
    udp->read_cnt = concurrency_cnt;
    udp->base.flags = flags;

    udp->tp.obj_name = udp->base.obj_name;
    udp->tp.info = udp->base.info;
    udp->tp.listener = &udp->base;
    udp->tp.sendto = &udp_sendto;
    udp->tp.add_ref = &udp_add_ref;
    udp->tp.dec_ref = &udp_dec_ref;

    /* Create socket */
    status = pj_sock_socket(af, pj_SOCK_DGRAM(), 0, &udp->base.sock);
    if (status != PJ_SUCCESS)
	goto on_error;

    /* Init bind address */
    status = pj_sockaddr_init(af, &udp->base.addr, bound_addr, 
			      (pj_uint16_t)port);
    if (status != PJ_SUCCESS) 
	goto on_error;
    
    /* Create info */
    pj_ansi_strcpy(udp->base.info, "UDP:");
    pj_sockaddr_print(&udp->base.addr, udp->base.info+4, 
		      sizeof(udp->base.info)-4, 3);

    /* Bind socket */
    status = pj_sock_bind(udp->base.sock, &udp->base.addr, 
			  pj_sockaddr_get_len(&udp->base.addr));
    if (status != PJ_SUCCESS)
	goto on_error;

    /* Register to ioqueue */
    pj_bzero(&ioqueue_cb, sizeof(ioqueue_cb));
    ioqueue_cb.on_read_complete = on_read_complete;
    status = pj_ioqueue_register_sock(pool, srv->core.ioqueue, udp->base.sock,
				      udp, &ioqueue_cb, &udp->key);

    /* Create op keys */
    udp->read_op = (struct read_op**)pj_pool_calloc(pool, concurrency_cnt, 
						    sizeof(struct read_op*));

    /* Create each read_op and kick off read operation */
    for (i=0; i<concurrency_cnt; ++i) {
	pj_pool_t *rpool = pj_pool_create(srv->core.pf, "rop%p", 
					  1000, 1000, NULL);

	udp->read_op[i] = PJ_POOL_ZALLOC_T(pool, struct read_op);
	udp->read_op[i]->pkt.pool = rpool;

	on_read_complete(udp->key, &udp->read_op[i]->op_key, 0);
    }

    /* Done */
    PJ_LOG(4,(udp->base.obj_name, "Listener %s created", udp->base.info));

    *p_listener = &udp->base;
    return PJ_SUCCESS;


on_error:
    udp_destroy(&udp->base);
    return status;
}

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

    /* Transport flag */
    tp->base.flag = pjsip_transport_get_flag_from_type(type);


    /* Length of addressess. */
    tp->base.addr_len = sizeof(tp->base.local_addr);

    /* Init local address. */
    status = pj_sock_getsockname(sock, &tp->base.local_addr, 
				 &tp->base.addr_len);
    if (status != PJ_SUCCESS)
	goto on_error;

    /* Init remote name. */
    if (type == PJSIP_TRANSPORT_UDP)
	tp->base.remote_name.host = pj_str("0.0.0.0");
    else
	tp->base.remote_name.host = pj_str("::0");
    tp->base.remote_name.port = 0;

    /* Init direction */
    tp->base.dir = PJSIP_TP_DIR_NONE;

    /* Set endpoint. */
    tp->base.endpt = endpt;

    /* Transport manager and timer will be initialized by tpmgr */

    /* Attach socket and assign name. */
    udp_set_socket(tp, sock, a_name);

    /* Register to ioqueue */
    status = register_to_ioqueue(tp);
    if (status != PJ_SUCCESS)
	goto on_error;

    /* Set functions. */
    tp->base.send_msg = &udp_send_msg;
    tp->base.do_shutdown = &udp_shutdown;
    tp->base.destroy = &udp_destroy;

    /* This is a permanent transport, so we initialize the ref count
     * to one so that transport manager don't destroy this transport
     * when there's no user!
     */
    pj_atomic_inc(tp->base.ref_cnt);

    /* Register to transport manager. */
    tp->base.tpmgr = pjsip_endpt_get_tpmgr(endpt);
    status = pjsip_transport_register( tp->base.tpmgr, (pjsip_transport*)tp);
    if (status != PJ_SUCCESS)
	goto on_error;


    /* Create rdata and put it in the array. */
    tp->rdata_cnt = 0;
    tp->rdata = (pjsip_rx_data**)
    		pj_pool_calloc(tp->base.pool, async_cnt, 
			       sizeof(pjsip_rx_data*));
    for (i=0; i<async_cnt; ++i) {
	pj_pool_t *rdata_pool = pjsip_endpt_create_pool(endpt, "rtd%p", 
							PJSIP_POOL_RDATA_LEN,
							PJSIP_POOL_RDATA_INC);
	if (!rdata_pool) {
	    pj_atomic_set(tp->base.ref_cnt, 0);
	    pjsip_transport_destroy(&tp->base);
	    return PJ_ENOMEM;
	}

	init_rdata(tp, i, rdata_pool, NULL);
	tp->rdata_cnt++;
    }

    /* Start reading the ioqueue. */
    status = start_async_read(tp);
    if (status != PJ_SUCCESS) {
	pjsip_transport_destroy(&tp->base);
	return status;
    }

    /* Done. */
    if (p_transport)
	*p_transport = &tp->base;
    
    PJ_LOG(4,(tp->base.obj_name, 
	      "SIP %s started, published address is %s%.*s%s:%d",
	      pjsip_transport_get_type_desc((pjsip_transport_type_e)tp->base.key.type),
	      ipv6_quoteb,
	      (int)tp->base.local_name.host.slen,
	      tp->base.local_name.host.ptr,
	      ipv6_quotee,
	      tp->base.local_name.port));

    return PJ_SUCCESS;

on_error:
    udp_destroy((pjsip_transport*)tp);
    return status;
}

    //
    // Allocate elements and zero fill the memory.
    //
    void *calloc(pj_size_t count, pj_size_t elem)
    {
	return pj_pool_calloc(p_, count, elem);
    }