transport_udp.c

来自「基于sip协议的网络电话源码」· C语言 代码 · 共 705 行 · 第 1/2 页

	pj_ioqueue_unregister(udp->rtp_key);
	udp->rtp_key = NULL;
    } else if (udp->rtp_sock != PJ_INVALID_SOCKET) {
	pj_sock_close(udp->rtp_sock);
	udp->rtp_sock = PJ_INVALID_SOCKET;
    }

    if (udp->rtcp_key) {
	pj_ioqueue_unregister(udp->rtcp_key);
	udp->rtcp_key = NULL;
    } else if (udp->rtcp_sock != PJ_INVALID_SOCKET) {
	pj_sock_close(udp->rtcp_sock);
	udp->rtcp_sock = PJ_INVALID_SOCKET;
    }

    pj_pool_release(udp->pool);

    return PJ_SUCCESS;
}


/* Notification from ioqueue about incoming RTP packet */
static void on_rx_rtp( pj_ioqueue_key_t *key, 
                       pj_ioqueue_op_key_t *op_key, 
                       pj_ssize_t bytes_read)
{
    struct transport_udp *udp;
    pj_status_t status;

    PJ_UNUSED_ARG(op_key);

    udp = pj_ioqueue_get_user_data(key);

    do {
	void (*cb)(void*,const void*,pj_ssize_t);
	void *user_data;

	cb = udp->rtp_cb;
	user_data = udp->user_data;

	/* Simulate packet lost on RX direction */
	if (udp->rx_drop_pct) {
	    if ((pj_rand() % 100) <= (int)udp->rx_drop_pct) {
		PJ_LOG(5,(udp->base.name, 
			  "RX RTP packet dropped because of pkt lost "
			  "simulation"));
		goto read_next_packet;
	    }
	}


	if (udp->attached && cb)
	    (*cb)(user_data, udp->rtp_pkt, bytes_read);

	/* See if source address of RTP packet is different than the 
	 * configured address, and switch RTP remote address to 
	 * source packet address after several consecutive packets
	 * have been received.
	 */
	if (bytes_read>0 && 
	    (udp->options & PJMEDIA_UDP_NO_SRC_ADDR_CHECKING)==0) 
	{
	    if ((udp->rem_rtp_addr.sin_addr.s_addr != 
		 udp->rtp_src_addr.sin_addr.s_addr) ||
		(udp->rem_rtp_addr.sin_port != 
		 udp->rtp_src_addr.sin_port))
	    {
		udp->rtp_src_cnt++;

		if (udp->rtp_src_cnt >= PJMEDIA_RTP_NAT_PROBATION_CNT) {
		
		    /* Set remote RTP address to source address */
		    udp->rem_rtp_addr = udp->rtp_src_addr;

		    /* Reset counter */
		    udp->rtp_src_cnt = 0;

		    PJ_LOG(4,(udp->base.name,
			      "Remote RTP address switched to %s:%d",
			      pj_inet_ntoa(udp->rtp_src_addr.sin_addr),
			      pj_ntohs(udp->rtp_src_addr.sin_port)));

		    /* Also update remote RTCP address if actual RTCP source
		     * address is not heard yet.
		     */
		    if (udp->rtcp_src_addr.sin_addr.s_addr == 0) {
			pj_uint16_t port;

			pj_memcpy(&udp->rem_rtcp_addr, &udp->rem_rtp_addr, 
				  sizeof(pj_sockaddr_in));
			port = (pj_uint16_t)
			       (pj_ntohs(udp->rem_rtp_addr.sin_port)+1);
			udp->rem_rtcp_addr.sin_port = pj_htons(port);

			pj_memcpy(&udp->rtcp_src_addr, &udp->rem_rtcp_addr, 
				  sizeof(pj_sockaddr_in));

			PJ_LOG(4,(udp->base.name,
				  "Remote RTCP address switched to %s:%d",
				  pj_inet_ntoa(udp->rtcp_src_addr.sin_addr),
				  pj_ntohs(udp->rtcp_src_addr.sin_port)));

		    }
		}
	    }
	}

read_next_packet:
	bytes_read = sizeof(udp->rtp_pkt);
	udp->rtp_addrlen = sizeof(pj_sockaddr_in);
	status = pj_ioqueue_recvfrom(udp->rtp_key, &udp->rtp_read_op,
				     udp->rtp_pkt, &bytes_read, 0,
				     &udp->rtp_src_addr, 
				     &udp->rtp_addrlen);

	if (status != PJ_EPENDING && status != PJ_SUCCESS)
	    bytes_read = -status;

    } while (status != PJ_EPENDING);
}


/* Notification from ioqueue about incoming RTCP packet */
static void on_rx_rtcp(pj_ioqueue_key_t *key, 
                       pj_ioqueue_op_key_t *op_key, 
                       pj_ssize_t bytes_read)
{
    struct transport_udp *udp;
    pj_status_t status;

    PJ_UNUSED_ARG(op_key);

    udp = pj_ioqueue_get_user_data(key);

    do {
	void (*cb)(void*,const void*,pj_ssize_t);
	void *user_data;

	cb = udp->rtcp_cb;
	user_data = udp->user_data;

	if (udp->attached && cb)
	    (*cb)(user_data, udp->rtcp_pkt, bytes_read);

	/* Check if RTCP source address is the same as the configured
	 * remote address, and switch the address when they are
	 * different.
	 */
	if (bytes_read>0 &&
	    (udp->options & PJMEDIA_UDP_NO_SRC_ADDR_CHECKING)==0 &&
	    ((udp->rem_rtcp_addr.sin_addr.s_addr != 
	       udp->rtcp_src_addr.sin_addr.s_addr) ||
	     (udp->rem_rtcp_addr.sin_port != 
	       udp->rtcp_src_addr.sin_port)))
	{
	    pj_memcpy(&udp->rem_rtcp_addr, &udp->rtcp_src_addr,
		      sizeof(pj_sockaddr_in));
	    PJ_LOG(4,(udp->base.name,
		      "Remote RTCP address switched to %s:%d",
		      pj_inet_ntoa(udp->rtcp_src_addr.sin_addr),
		      pj_ntohs(udp->rtcp_src_addr.sin_port)));
	}

	bytes_read = sizeof(udp->rtcp_pkt);
	udp->rtcp_addr_len = sizeof(udp->rtcp_src_addr);
	status = pj_ioqueue_recvfrom(udp->rtcp_key, &udp->rtcp_read_op,
				     udp->rtcp_pkt, &bytes_read, 0,
				     &udp->rtcp_src_addr, 
				     &udp->rtcp_addr_len);
	if (status != PJ_EPENDING && status != PJ_SUCCESS)
	    bytes_read = -status;

    } while (status != PJ_EPENDING);
}


/* Called by application to initialize the transport */
static pj_status_t transport_attach(   pjmedia_transport *tp,
				       void *user_data,
				       const pj_sockaddr_t *rem_addr,
				       const pj_sockaddr_t *rem_rtcp,
				       unsigned addr_len,
				       void (*rtp_cb)(void*,
						      const void*,
						      pj_ssize_t),
				       void (*rtcp_cb)(void*,
						       const void*,
						       pj_ssize_t))
{
    struct transport_udp *udp = (struct transport_udp*) tp;
    const pj_sockaddr_in *rtcp_addr;

    /* Validate arguments */
    PJ_ASSERT_RETURN(tp && rem_addr && addr_len, PJ_EINVAL);

    /* Must not be "attached" to existing application */
    PJ_ASSERT_RETURN(!udp->attached, PJ_EINVALIDOP);

    /* "Attach" the application: */

    /* Copy remote RTP address */
    pj_memcpy(&udp->rem_rtp_addr, rem_addr, sizeof(pj_sockaddr_in));

    /* Copy remote RTP address, if one is specified. */
    rtcp_addr = rem_rtcp;
    if (rtcp_addr && rtcp_addr->sin_addr.s_addr != 0) {
	pj_memcpy(&udp->rem_rtcp_addr, rem_rtcp, sizeof(pj_sockaddr_in));

    } else {
	int rtcp_port;

	/* Otherwise guess the RTCP address from the RTP address */
	pj_memcpy(&udp->rem_rtcp_addr, rem_addr, sizeof(pj_sockaddr_in));
	rtcp_port = pj_ntohs(udp->rem_rtp_addr.sin_port) + 1;
	udp->rem_rtcp_addr.sin_port = pj_htons((pj_uint16_t)rtcp_port);
    }

    /* Save the callbacks */
    udp->rtp_cb = rtp_cb;
    udp->rtcp_cb = rtcp_cb;
    udp->user_data = user_data;

    /* Last, mark transport as attached */
    udp->attached = PJ_TRUE;

    return PJ_SUCCESS;
}


/* Called by application when it no longer needs the transport */
static void transport_detach( pjmedia_transport *tp,
			      void *user_data)
{
    struct transport_udp *udp = (struct transport_udp*) tp;

    pj_assert(tp);

    if (udp->attached) {
	/* User data is unreferenced on Release build */
	PJ_UNUSED_ARG(user_data);

	/* As additional checking, check if the same user data is specified */
	pj_assert(user_data == udp->user_data);

	/* First, mark transport as unattached */
	udp->attached = PJ_FALSE;

	/* Clear up application infos from transport */
	udp->rtp_cb = NULL;
	udp->rtcp_cb = NULL;
	udp->user_data = NULL;
    }
}


/* Called by application to send RTP packet */
static pj_status_t transport_send_rtp( pjmedia_transport *tp,
				       const void *pkt,
				       pj_size_t size)
{
    struct transport_udp *udp = (struct transport_udp*)tp;
    pj_ssize_t sent;
    unsigned id;
    struct pending_write *pw;
    pj_status_t status;

    /* Must be attached */
    PJ_ASSERT_RETURN(udp->attached, PJ_EINVALIDOP);

    /* Check that the size is supported */
    PJ_ASSERT_RETURN(size <= RTP_LEN, PJ_ETOOBIG);

    /* Simulate packet lost on TX direction */
    if (udp->tx_drop_pct) {
	if ((pj_rand() % 100) <= (int)udp->tx_drop_pct) {
	    PJ_LOG(5,(udp->base.name, 
		      "TX RTP packet dropped because of pkt lost "
		      "simulation"));
	    return PJ_SUCCESS;
	}
    }


    id = udp->rtp_write_op_id;
    pw = &udp->rtp_pending_write[id];

    /* We need to copy packet to our buffer because when the
     * operation is pending, caller might write something else
     * to the original buffer.
     */
    pj_memcpy(pw->buffer, pkt, size);

    sent = size;
    status = pj_ioqueue_sendto( udp->rtp_key, 
				&udp->rtp_pending_write[id].op_key,
				pw->buffer, &sent, 0,
				&udp->rem_rtp_addr, 
				sizeof(pj_sockaddr_in));

    udp->rtp_write_op_id = (udp->rtp_write_op_id + 1) %
			   PJ_ARRAY_SIZE(udp->rtp_pending_write);

    if (status==PJ_SUCCESS || status==PJ_EPENDING)
	return PJ_SUCCESS;

    return status;
}

/* Called by application to send RTCP packet */
static pj_status_t transport_send_rtcp(pjmedia_transport *tp,
				       const void *pkt,
				       pj_size_t size)
{
    struct transport_udp *udp = (struct transport_udp*)tp;
    pj_ssize_t sent;
    pj_status_t status;

    PJ_ASSERT_RETURN(udp->attached, PJ_EINVALIDOP);

    sent = size;
    status = pj_ioqueue_sendto( udp->rtcp_key, &udp->rtcp_write_op,
				pkt, &sent, 0,
				&udp->rem_rtcp_addr, sizeof(pj_sockaddr_in));

    if (status==PJ_SUCCESS || status==PJ_EPENDING)
	return PJ_SUCCESS;

    return status;
}


PJ_DEF(pj_status_t) pjmedia_transport_udp_simulate_lost(pjmedia_transport *tp,
							pjmedia_dir dir,
							unsigned pct_lost)
{
    struct transport_udp *udp = (struct transport_udp*)tp;

    PJ_ASSERT_RETURN(tp && 
		     (dir==PJMEDIA_DIR_ENCODING||dir==PJMEDIA_DIR_DECODING) &&
		     pct_lost <= 100, PJ_EINVAL);

    if (dir == PJMEDIA_DIR_ENCODING)
	udp->tx_drop_pct = pct_lost;
    else if (dir == PJMEDIA_DIR_DECODING)
	udp->rx_drop_pct = pct_lost;
    else
	return PJ_EINVAL;

    return PJ_SUCCESS;
}