turn_usage.c

来自「一个开源的sip源代码」· C语言 代码 · 共 1,431 行 · 第 1/3 页

    status = client_create(sd->tu, src_addr, src_addr_len, &client);
    if (status != PJ_SUCCESS) {
	pj_stun_perror(THIS_FILE, "Error creating new TURN client", 
		       status);
	return status;
    }


    /* Hand over message to client */
    pj_mutex_lock(client->mutex);
    status = client_handle_stun_msg(client, msg, src_addr, src_addr_len);
    pj_mutex_unlock(client->mutex);

    return status;
}


/*
 * This callback is called by STUN session when it needs to send packet
 * to the network.
 */
static pj_status_t tu_sess_on_send_msg(pj_stun_session *sess,
				       const void *pkt,
				       pj_size_t pkt_size,
				       const pj_sockaddr_t *dst_addr,
				       unsigned addr_len)
{
    struct session_data *sd;

    sd = (struct session_data*) pj_stun_session_get_user_data(sess);

    if (sd->tu->type == PJ_SOCK_DGRAM) {
	return pj_stun_usage_sendto(sd->tu->usage, pkt, pkt_size, 0,
				    dst_addr, addr_len);
    } else {
	return PJ_ENOTSUP;
    }
}


/****************************************************************************/
/*
 * TURN client operations.
 */

/* Function prototypes */
static pj_status_t client_create_relay(struct turn_client *client);
static pj_status_t client_destroy_relay(struct turn_client *client);
static void	   client_on_expired(pj_timer_heap_t *th, pj_timer_entry *e);
static void	   client_on_read_complete(pj_ioqueue_key_t *key, 
					   pj_ioqueue_op_key_t *op_key, 
					   pj_ssize_t bytes_read);
static pj_status_t client_respond(struct turn_client *client, 
				  const pj_stun_msg *msg,
				  int err_code,
				  const char *err_msg,
				  const pj_sockaddr_t *dst_addr,
				  int dst_addr_len);
static struct peer* client_get_peer(struct turn_client *client,
				    const pj_sockaddr_in *peer_addr,
				    pj_uint32_t *hval);
static struct peer* client_add_peer(struct turn_client *client,
				    const pj_sockaddr_in *peer_addr,
				    pj_uint32_t hval);

static const char *get_tp_type(int type)
{
    if (type==PJ_SOCK_DGRAM)
	return "udp";
    else if (type==PJ_SOCK_STREAM)
	return "tcp";
    else
	return "???";
}


/*
 * This callback is called when incoming STUN message is received
 * in the TURN usage. This is called from by tu_on_rx_data() when
 * the packet is handed over to the client.
 */
static pj_status_t client_sess_on_rx_msg(pj_stun_session *sess,
					 const pj_uint8_t *pkt,
					 unsigned pkt_len,
					 const pj_stun_msg *msg,
					 const pj_sockaddr_t *src_addr,
					 unsigned src_addr_len)
{
    struct session_data *sd;

    PJ_UNUSED_ARG(pkt);
    PJ_UNUSED_ARG(pkt_len);

    sd = (struct session_data*) pj_stun_session_get_user_data(sess);
    pj_assert(sd->client != PJ_SUCCESS);

    return client_handle_stun_msg(sd->client, msg, src_addr, src_addr_len);
}


/*
 * This callback is called by client's STUN session to send outgoing
 * STUN packet. It's called when client calls pj_stun_session_send_msg()
 * function.
 */
static pj_status_t client_sess_on_send_msg(pj_stun_session *sess,
					   const void *pkt,
					   pj_size_t pkt_size,
					   const pj_sockaddr_t *dst_addr,
					   unsigned addr_len)
{
    struct session_data *sd;

    sd = (struct session_data*) pj_stun_session_get_user_data(sess);

    if (sd->tu->type == PJ_SOCK_DGRAM) {
	return pj_stun_usage_sendto(sd->tu->usage, pkt, pkt_size, 0,
				    dst_addr, addr_len);
    } else {
	return PJ_ENOTSUP;
    }
}


/*
 * Create a new TURN client for the specified source address.
 */
static pj_status_t client_create(struct turn_usage *tu,
				 const pj_sockaddr_t *src_addr,
				 unsigned src_addr_len,
				 struct turn_client **p_client)
{
    pj_pool_t *pool;
    struct turn_client *client;
    pj_stun_session_cb sess_cb;
    struct session_data *sd;
    pj_status_t status;

    PJ_ASSERT_RETURN(src_addr_len==sizeof(pj_sockaddr_in), PJ_EINVAL);

    pool = pj_pool_create(tu->pf, "turnc%p", 4000, 4000, NULL);
    client = PJ_POOL_ZALLOC_T(pool, struct turn_client);
    client->pool = pool;
    client->tu = tu;
    client->sock = PJ_INVALID_SOCKET;

    pj_memcpy(&client->client_src_addr, src_addr,
	      sizeof(client->client_src_addr));

    if (src_addr) {
	const pj_sockaddr_in *a4 = (const pj_sockaddr_in *)src_addr;
	pj_ansi_snprintf(client->obj_name, sizeof(client->obj_name),
			 "%s:%s:%d",
			 get_tp_type(tu->type),
			 pj_inet_ntoa(a4->sin_addr),
			 (int)pj_ntohs(a4->sin_port));
	client->obj_name[sizeof(client->obj_name)-1] = '\0';
    }

    /* Create session */
    pj_bzero(&sess_cb, sizeof(sess_cb));
    sess_cb.on_send_msg = &client_sess_on_send_msg;
    sess_cb.on_rx_request = &client_sess_on_rx_msg;
    sess_cb.on_rx_indication = &client_sess_on_rx_msg;
    status = pj_stun_session_create(tu->cfg, client->obj_name, 
				    &sess_cb, tu->use_fingerprint,
				    &client->session);
    if (status != PJ_SUCCESS) {
	pj_pool_release(pool);
	return status;
    }

    if (tu->cred)
	pj_stun_session_set_credential(client->session, tu->cred);

    sd = PJ_POOL_ZALLOC_T(pool, struct session_data);
    sd->tu = tu;
    sd->client = client;
    pj_stun_session_set_user_data(client->session, sd);

    /* Mutex */
    status = pj_mutex_create_recursive(client->pool, pool->obj_name,
				       &client->mutex);
    if (status != PJ_SUCCESS) {
	client_destroy(client, status);
	return status;
    }

    /* Create hash table */
    client->peer_htable = pj_hash_create(client->pool, MAX_PEER_PER_CLIENT);
    if (client->peer_htable == NULL) {
	client_destroy(client, status);
	return PJ_ENOMEM;
    }

    /* Init timer entry */
    client->expiry_timer.user_data = client;
    client->expiry_timer.cb = &client_on_expired;
    client->expiry_timer.id = 0;

    /* Register to hash table */
    pj_mutex_lock(tu->mutex);
    pj_hash_set(pool, tu->client_htable, src_addr, src_addr_len, 0, client);
    pj_mutex_unlock(tu->mutex);

    /* Done */
    *p_client = client;

    PJ_LOG(4,(THIS_FILE, "TURN client %s created", client->obj_name));

    return PJ_SUCCESS;
}


/*
 * Destroy TURN client.
 */
static pj_status_t client_destroy(struct turn_client *client,
				  pj_status_t reason)
{
    struct turn_usage *tu = client->tu;
    char name[PJ_MAX_OBJ_NAME];

    pj_assert(sizeof(name)==sizeof(client->obj_name));
    pj_memcpy(name, client->obj_name, sizeof(name));

    /* Kill timer if it's active */
    if (client->expiry_timer.id != 0) {
	pj_timer_heap_cancel(tu->timer_heap, &client->expiry_timer);
	client->expiry_timer.id = PJ_FALSE;
    }

    /* Destroy relay */
    client_destroy_relay(client);

    /* Unregister client from hash table */
    pj_mutex_lock(tu->mutex);
    pj_hash_set(NULL, tu->client_htable, 
		&client->client_src_addr, sizeof(client->client_src_addr), 
		0, NULL);
    pj_mutex_unlock(tu->mutex);

    /* Destroy STUN session */
    if (client->session) {
	pj_stun_session_destroy(client->session);
	client->session = NULL;
    }

    /* Mutex */
    if (client->mutex) {
	pj_mutex_destroy(client->mutex);
	client->mutex = NULL;
    }

    /* Finally destroy pool */
    if (client->pool) {
	pj_pool_t *pool = client->pool;
	client->pool = NULL;
	pj_pool_release(pool);
    }

    if (reason == PJ_SUCCESS) {
	PJ_LOG(4,(THIS_FILE, "TURN client %s destroyed", name));
    }

    return PJ_SUCCESS;
}


/*
 * This utility function is used to setup relay (with ioqueue) after
 * socket has been allocated for the TURN client.
 */
static pj_status_t client_create_relay(struct turn_client *client)
{
    pj_ioqueue_callback client_ioq_cb;
    int addrlen;
    pj_status_t status;

    /* Update address */
    addrlen = sizeof(pj_sockaddr_in);
    status = pj_sock_getsockname(client->sock, &client->alloc_addr, 
			         &addrlen);
    if (status != PJ_SUCCESS) {
	pj_sock_close(client->sock);
	client->sock = PJ_INVALID_SOCKET;
	return status;
    }

    if (client->alloc_addr.sin_addr.s_addr == 0) {
	status = pj_gethostip(&client->alloc_addr.sin_addr);
	if (status != PJ_SUCCESS) {
	    pj_sock_close(client->sock);
	    client->sock = PJ_INVALID_SOCKET;
	    return status;
	}
    }

    /* Register to ioqueue */
    pj_bzero(&client_ioq_cb, sizeof(client_ioq_cb));
    client_ioq_cb.on_read_complete = &client_on_read_complete;
    status = pj_ioqueue_register_sock(client->pool, client->tu->ioqueue, 
				      client->sock, client,
				      &client_ioq_cb, &client->key);
    if (status != PJ_SUCCESS) {
	pj_sock_close(client->sock);
	client->sock = PJ_INVALID_SOCKET;
	return status;
    }

    pj_ioqueue_op_key_init(&client->pkt_read_key, 
			   sizeof(client->pkt_read_key));
    pj_ioqueue_op_key_init(&client->pkt_write_key, 
			   sizeof(client->pkt_write_key));

    /* Trigger the first read */
    client_on_read_complete(client->key, &client->pkt_read_key, 0);

    PJ_LOG(4,(THIS_FILE, "TURN client %s: relay allocated on %s:%s:%d",
	      client->obj_name,
	      get_tp_type(client->sock_type),
	      pj_inet_ntoa(client->alloc_addr.sin_addr),
	      (int)pj_ntohs(client->alloc_addr.sin_port)));

    return PJ_SUCCESS;
}


/*
 * This utility function is used to destroy the port allocated for
 * the TURN client.
 */
static pj_status_t client_destroy_relay(struct turn_client *client)
{
    /* Close socket */
    if (client->key) {
	pj_ioqueue_unregister(client->key);
	client->key = NULL;
	client->sock = PJ_INVALID_SOCKET;
    } else if (client->sock && client->sock != PJ_INVALID_SOCKET) {
	pj_sock_close(client->sock);
	client->sock = PJ_INVALID_SOCKET;
    }

    PJ_LOG(4,(THIS_FILE, "TURN client %s: relay allocation %s:%s:%d destroyed",
	      client->obj_name,
	      get_tp_type(client->sock_type),
	      pj_inet_ntoa(client->alloc_addr.sin_addr),
	      (int)pj_ntohs(client->alloc_addr.sin_port)));
    return PJ_SUCCESS;
}


/*
 * From the source packet address, get the peer instance from hash table.
 */
static struct peer* client_get_peer(struct turn_client *client,
				    const pj_sockaddr_in *peer_addr,
				    pj_uint32_t *hval)
{
    return (struct peer*)
	pj_hash_get(client->peer_htable, peer_addr, sizeof(*peer_addr), hval);
}


/*
 * Add a peer instance to the peer hash table.
 */
static struct peer* client_add_peer(struct turn_client *client,
				    const pj_sockaddr_in *peer_addr,
				    unsigned hval)
{
    struct peer *peer;

    peer = PJ_POOL_ZALLOC_T(client->pool, struct peer);
    peer->client = client;
    pj_memcpy(&peer->addr, peer_addr, sizeof(peer->addr));

    pj_hash_set(client->pool, client->peer_htable,
		&peer->addr, sizeof(peer->addr), hval, peer);

    PJ_LOG(4,(THIS_FILE, "TURN client %s: peer %s:%s:%d added",
	      client->obj_name, get_tp_type(client->sock_type), 
	      pj_inet_ntoa(peer->addr.sin_addr),
	      (int)pj_ntohs(peer->addr.sin_port)));

    return peer;
}


/*
 * Utility to send STUN response message (normally to send error response).
 */
static pj_status_t client_respond(struct turn_client *client, 
				  const pj_stun_msg *msg,
				  int err_code,
				  const char *custom_msg,
				  const pj_sockaddr_t *dst_addr,
				  int dst_addr_len)
{
    pj_str_t err_msg;
    pj_str_t *p_err_msg = NULL;
    pj_stun_tx_data *response;
    pj_status_t status;

    if (custom_msg)
	pj_cstr(&err_msg, custom_msg), p_err_msg = &err_msg;
    
    status = pj_stun_session_create_res(client->session, msg, 
					err_code, p_err_msg, 
					&response);
    if (status == PJ_SUCCESS)
	status = pj_stun_session_send_msg(client->session, PJ_TRUE,
					  dst_addr, dst_addr_len, response);

    return status;
}


/*
 * Handle incoming initial or subsequent Allocate Request.
 * This function is called by client_handle_stun_msg() below.
 */
static pj_status_t client_handle_allocate_req(struct turn_client *client,
					      const pj_stun_msg *msg,
					      const pj_sockaddr_t *src_addr,
					      unsigned src_addr_len)
{
    const pj_stun_bandwidth_attr *a_bw;
    const pj_stun_lifetime_attr *a_lf;
    const pj_stun_req_port_props_attr *a_rpp;
    const pj_stun_req_transport_attr *a_rt;
    const pj_stun_req_ip_attr *a_rip;
    pj_stun_tx_data *response;
    pj_sockaddr_in req_addr;
    int addr_len;
    unsigned req_bw, rpp_bits;
    pj_time_val timeout;
    pj_status_t status;

    a_bw = (const pj_stun_bandwidth_attr *)
	   pj_stun_msg_find_attr(msg, PJ_STUN_ATTR_BANDWIDTH, 0);
    a_lf = (const pj_stun_lifetime_attr*)
	    pj_stun_msg_find_attr(msg, PJ_STUN_ATTR_LIFETIME, 0);
    a_rpp = (const pj_stun_req_port_props_attr*)
	    pj_stun_msg_find_attr(msg, PJ_STUN_ATTR_REQ_PORT_PROPS, 0);
    a_rt = (const pj_stun_req_transport_attr*)
	   pj_stun_msg_find_attr(msg, PJ_STUN_ATTR_REQ_TRANSPORT, 0);
    a_rip = (const pj_stun_req_ip_attr*)
	    pj_stun_msg_find_attr(msg, PJ_STUN_ATTR_REQ_IP, 0);

    /* Init requested local address */
    pj_sockaddr_in_init(&req_addr, NULL, 0);

    /* Process BANDWIDTH attribute */
    if (a_bw && a_bw->value > client->tu->max_bw_kbps) {
	client_respond(client, msg, PJ_STUN_SC_INSUFFICIENT_CAPACITY, NULL,
		       src_addr, src_addr_len);
	return PJ_SUCCESS;
    } else if (a_bw) {
	client->bw_kbps = req_bw = a_bw->value;
    } else {
	req_bw = 0;
	client->bw_kbps = client->tu->max_bw_kbps;
    }

    /* Process REQUESTED-TRANSPORT attribute */
    if (a_rt && a_rt->value != 0) {
	client_respond(client, msg, PJ_STUN_SC_UNSUPP_TRANSPORT_PROTO, NULL,
		       src_addr, src_addr_len);
	return PJ_SUCCESS;
    } else if (a_rt) {
	client->sock_type = a_rt->value ? PJ_SOCK_STREAM : PJ_SOCK_DGRAM;
    } else {
	client->sock_type = client->tu->type;;
    }