rtp.c

网络MPEG4IP流媒体开发源代码

	}
}

static void process_rtcp_sr(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
	uint32_t	 ssrc;
	rtp_event	 event;
	rtcp_sr		*sr;
	source		*s;

	ssrc = ntohl(packet->r.sr.sr.ssrc);
	s = create_source(session, ssrc, FALSE);
	if (s == NULL) {
		rtp_message(LOG_WARNING, "Source 0x%08x invalid, skipping...", ssrc);
		return;
	}

	/* Mark as an active sender, if we get a sender report... */
	if (s->sender == FALSE) {
		s->sender = TRUE;
		session->sender_count++;
	}

	/* Process the SR... */
	sr = (rtcp_sr *) xmalloc(sizeof(rtcp_sr));
	sr->ssrc          = ssrc;
	sr->ntp_sec       = ntohl(packet->r.sr.sr.ntp_sec);
	sr->ntp_frac      = ntohl(packet->r.sr.sr.ntp_frac);
	sr->rtp_ts        = ntohl(packet->r.sr.sr.rtp_ts);
	sr->sender_pcount = ntohl(packet->r.sr.sr.sender_pcount);
	sr->sender_bcount = ntohl(packet->r.sr.sr.sender_bcount);

	/* Store the SR for later retrieval... */
	if (s->sr != NULL) {
		xfree(s->sr);
	}
	s->sr = sr;
	s->last_sr = *event_ts;

	/* Call the event handler... */
	if (!filter_event(session, ssrc)) {
		event.ssrc = ssrc;
		event.type = RX_SR;
		event.data = (void *) sr;
		event.ts   = event_ts;
		session->callback(session, &event);
	}

	process_report_blocks(session, packet, ssrc, packet->r.sr.rr, event_ts);

	if (((packet->common.count * 6) + 1) < (ntohs(packet->common.length) - 5)) {
		rtp_message(LOG_NOTICE, "Profile specific SR extension ignored");
	}
}

static void process_rtcp_rr(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
	uint32_t		 ssrc;
	source		*s;

	ssrc = ntohl(packet->r.rr.ssrc);
	s = create_source(session, ssrc, FALSE);
	if (s == NULL) {
		rtp_message(LOG_WARNING, "Source 0x%08x invalid, skipping...", ssrc);
		return;
	}

	process_report_blocks(session, packet, ssrc, packet->r.rr.rr, event_ts);

	if (((packet->common.count * 6) + 1) < ntohs(packet->common.length)) {
		rtp_message(LOG_INFO, "Profile specific RR extension ignored");
	}
}

static void process_rtcp_sdes(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
	int 			count = packet->common.count;
	struct rtcp_sdes_t 	*sd   = &packet->r.sdes;
	rtcp_sdes_item 		*rsp; 
	rtcp_sdes_item		*rspn;
	rtcp_sdes_item 		*end  = (rtcp_sdes_item *) ((uint32_t *)packet + packet->common.length + 1);
	source 			*s;
	rtp_event		 event;

	while (--count >= 0) {
		rsp = &sd->item[0];
		if (rsp >= end) {
			break;
		}
		sd->ssrc = ntohl(sd->ssrc);
		s = create_source(session, sd->ssrc, FALSE);
		if (s == NULL) {
			rtp_message(LOG_NOTICE, "Cannot get valid source entry for 0x%08x, skipping...", sd->ssrc);
		} else {
			for (; rsp->type; rsp = rspn ) {
				rspn = (rtcp_sdes_item *)((char*)rsp+rsp->length+2);
				if (rspn >= end) {
					rsp = rspn;
					break;
				}
				if (rtp_set_sdes(session, sd->ssrc, rsp->type, rsp->data, rsp->length)) {
					if (!filter_event(session, sd->ssrc)) {
						event.ssrc = sd->ssrc;
						event.type = RX_SDES;
						event.data = (void *) rsp;
						event.ts   = event_ts;
						session->callback(session, &event);
					}
				} else {
					rtp_message(LOG_WARNING, "Invalid sdes item for source 0x%08x, skipping...", sd->ssrc);
				}
			}
		}
		sd = (struct rtcp_sdes_t *) ((uint32_t *)sd + (((char *)rsp - (char *)sd) >> 2)+1);
	}
	if (count >= 0) {
		rtp_message(LOG_INFO, "Invalid RTCP SDES packet, some items ignored.");
	}
}

static void process_rtcp_bye(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
	int		 i;
	uint32_t	 ssrc;
	rtp_event	 event;
	source		*s;

	for (i = 0; i < packet->common.count; i++) {
		ssrc = ntohl(packet->r.bye.ssrc[i]);
		/* This is kind-of strange, since we create a source we are about to delete. */
		/* This is done to ensure that the source mentioned in the event which is    */
		/* passed to the user of the RTP library is valid, and simplify client code. */
		create_source(session, ssrc, FALSE);
		/* Call the event handler... */
		if (!filter_event(session, ssrc)) {
			event.ssrc = ssrc;
			event.type = RX_BYE;
			event.data = NULL;
			event.ts   = event_ts;
			session->callback(session, &event);
		}
		/* Mark the source as ready for deletion. Sources are not deleted immediately */
		/* since some packets may be delayed and arrive after the BYE...              */
		s = get_source(session, ssrc);
		s->got_bye = TRUE;
		check_source(s);
		session->bye_count++;
	}
}

static void process_rtcp_app(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
	uint32_t         ssrc;
	rtp_event        event;
	rtcp_app        *app;
	source          *s;
	int              data_len;

	/* Update the database for this source. */
	ssrc = ntohl(packet->r.app.ssrc);
	create_source(session, ssrc, FALSE);
	s = get_source(session, ssrc);
	if (s == NULL) {
	        /* This should only occur in the event of database malfunction. */
	        rtp_message(LOG_NOTICE, "Source 0x%08x invalid, skipping...", ssrc);
	        return;
	}
	check_source(s);

	/* Copy the entire packet, converting the header (only) into host byte order. */
	app = (rtcp_app *) xmalloc(RTP_MAX_PACKET_LEN);
	app->version        = packet->common.version;
	app->p              = packet->common.p;
	app->subtype        = packet->common.count;
	app->pt             = packet->common.pt;
	app->length         = ntohs(packet->common.length);
	app->ssrc           = ssrc;
	app->name[0]        = packet->r.app.name[0];
	app->name[1]        = packet->r.app.name[1];
	app->name[2]        = packet->r.app.name[2];
	app->name[3]        = packet->r.app.name[3];
	data_len            = (app->length - 2) * 4;
	memcpy(app->data, packet->r.app.data, data_len);

	/* Callback to the application to process the app packet... */
	if (!filter_event(session, ssrc)) {
		event.ssrc = ssrc;
		event.type = RX_APP;
		event.data = (void *) app;       /* The callback function MUST free this! */
		event.ts   = event_ts;
		session->callback(session, &event);
	}
}

void rtp_process_ctrl(struct rtp *session, uint8_t *buffer, int buflen)
{
	/* This routine processes incoming RTCP packets */
	rtp_event	 event;
	struct timeval	 event_ts;
	rtcp_t		*packet;
	int		 first;
	uint32_t	 packet_ssrc = rtp_my_ssrc(session);

	gettimeofday(&event_ts, NULL);
	if (buflen > 0) {
		if (session->encryption_enabled)
		{
			/* Decrypt the packet... */
			(session->decrypt_func)(session->encrypt_userdata, buffer, &buflen);
			buffer += 4;	/* Skip the random prefix... */
			buflen -= 4;
		}
		if (validate_rtcp(buffer, buflen)) {
			first  = TRUE;
			packet = (rtcp_t *) buffer;
			while (packet < (rtcp_t *) (buffer + buflen)) {
				switch (packet->common.pt) {
					case RTCP_SR:
						if (first && !filter_event(session, ntohl(packet->r.sr.sr.ssrc))) {
							event.ssrc  = ntohl(packet->r.sr.sr.ssrc);
							event.type  = RX_RTCP_START;
							event.data  = &buflen;
							event.ts    = &event_ts;
							packet_ssrc = event.ssrc;
							session->callback(session, &event);
						}
						process_rtcp_sr(session, packet, &event_ts);
						break;
					case RTCP_RR:
						if (first && !filter_event(session, ntohl(packet->r.rr.ssrc))) {
							event.ssrc  = ntohl(packet->r.rr.ssrc);
							event.type  = RX_RTCP_START;
							event.data  = &buflen;
							event.ts    = &event_ts;
							packet_ssrc = event.ssrc;
							session->callback(session, &event);
						}
						process_rtcp_rr(session, packet, &event_ts);
						break;
					case RTCP_SDES:
						if (first && !filter_event(session, ntohl(packet->r.sdes.ssrc))) {
							event.ssrc  = ntohl(packet->r.sdes.ssrc);
							event.type  = RX_RTCP_START;
							event.data  = &buflen;
							event.ts    = &event_ts;
							packet_ssrc = event.ssrc;
							session->callback(session, &event);
						}
						process_rtcp_sdes(session, packet, &event_ts);
						break;
					case RTCP_BYE:
						if (first && !filter_event(session, ntohl(packet->r.bye.ssrc[0]))) {
							event.ssrc  = ntohl(packet->r.bye.ssrc[0]);
							event.type  = RX_RTCP_START;
							event.data  = &buflen;
							event.ts    = &event_ts;
							packet_ssrc = event.ssrc;
							session->callback(session, &event);
						}
						process_rtcp_bye(session, packet, &event_ts);
						break;
				        case RTCP_APP:
						if (first && !filter_event(session, ntohl(packet->r.app.ssrc))) {
							event.ssrc  = ntohl(packet->r.app.ssrc);
							event.type  = RX_RTCP_START;
							event.data  = &buflen;
							event.ts    = &event_ts;
							packet_ssrc = event.ssrc;
							session->callback(session, &event);
						}
					        process_rtcp_app(session, packet, &event_ts);
						break;
					default: 
						rtp_message(LOG_WARNING, "RTCP packet with unknown type (%d) ignored.", packet->common.pt);
						break;
				}
				packet = (rtcp_t *) ((uint8_t *) packet + (4 * (ntohs(packet->common.length) + 1)));
				first  = FALSE;
			}
			if (session->avg_rtcp_size < 0) {
				/* This is the first RTCP packet we've received, set our initial estimate */
				/* of the average  packet size to be the size of this packet.             */
				session->avg_rtcp_size = buflen + RTP_LOWER_LAYER_OVERHEAD;
			} else {
				/* Update our estimate of the average RTCP packet size. The constants are */
				/* 1/16 and 15/16 (section 6.3.3 of draft-ietf-avt-rtp-new-02.txt).       */
				session->avg_rtcp_size = (0.0625 * (buflen + RTP_LOWER_LAYER_OVERHEAD)) + (0.9375 * session->avg_rtcp_size);
			}
			/* Signal that we've finished processing this packet */
			if (!filter_event(session, packet_ssrc)) {
				event.ssrc = packet_ssrc;
				event.type = RX_RTCP_FINISH;
				event.data = NULL;
				event.ts   = &event_ts;
				session->callback(session, &event);
			}
		} else {
			rtp_message(LOG_INFO, "Invalid RTCP packet discarded");
			session->invalid_rtcp_count++;
		}
	}
}

/**
 * rtp_recv:
 * @session: the session pointer (returned by rtp_init())
 * @timeout: the amount of time that rtcp_recv() is allowed to block
 * @curr_rtp_ts: the current time expressed in units of the media
 * timestamp.
 *
 * Receive RTP packets and dispatch them.
 *
 * Returns: TRUE if data received, FALSE if the timeout occurred.
 */
int rtp_recv(struct rtp *session, struct timeval *timeout, uint32_t curr_rtp_ts)
{
	check_database(session);
	udp_fd_zero();
	udp_fd_set(session->rtp_socket);
	udp_fd_set(session->rtcp_socket);
	if (udp_select(timeout) > 0) {
		if (udp_fd_isset(session->rtp_socket)) {
			rtp_recv_data(session, curr_rtp_ts);
		}
		if (udp_fd_isset(session->rtcp_socket)) {
                        uint8_t		 buffer[RTP_MAX_PACKET_LEN];
                        int		 buflen;
                        buflen = udp_recv(session->rtcp_socket, buffer, RTP_MAX_PACKET_LEN);
			rtp_process_ctrl(session, buffer, buflen);
		}
		check_database(session);
                return TRUE;
	}
	check_database(session);
        return FALSE;
}



/**
 * rtp_add_csrc:
 * @session: the session pointer (returned by rtp_init()) 
 * @csrc: Constributing SSRC identifier
 * 
 * Adds @csrc to list of contributing sources used in SDES items.
 * Used by mixers and transcoders.
 * 
 * Return value: TRUE.
 **/
int rtp_add_csrc(struct rtp *session, uint32_t csrc)
{
	/* Mark csrc as something for which we should advertise RTCP SDES items, */
	/* in addition to our own SDES.                                          */
	source	*s;

	check_database(session);
	s = get_source(session, csrc);
	if (s == NULL) {
		s = create_source(session, csrc, FALSE);
		rtp_message(LOG_INFO, "Created source 0x%08x as CSRC", csrc);
	}
	check_source(s);
	s->should_advertise_sdes = TRUE;
	session->csrc_count++;
	rtp_message(LOG_INFO, "Added CSRC 0x%08x as CSRC %d", csrc, session->csrc_count);
	return TRUE;
}

/**
 * rtp_del_csrc:
 * @session: the session pointer (returned by rtp_init()) 
 * @csrc: Constributing SSRC identifier
 * 
 * Removes @csrc from list of contributing sources used in SDES items.
 * Used by mixers and transcoders.
 * 
 * Return value: TRUE on success, FALSE if @csrc is not a valid source.
 **/
int rtp_del_csrc(struct rtp *session, uint32_t csrc)
{
	source	*s;

	check_database(session);
	s = get_source(session, csrc);
	if (s == NULL) {
		rtp_message(LOG_ERR, "Invalid source 0x%08x when deleting", csrc);
		return FALSE;
	}
	check_source(s);
	s->should_advertise_sdes = FALSE;
	session->csrc_count--;
	if (session->last_advertised_csrc >= session->csrc_count) {
                session->last_advertised_csrc = 0;
        }
	return TRUE;
}

/**
 * rtp_set_sdes:
 * @session: the session pointer (returned by rtp_init()