Files
hisi_SS524V100_SDK/mpp/sample/common/rtsp_server.c
T

2312 lines
63 KiB
C

/*************************************************************************
> File Name: rtsp_demo.c
> Author: bxq
> Mail: 544177215@qq.com
> Created Time: Monday, November 23, 2015 AM12:34:09 CST
************************************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <stdint.h>
#include <poll.h>
#if defined(__linux__) || defined(linux) || defined(__gnu_linux__)
#include <time.h>
#include <sys/time.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#endif
#include "rtsp_comm.h"
#include "rtsp_server.h"
#include "rtsp_msg.h"
#include "rtp_enc.h"
#include "queue.h"
#include "stream_queue.h"
#include "utils.h"
#include "ourIPAddress.h"
//TODO LIST 20160529
//support authentication
#if defined(_WIN32) || defined(_WIN64)
#define MSG_DONTWAIT 0
#define SK_EAGAIN (WSAEWOULDBLOCK)
#define SK_EINTR (WSAEINTR)
typedef int SOCKLEN;
#endif
#if defined(__linux__) || defined(linux) || defined(__gnu_linux__)
#define SOCKET_ERROR (-1)
#define INVALID_SOCKET (-1)
#define SK_EAGAIN (EAGAIN)
#define SK_EINTR (EINTR)
#define closesocket(x) close(x)
typedef int SOCKET;
typedef socklen_t SOCKLEN;
#endif
static int sk_errno (void)
{
#ifdef __WIN32__
return WSAGetLastError();
#endif
#if defined(__linux__) || defined(linux) || defined(__gnu_linux__)
return (errno);
#endif
}
static const char *sk_strerror (int err)
{
#ifdef __WIN32__
static char serr_code_buf[24];
sprintf(serr_code_buf, "WSAE-%d", err);
return serr_code_buf;
#endif
#if defined(__linux__) || defined(linux) || defined(__gnu_linux__)
return strerror(err);
#endif
}
struct rtsp_session;
struct rtsp_client_connection;
TAILQ_HEAD(rtsp_session_queue_head, rtsp_session);
TAILQ_HEAD(rtsp_client_connection_queue_head, rtsp_client_connection);
struct rtsp_session
{
char path[64];
int vcodec_id;
int acodec_id;
union {
struct codec_data_h264 h264;
struct codec_data_h265 h265;
} vcodec_data;
union {
struct codec_data_g726 g726;
struct codec_data_aac aac;
} acodec_data;
rtp_enc vrtpe;
rtp_enc artpe;
struct stream_queue *vstreamq;
struct stream_queue *astreamq;
uint64_t video_ntptime_of_zero_ts;
uint64_t audio_ntptime_of_zero_ts;
struct rtsp_demo *demo;
struct rtsp_client_connection_queue_head connections_qhead;
TAILQ_ENTRY(rtsp_session) demo_entry;
};
struct rtp_connection
{
int is_over_tcp;
SOCKET tcp_sockfd; //if is_over_tcp=1. rtsp socket
int tcp_interleaved[2];//if is_over_tcp=1. [0] is rtp interleaved, [1] is rtcp interleaved
SOCKET udp_sockfd[2]; //if is_over_tcp=0. [0] is rtp socket, [1] is rtcp socket
uint16_t udp_localport[2]; //if is_over_tcp=0. [0] is rtp local port, [1] is rtcp local port
uint16_t udp_peerport[2]; //if is_over_tcp=0. [0] is rtp peer port, [1] is rtcp peer port
struct in_addr peer_addr; //peer ipv4 addr
int streamq_index;
uint32_t ssrc;
uint32_t rtcp_packet_count;
uint32_t rtcp_octet_count;
uint64_t rtcp_last_ts;
};
#define RTSP_CC_STATE_INIT 0
#define RTSP_CC_STATE_READY 1
#define RTSP_CC_STATE_PLAYING 2
#define RTSP_CC_STATE_RECORDING 3
struct rtsp_client_connection
{
int state; //session state
#define RTSP_CC_STATE_INIT 0
#define RTSP_CC_STATE_READY 1
#define RTSP_CC_STATE_PLAYING 2
#define RTSP_CC_STATE_RECORDING 3
SOCKET sockfd; //rtsp client socket
struct in_addr peer_addr; //peer ipv4 addr
unsigned long session_id; //session id
char reqbuf[1024];
int reqlen;
struct rtp_connection *vrtp;
struct rtp_connection *artp;
struct rtsp_demo *demo;
struct rtsp_session *session;
TAILQ_ENTRY(rtsp_client_connection) demo_entry;
TAILQ_ENTRY(rtsp_client_connection) session_entry;
};
struct rtsp_demo
{
SOCKET sockfd; //rtsp server socket 0:invalid
struct rtsp_session_queue_head sessions_qhead;
struct rtsp_client_connection_queue_head connections_qhead;
};
static struct rtsp_demo *__alloc_demo (void)
{
struct rtsp_demo *d = (struct rtsp_demo*) calloc(1, sizeof(struct rtsp_demo));
if (NULL == d) {
err("alloc memory for rtsp_demo failed\n");
return NULL;
}
TAILQ_INIT(&d->sessions_qhead);
TAILQ_INIT(&d->connections_qhead);
return d;
}
static void __free_demo (struct rtsp_demo *d)
{
if (d) {
free(d);
}
}
static struct rtsp_session *__alloc_session (struct rtsp_demo *d)
{
struct rtsp_session *s = (struct rtsp_session*) calloc(1, sizeof(struct rtsp_session));
if (NULL == s) {
err("alloc memory for rtsp_session failed\n");
return NULL;
}
s->demo = d;
TAILQ_INIT(&s->connections_qhead);
TAILQ_INSERT_TAIL(&d->sessions_qhead, s, demo_entry);
return s;
}
static void __free_session (struct rtsp_session *s)
{
if (s) {
struct rtsp_demo *d = s->demo;
TAILQ_REMOVE(&d->sessions_qhead, s, demo_entry);
free(s);
}
}
static struct rtsp_client_connection *__alloc_client_connection (struct rtsp_demo *d)
{
struct rtsp_client_connection *cc = (struct rtsp_client_connection*) calloc(1, sizeof(struct rtsp_client_connection));
if (NULL == cc) {
err("alloc memory for rtsp_session failed\n");
return NULL;
}
cc->demo = d;
TAILQ_INSERT_TAIL(&d->connections_qhead, cc, demo_entry);
return cc;
}
static void __free_client_connection (struct rtsp_client_connection *cc)
{
if (cc) {
struct rtsp_demo *d = cc->demo;
TAILQ_REMOVE(&d->connections_qhead, cc, demo_entry);
free(cc);
}
}
static void __client_connection_bind_session (struct rtsp_client_connection *cc, struct rtsp_session *s)
{
if (cc->session == NULL) {
cc->session = s;
TAILQ_INSERT_TAIL(&s->connections_qhead, cc, session_entry);
}
}
static void __client_connection_unbind_session (struct rtsp_client_connection *cc)
{
struct rtsp_session *s = cc->session;
if (s) {
TAILQ_REMOVE(&s->connections_qhead, cc, session_entry);
cc->session = NULL;
}
}
static rtsp_demo_handle rtsp_new_demo (int *_port)
{
struct rtsp_demo *d = NULL;
struct sockaddr_in inaddr;
SOCKET sockfd;
int ret;
int opt = 1;
int port = *_port;
#ifdef __WIN32__
WSADATA ws;
WSAStartup(MAKEWORD(2,2), &ws);
#endif
d = __alloc_demo();
if (NULL == d) {
return NULL;
}
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == INVALID_SOCKET) {
err("create socket failed : %s\n", sk_strerror(sk_errno()));
__free_demo(d);
return NULL;
}
ret = setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const void *)&opt, sizeof(opt));
if (ret == SOCKET_ERROR) {
warn("setsockopt SO_REUSEADDR failed: %s\n", sk_strerror(sk_errno()));
}
opt = 1;
ret = setsockopt(sockfd, SOL_SOCKET, MSG_NOSIGNAL, (const void *)&opt, sizeof(opt));
if (ret == SOCKET_ERROR) {
warn("setsockopt MSG_NOSIGNAL failed: %s\n", sk_strerror(sk_errno()));
}
if (port <= 0)
port = 554;
#if 0
{
int i = 0;
for(i=0;i<10;i++){
memset(&inaddr, 0, sizeof(inaddr));
inaddr.sin_family = AF_INET;
inaddr.sin_addr.s_addr = htonl(INADDR_ANY);
inaddr.sin_port = htons(port+i);
*_port = port+i;
ret = bind(sockfd, (struct sockaddr*)&inaddr, sizeof(inaddr));
if (ret == SOCKET_ERROR) {
continue;
}else{
break;
}
}
}
if (ret == SOCKET_ERROR) {
err("port %d bind socket to address failed : %s\n", port, sk_strerror(sk_errno()));
closesocket(sockfd);
__free_demo(d);
return NULL;
}
#else
memset(&inaddr, 0, sizeof(inaddr));
inaddr.sin_family = AF_INET;
inaddr.sin_addr.s_addr = htonl(INADDR_ANY);
inaddr.sin_port = htons(port);
*_port = port;
ret = bind(sockfd, (struct sockaddr*)&inaddr, sizeof(inaddr));
if (ret == SOCKET_ERROR) {
err("port %d bind socket to address failed : %s\n",port, sk_strerror(sk_errno()));
closesocket(sockfd);
__free_demo(d);
return NULL;
}
#endif
ret = listen(sockfd, 128); //XXX
if (ret == SOCKET_ERROR) {
err("listen socket failed : %s\n", sk_strerror(sk_errno()));
closesocket(sockfd);
__free_demo(d);
return NULL;
}
d->sockfd = sockfd;
info("rtsp server demo starting on port %d\n", port);
return (rtsp_demo_handle)d;
}
#ifdef __WIN32__
#include <mstcpip.h>
#endif
#if defined(__linux__) || defined(linux) || defined(__gnu_linux__)
#include <fcntl.h>
#include <netinet/in.h> // Add this line
#include <netinet/tcp.h>
#endif
static int rtsp_set_client_socket (SOCKET sockfd)
{
int ret;
#ifdef __WIN32__
unsigned long nonblocked = 1;
int sndbufsiz = 1024 * 1024;
int keepalive = 1;
struct tcp_keepalive alive_in, alive_out;
unsigned long alive_retlen;
struct linger ling;
ret = ioctlsocket(sockfd, FIONBIO, &nonblocked);
if (ret == SOCKET_ERROR) {
warn("ioctlsocket FIONBIO failed: %s\n", sk_strerror(sk_errno()));
}
ret = setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, (const char*)&sndbufsiz, sizeof(sndbufsiz));
if (ret == SOCKET_ERROR) {
warn("setsockopt SO_SNDBUF failed: %s\n", sk_strerror(sk_errno()));
}
ret = setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, (const char*)&keepalive, sizeof(keepalive));
if (ret == SOCKET_ERROR) {
warn("setsockopt setsockopt SO_KEEPALIVE failed: %s\n", sk_strerror(sk_errno()));
}
alive_in.onoff = TRUE;
alive_in.keepalivetime = 60000;
alive_in.keepaliveinterval = 30000;
ret = WSAIoctl(sockfd, SIO_KEEPALIVE_VALS, &alive_in, sizeof(alive_in),
&alive_out, sizeof(alive_out), &alive_retlen, NULL, NULL);
if (ret == SOCKET_ERROR) {
warn("WSAIoctl SIO_KEEPALIVE_VALS failed: %s\n", sk_strerror(sk_errno()));
}
memset(&ling, 0, sizeof(ling));
ling.l_onoff = 1;
ling.l_linger = 0;
ret = setsockopt(sockfd, SOL_SOCKET, SO_LINGER, (const char*)&ling, sizeof(ling)); //resolve too many TCP CLOSE_WAIT
if (ret == SOCKET_ERROR) {
warn("setsockopt SO_LINGER failed: %s\n", sk_strerror(sk_errno()));
}
#endif
#if defined(__linux__) || defined(linux) || defined(__gnu_linux__)
int sndbufsiz = 8192 * 1024;
int keepalive = 1;
int keepidle = 60;
int keepinterval = 3;
int keepcount = 5;
struct linger ling;
ret = fcntl(sockfd, F_GETFL, 0);
if (ret < 0) {
warn("fcntl F_GETFL failed: %s\n", strerror(errno));
} else {
ret |= O_NONBLOCK;
ret = fcntl(sockfd, F_SETFL, ret);
if (ret < 0) {
warn("fcntl F_SETFL failed: %s\n", strerror(errno));
}
}
ret = setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, (const char*)&sndbufsiz, sizeof(sndbufsiz));
if (ret == SOCKET_ERROR) {
warn("setsockopt SO_SNDBUF failed: %s\n", sk_strerror(sk_errno()));
}
ret = setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, (const char*)&keepalive, sizeof(keepalive));
if (ret == SOCKET_ERROR) {
warn("setsockopt SO_KEEPALIVE failed: %s\n", sk_strerror(sk_errno()));
}
ret = setsockopt(sockfd, SOL_TCP, TCP_KEEPIDLE, (const char*)&keepidle, sizeof(keepidle));
if (ret == SOCKET_ERROR) {
warn("setsockopt TCP_KEEPIDLE failed: %s\n", sk_strerror(sk_errno()));
}
ret = setsockopt(sockfd, SOL_TCP, TCP_KEEPINTVL, (const char*)&keepinterval, sizeof(keepinterval));
if (ret == SOCKET_ERROR) {
warn("setsockopt TCP_KEEPINTVL failed: %s\n", sk_strerror(sk_errno()));
}
ret = setsockopt(sockfd, SOL_TCP, TCP_KEEPCNT, (const char*)&keepcount, sizeof(keepcount));
if (ret == SOCKET_ERROR) {
warn("setsockopt TCP_KEEPCNT failed: %s\n", sk_strerror(sk_errno()));
}
memset(&ling, 0, sizeof(ling));
ling.l_onoff = 1;
ling.l_linger = 0;
ret = setsockopt(sockfd, SOL_SOCKET, SO_LINGER, (const char*)&ling, sizeof(ling)); //resolve too many TCP CLOSE_WAIT
if (ret == SOCKET_ERROR) {
warn("setsockopt SO_LINGER failed: %s\n", sk_strerror(sk_errno()));
}
#endif
return 0;
}
void rtsp_codec_to_char(rtsp_codec_id codecID,char *codecBuf,int buflen)
{
if(codecBuf == NULL)return;
if(codecID == RTSP_CODEC_ID_VIDEO_H264){
strncpy(codecBuf,"H264",buflen);
}else if(codecID == RTSP_CODEC_ID_VIDEO_H265){
strncpy(codecBuf,"H265",buflen);
}else if(codecID == RTSP_CODEC_ID_VIDEO_MPEG4){
strncpy(codecBuf,"MPEG4",buflen);
}else if(codecID == RTSP_CODEC_ID_AUDIO_G711A){
strncpy(codecBuf,"G711A",buflen);
}else if(codecID == RTSP_CODEC_ID_AUDIO_G711A){
strncpy(codecBuf,"G711A",buflen);
}else if(codecID == RTSP_CODEC_ID_AUDIO_G711U){
strncpy(codecBuf,"G711U",buflen);
}else if(codecID == RTSP_CODEC_ID_AUDIO_G726){
strncpy(codecBuf,"G726",buflen);
}else if(codecID == RTSP_CODEC_ID_AUDIO_AAC){
strncpy(codecBuf,"AAC",buflen);
}else{
strncpy(codecBuf,"null",buflen);
}
}
static struct rtsp_client_connection *rtsp_new_client_connection (rtspClientConnect connectCB, void *user_data, struct rtsp_demo *d)
{
struct rtsp_client_connection *cc = NULL;
struct sockaddr_in inaddr;
SOCKET sockfd;
SOCKLEN addrlen = sizeof(inaddr);
sockfd = accept(d->sockfd, (struct sockaddr*)&inaddr, &addrlen);
if (sockfd == INVALID_SOCKET) {
err("accept failed : %s\n", sk_strerror(sk_errno()));
return NULL;
}
rtsp_set_client_socket(sockfd);//XXX DEBUG
info("new rtsp client %s:%u comming\n",
inet_ntoa(inaddr.sin_addr), ntohs(inaddr.sin_port));
if(connectCB != NULL){
connectCB(inet_ntoa(inaddr.sin_addr),ntohs(inaddr.sin_port), user_data);
}
cc = __alloc_client_connection(d);
if (cc == NULL) {
closesocket(sockfd);
return NULL;
}
cc->state = RTSP_CC_STATE_INIT;
cc->sockfd = sockfd;
cc->peer_addr = inaddr.sin_addr;
return cc;
}
static void rtsp_del_rtp_connection(struct rtsp_client_connection *cc, int isaudio);
static void rtsp_del_client_connection (rtspClientDisconnect disconnectCB, void *user_data, struct rtsp_client_connection *cc)
{
if (cc) {
info("delete client %d from %s\n", cc->sockfd, inet_ntoa(cc->peer_addr));
if(disconnectCB != NULL){
disconnectCB(inet_ntoa(cc->peer_addr), user_data);
}
__client_connection_unbind_session(cc);
rtsp_del_rtp_connection(cc, 0);
rtsp_del_rtp_connection(cc, 1);
closesocket(cc->sockfd);
__free_client_connection(cc);
}
}
static int rtsp_path_match (const char *main_path, const char *full_path)
{
char path0[64] = {0};
char path1[64] = {0};
strncpy(path0, main_path, sizeof(path0) - 2);
strncpy(path1, full_path, sizeof(path1) - 2);
if (path0[strlen(path0) - 1] != '/')
strcat(path0, "/");
if (path1[strlen(path1) - 1] != '/')
strcat(path1, "/");
if (strncmp(path0, path1, strlen(path0)))
return 0;
return 1;
}
static rtsp_session_handle rtsp_new_session (rtsp_demo_handle demo, const char *path)
{
struct rtsp_demo *d = (struct rtsp_demo*)demo;
struct rtsp_session *s = NULL;
if (!d || !path || strlen(path) == 0) {
err("param invalid\n");
goto fail;
}
TAILQ_FOREACH(s, &d->sessions_qhead, demo_entry) {
if (rtsp_path_match(s->path, path) || rtsp_path_match(path, s->path)) {
err("path:%s (%s) is exist!!!\n", s->path, path);
goto fail;
}
}
s = __alloc_session(d);
if (NULL == s) {
goto fail;
}
memset(s->path, 0, sizeof(s->path));
strncpy(s->path, path, sizeof(s->path) - 1);
s->vcodec_id = RTSP_CODEC_ID_NONE;
s->acodec_id = RTSP_CODEC_ID_NONE;
//dbg("add session path: %s\n", s->path);
return (rtsp_session_handle)s;
fail:
if (s) {
free(s);
}
return NULL;
}
static rtsp_demo_handle create_rtsp_demo(int *port)
{
return rtsp_new_demo(port);
}
#define RTP_MAX_PKTSIZ ((1500-42)/4*4)
#define VRTP_MAX_NBPKTS (1000)
#define ARTP_MAX_NBPKTS (100)
#define VRTP_PT_ID (96)
#define ARTP_PT_ID (97)
#define VRTSP_SUBPATH "track1"
#define ARTSP_SUBPATH "track2"
static int rtsp_set_video (rtsp_session_handle session, int codec_id, const uint8_t *codec_data, int data_len)
{
struct rtsp_session *s = (struct rtsp_session*)session;
if (!s || (s->vcodec_id != RTSP_CODEC_ID_NONE && s->vcodec_id != codec_id))
return -1;
switch (codec_id) {
case RTSP_CODEC_ID_VIDEO_H264:
case RTSP_CODEC_ID_VIDEO_H265:
break;
default:
err("not supported codec_id %d for video\n", codec_id);
return -1;
}
s->vcodec_id = codec_id;
s->vrtpe.pt = VRTP_PT_ID;
s->vrtpe.seq = 0;
s->vrtpe.ssrc = 0;
s->vrtpe.sample_rate = 90000;
memset(&s->vcodec_data, 0, sizeof(s->vcodec_data));
if (codec_data && data_len > 0) {
switch (codec_id) {
case RTSP_CODEC_ID_VIDEO_H264:
if (rtsp_codec_data_parse_from_user_h264(codec_data, data_len, &s->vcodec_data.h264) <= 0) {
warn("parse codec_data failed\n");
break;
}
break;
case RTSP_CODEC_ID_VIDEO_H265:
if (rtsp_codec_data_parse_from_user_h265(codec_data, data_len, &s->vcodec_data.h265) <= 0) {
warn("parse codec_data failed\n");
break;
}
break;
}
}
if (!s->vstreamq) {
s->vstreamq = streamq_alloc(RTP_MAX_PKTSIZ, VRTP_MAX_NBPKTS * 2 + 1);
if (!s->vstreamq) {
err("alloc memory for video rtp queue failed\n");
s->vcodec_id = RTSP_CODEC_ID_NONE;
return -1;
}
}
return 0;
}
static int rtsp_set_audio (rtsp_session_handle session, int codec_id, const uint8_t *codec_data, int data_len)
{
struct rtsp_session *s = (struct rtsp_session*)session;
if (!s || (s->acodec_id != RTSP_CODEC_ID_NONE && s->acodec_id != codec_id))
return -1;
switch (codec_id) {
case RTSP_CODEC_ID_AUDIO_G711A:
case RTSP_CODEC_ID_AUDIO_G711U:
case RTSP_CODEC_ID_AUDIO_G726:
case RTSP_CODEC_ID_AUDIO_AAC:
break;
default:
err("not supported codec_id %d for audio\n", codec_id);
return -1;
}
s->acodec_id = codec_id;
s->artpe.pt = ARTP_PT_ID;
s->artpe.seq = 0;
s->artpe.ssrc = 0;
s->artpe.sample_rate = 8000;
memset(&s->acodec_data, 0, sizeof(s->acodec_data));
if (codec_data && data_len > 0) {
switch (codec_id) {
case RTSP_CODEC_ID_AUDIO_G726:
if (rtsp_codec_data_parse_from_user_g726(codec_data, data_len, &s->acodec_data.g726) <= 0) {
warn("parse codec_data failed\n");
break;
}
break;
case RTSP_CODEC_ID_AUDIO_AAC:
if (rtsp_codec_data_parse_from_user_aac(codec_data, data_len, &s->acodec_data.aac) <= 0) {
warn("parse codec_data failed\n");
break;
}
s->artpe.sample_rate = s->acodec_data.aac.sample_rate;
break;
}
}
if (!s->astreamq) {
s->astreamq = streamq_alloc(RTP_MAX_PKTSIZ, ARTP_MAX_NBPKTS * 2 + 1);
if (!s->astreamq) {
err("alloc memory for audio rtp queue failed\n");
s->acodec_id = RTSP_CODEC_ID_NONE;
return -1;
}
}
return 0;
}
static rtsp_session_handle create_rtsp_session(rtsp_demo_handle demo,rtsp_codec_id codecID, const char *path)
{
rtsp_session_handle session;
session = rtsp_new_session(demo, path);
rtsp_set_video(session, codecID, NULL, 0);
//rtsp_set_audio(session, RTSP_CODEC_ID_AUDIO_G711A, NULL, 0);
return session;
}
static void rtsp_del_session (rtspClientDisconnect disconnectCB, void *userData, rtsp_session_handle session)
{
struct rtsp_session *s = (struct rtsp_session*)session;
if (s) {
struct rtsp_client_connection *cc;
while ((cc = TAILQ_FIRST(&s->connections_qhead))) {
rtsp_del_client_connection(disconnectCB, userData, cc);
}
//dbg("del session path: %s\n", s->path);
if (s->vstreamq)
streamq_free(s->vstreamq);
if (s->astreamq)
streamq_free(s->astreamq);
__free_session(s);
}
}
static void rtsp_del_demo (rtspClientDisconnect disconnectCB, void *userData, rtsp_demo_handle demo)
{
struct rtsp_demo *d = (struct rtsp_demo*)demo;
if (d) {
struct rtsp_session *s;
struct rtsp_client_connection *cc;
while ((cc = TAILQ_FIRST(&d->connections_qhead))) {
rtsp_del_client_connection(disconnectCB, userData, cc);
}
while ((s = TAILQ_FIRST(&d->sessions_qhead))) {
rtsp_del_session(disconnectCB, userData, s);
}
closesocket(d->sockfd);
__free_demo(d);
}
}
static int build_simple_sdp (struct rtsp_session *s, const char *uri, char *sdpbuf, int maxlen)
{
char *p = sdpbuf;
p += sprintf(p, "v=0\r\n");
p += sprintf(p, "o=- 0 0 IN IP4 0.0.0.0\r\n");
p += sprintf(p, "s=rtsp_demo\r\n");
p += sprintf(p, "t=0 0\r\n");
p += sprintf(p, "a=control:%s\r\n", uri ? uri : "*");
p += sprintf(p, "a=range:npt=0-\r\n");
if (s->vcodec_id != RTSP_CODEC_ID_NONE) {
switch (s->vcodec_id) {
case RTSP_CODEC_ID_VIDEO_H264:
p += rtsp_build_sdp_media_attr_h264(VRTP_PT_ID, s->vrtpe.sample_rate, &s->vcodec_data.h264, p, maxlen - (p - sdpbuf));
break;
case RTSP_CODEC_ID_VIDEO_H265:
p += rtsp_build_sdp_media_attr_h265(VRTP_PT_ID, s->vrtpe.sample_rate, &s->vcodec_data.h265, p, maxlen - (p - sdpbuf));
break;
}
if (uri)
p += sprintf(p, "a=control:%s/%s\r\n", uri, VRTSP_SUBPATH);
else
p += sprintf(p, "a=control:%s\r\n", VRTSP_SUBPATH);
}
if (s->acodec_id != RTSP_CODEC_ID_NONE) {
switch (s->acodec_id) {
case RTSP_CODEC_ID_AUDIO_G711A:
p += rtsp_build_sdp_media_attr_g711a(ARTP_PT_ID, s->artpe.sample_rate, p, maxlen - (p - sdpbuf));
break;
case RTSP_CODEC_ID_AUDIO_G711U:
p += rtsp_build_sdp_media_attr_g711u(ARTP_PT_ID, s->artpe.sample_rate, p, maxlen - (p - sdpbuf));
break;
case RTSP_CODEC_ID_AUDIO_G726:
p += rtsp_build_sdp_media_attr_g726(ARTP_PT_ID, s->artpe.sample_rate, &s->acodec_data.g726, p, maxlen - (p - sdpbuf));
break;
case RTSP_CODEC_ID_AUDIO_AAC:
p += rtsp_build_sdp_media_attr_aac(ARTP_PT_ID, s->artpe.sample_rate, &s->acodec_data.aac, p, maxlen - (p - sdpbuf));
break;
}
if (uri)
p += sprintf(p, "a=control:%s/%s\r\n", uri, ARTSP_SUBPATH);
else
p += sprintf(p, "a=control:%s\r\n", ARTSP_SUBPATH);
}
return (p - sdpbuf);
}
static int rtsp_handle_OPTIONS(struct rtsp_client_connection *cc, const rtsp_msg_s *reqmsg, rtsp_msg_s *resmsg)
{
// struct rtsp_demo *d = cc->demo;
// struct rtsp_session *s = cc->session;
uint32_t public_ = 0;
dbg("\n");
public_ |= RTSP_MSG_PUBLIC_OPTIONS;
public_ |= RTSP_MSG_PUBLIC_DESCRIBE;
public_ |= RTSP_MSG_PUBLIC_SETUP;
public_ |= RTSP_MSG_PUBLIC_PAUSE;
public_ |= RTSP_MSG_PUBLIC_PLAY;
public_ |= RTSP_MSG_PUBLIC_TEARDOWN;
rtsp_msg_set_public(resmsg, public_);
return 0;
}
static int rtsp_handle_DESCRIBE(struct rtsp_client_connection *cc, const rtsp_msg_s *reqmsg, rtsp_msg_s *resmsg)
{
// struct rtsp_demo *d = cc->demo;
struct rtsp_session *s = cc->session;
char sdpbuf[1024] = "";
int sdplen = 0;
uint32_t accept = 0;
const rtsp_msg_uri_s *puri = &reqmsg->hdrs.startline.reqline.uri;
char uri[128] = "";
dbg("\n");
if (rtsp_msg_get_accept(reqmsg, &accept) < 0 && !(accept & RTSP_MSG_ACCEPT_SDP)) {
rtsp_msg_set_response(resmsg, 406);
warn("client not support accept SDP\n");
return 0;
}
//build uri
if (puri->scheme == RTSP_MSG_URI_SCHEME_RTSPU)
strcat(uri, "rtspu://");
else
strcat(uri, "rtsp://");
strcat(uri, puri->ipaddr);
if (puri->port != 0)
sprintf(uri + strlen(uri), ":%u", puri->port);
strcat(uri, s->path);
sdplen = build_simple_sdp(s, uri, sdpbuf, sizeof(sdpbuf));
rtsp_msg_set_content_type(resmsg, RTSP_MSG_CONTENT_TYPE_SDP);
rtsp_msg_set_content_length(resmsg, sdplen);
resmsg->body.body = rtsp_mem_dup(sdpbuf, sdplen);
return 0;
}
static unsigned long __rtp_gen_ssrc(void)
{
static unsigned long ssrc = 0x22345678;
return ssrc++;
}
static int __rtp_udp_local_setup (struct rtp_connection *rtp)
{
int i, ret;
for (i = 65536/4*3/2*2; i < 65536; i += 2) {
SOCKET rtpsock, rtcpsock;
struct sockaddr_in inaddr;
uint16_t port;
rtpsock = socket(AF_INET, SOCK_DGRAM, 0);
if (rtpsock == INVALID_SOCKET) {
err("create rtp socket failed: %s\n", sk_strerror(sk_errno()));
return -1;
}
rtcpsock = socket(AF_INET, SOCK_DGRAM, 0);
if (rtcpsock == INVALID_SOCKET) {
err("create rtcp socket failed: %s\n", sk_strerror(sk_errno()));
closesocket(rtpsock);
return -1;
}
port = i;
memset(&inaddr, 0, sizeof(inaddr));
inaddr.sin_family = AF_INET;
inaddr.sin_addr.s_addr = htonl(INADDR_ANY);
inaddr.sin_port = htons(port);
ret = bind(rtpsock, (struct sockaddr*)&inaddr, sizeof(inaddr));
if (ret == SOCKET_ERROR) {
closesocket(rtpsock);
closesocket(rtcpsock);
continue;
}
port = i + 1;
memset(&inaddr, 0, sizeof(inaddr));
inaddr.sin_family = AF_INET;
inaddr.sin_addr.s_addr = htonl(INADDR_ANY);
inaddr.sin_port = htons(port);
ret = bind(rtcpsock, (struct sockaddr*)&inaddr, sizeof(inaddr));
if (ret == SOCKET_ERROR) {
closesocket(rtpsock);
closesocket(rtcpsock);
continue;
}
#ifdef __WIN32__
{
unsigned long nonblocked = 1;
ret = ioctlsocket(rtpsock, FIONBIO, &nonblocked);
if (ret == SOCKET_ERROR) {
warn("ioctlsocket FIONBIO failed: %s\n", sk_strerror(sk_errno()));
}
ret = ioctlsocket(rtcpsock, FIONBIO, &nonblocked);
if (ret == SOCKET_ERROR) {
warn("ioctlsocket FIONBIO failed: %s\n", sk_strerror(sk_errno()));
}
}
#endif
#ifdef __LINUX__
ret = fcntl(rtpsock, F_GETFL, 0);
if (ret < 0) {
warn("fcntl F_GETFL failed: %s\n", strerror(errno));
} else {
ret |= O_NONBLOCK;
ret = fcntl(rtpsock, F_SETFL, ret);
if (ret < 0) {
warn("fcntl F_SETFL failed: %s\n", strerror(errno));
}
}
ret = fcntl(rtcpsock, F_GETFL, 0);
if (ret < 0) {
warn("fcntl F_GETFL failed: %s\n", strerror(errno));
} else {
ret |= O_NONBLOCK;
ret = fcntl(rtcpsock, F_SETFL, ret);
if (ret < 0) {
warn("fcntl F_SETFL failed: %s\n", strerror(errno));
}
}
#endif
rtp->is_over_tcp = 0;
rtp->udp_sockfd[0] = rtpsock;
rtp->udp_sockfd[1] = rtcpsock;
rtp->udp_localport[0] = i;
rtp->udp_localport[1] = i + 1;
return 0;
}
err("not found free local port for rtp/rtcp\n");
return -1;
}
static int rtsp_new_rtp_connection(struct rtsp_client_connection *cc, int isaudio, int istcp, int peer_port, int peer_interleaved)
{
struct rtp_connection *rtp;
struct in_addr peer_addr = cc->peer_addr;
rtp = (struct rtp_connection*) calloc(1, sizeof(struct rtp_connection));
if (rtp == NULL) {
err("alloc mem for rtp session failed: %s\n", strerror(errno));
return -1;
}
rtp->is_over_tcp = !!istcp;
rtp->peer_addr = peer_addr;
rtp->ssrc = __rtp_gen_ssrc();
if (istcp) {
rtp->tcp_sockfd = cc->sockfd;
rtp->tcp_interleaved[0] = peer_interleaved;
rtp->tcp_interleaved[1] = peer_interleaved + 1;
info("new rtp over tcp for %s ssrc:%08x peer_addr:%s interleaved:%u-%u\n",
(isaudio ? "audio" : "video"),
rtp->ssrc,
inet_ntoa(peer_addr),
rtp->tcp_interleaved[0], rtp->tcp_interleaved[1]);
} else {
if (__rtp_udp_local_setup(rtp) < 0) {
free(rtp);
return -1;
}
rtp->udp_peerport[0] = peer_port;
rtp->udp_peerport[1] = peer_port + 1;
info("new rtp over udp for %s ssrc:%08x local_port:%u-%u peer_addr:%s peer_port:%u-%u\n",
(isaudio ? "audio" : "video"),
rtp->ssrc,
rtp->udp_localport[0], rtp->udp_localport[1],
inet_ntoa(peer_addr),
rtp->udp_peerport[0], rtp->udp_peerport[1]);
}
if (isaudio) {
cc->artp = rtp;
} else {
cc->vrtp = rtp;
}
return 0;
}
static void rtsp_del_rtp_connection(struct rtsp_client_connection *cc, int isaudio)
{
struct rtp_connection *rtp;
if (isaudio) {
rtp = cc->artp;
cc->artp = NULL;
} else {
rtp = cc->vrtp;
cc->vrtp = NULL;
}
if (rtp) {
if (!(rtp->is_over_tcp)) {
closesocket(rtp->udp_sockfd[0]);
closesocket(rtp->udp_sockfd[1]);
}
free(rtp);
}
}
static int rtsp_handle_SETUP(struct rtsp_client_connection *cc, const rtsp_msg_s *reqmsg, rtsp_msg_s *resmsg)
{
// struct rtsp_demo *d = cc->demo;
struct rtsp_session *s = cc->session;
struct rtp_connection *rtp = NULL;
int istcp = 0, isaudio = 0;
char vpath[128] = "";
char apath[128] = "";
int ret;
dbg("\n");
if (!reqmsg->hdrs.transport) {
rtsp_msg_set_response(resmsg, 461);
warn("rtsp no transport err\n");
return 0;
}
if (reqmsg->hdrs.transport->type == RTSP_MSG_TRANSPORT_TYPE_RTP_AVP_TCP) {
istcp = 1;
if (!(reqmsg->hdrs.transport->flags & RTSP_MSG_TRANSPORT_FLAG_INTERLEAVED)) {
warn("rtsp no interleaved err\n");
rtsp_msg_set_response(resmsg, 461);
return 0;
}
} else {
if (!(reqmsg->hdrs.transport->flags & RTSP_MSG_TRANSPORT_FLAG_CLIENT_PORT)) {
warn("rtsp no client_port err\n");
rtsp_msg_set_response(resmsg, 461);
return 0;
}
}
snprintf(vpath, sizeof(vpath) - 1, "%s/%s", s->path, VRTSP_SUBPATH);
snprintf(apath, sizeof(vpath) - 1, "%s/%s", s->path, ARTSP_SUBPATH);
if (s->vcodec_id != RTSP_CODEC_ID_NONE && rtsp_path_match(vpath, reqmsg->hdrs.startline.reqline.uri.abspath)) {
isaudio = 0;
} else if (s->acodec_id != RTSP_CODEC_ID_NONE && rtsp_path_match(apath, reqmsg->hdrs.startline.reqline.uri.abspath)) {
isaudio = 1;
} else {
warn("rtsp urlpath:%s err\n", reqmsg->hdrs.startline.reqline.uri.abspath);
rtsp_msg_set_response(resmsg, 461);
return 0;
}
rtsp_del_rtp_connection(cc, isaudio);
ret = rtsp_new_rtp_connection(cc, isaudio, istcp, reqmsg->hdrs.transport->client_port, reqmsg->hdrs.transport->interleaved);
if (ret < 0) {
rtsp_msg_set_response(resmsg, 500);
return 0;
}
rtp = isaudio ? cc->artp : cc->vrtp;
if (istcp) {
rtsp_msg_set_transport_tcp(resmsg, rtp->ssrc, rtp->tcp_interleaved[0]);
} else {
rtsp_msg_set_transport_udp(resmsg, rtp->ssrc, rtp->udp_peerport[0], rtp->udp_localport[0]);
}
if (cc->state == RTSP_CC_STATE_PLAYING) {
rtp->streamq_index = streamq_tail((isaudio ? s->astreamq : s->vstreamq));
}
if (cc->state == RTSP_CC_STATE_INIT) {
cc->state = RTSP_CC_STATE_READY;
cc->session_id = rtsp_msg_gen_session_id();
rtsp_msg_set_session(resmsg, cc->session_id);
}
return 0;
}
static int rtsp_handle_PAUSE(struct rtsp_client_connection *cc, const rtsp_msg_s *reqmsg, rtsp_msg_s *resmsg)
{
// struct rtsp_demo *d = cc->demo;
// struct rtsp_session *s = cc->session;
dbg("\n");
if (cc->state != RTSP_CC_STATE_READY && cc->state != RTSP_CC_STATE_PLAYING)
{
rtsp_msg_set_response(resmsg, 455);
warn("rtsp status err\n");
return 0;
}
if (cc->state != RTSP_CC_STATE_READY)
cc->state = RTSP_CC_STATE_READY;
return 0;
}
static int rtsp_handle_PLAY(struct rtsp_client_connection *cc, const rtsp_msg_s *reqmsg, rtsp_msg_s *resmsg)
{
// struct rtsp_demo *d = cc->demo;
struct rtsp_session *s = cc->session;
dbg("\n");
if (cc->state != RTSP_CC_STATE_READY && cc->state != RTSP_CC_STATE_PLAYING)
{
rtsp_msg_set_response(resmsg, 455);
warn("rtsp status err\n");
return 0;
}
if (cc->state != RTSP_CC_STATE_PLAYING) {
if (cc->vrtp && s->vstreamq)
cc->vrtp->streamq_index = streamq_tail(s->vstreamq);
if (cc->artp && s->astreamq)
cc->artp->streamq_index = streamq_tail(s->astreamq);
cc->state = RTSP_CC_STATE_PLAYING;
}
return 0;
}
static int rtsp_handle_TEARDOWN(struct rtsp_client_connection *cc, const rtsp_msg_s *reqmsg, rtsp_msg_s *resmsg)
{
// struct rtsp_demo *d = cc->demo;
struct rtsp_session *s = cc->session;
char vpath[128] = "";
char apath[128] = "";
dbg("\n");
snprintf(vpath, sizeof(vpath) - 1, "%s/%s", s->path, VRTSP_SUBPATH);
snprintf(apath, sizeof(vpath) - 1, "%s/%s", s->path, ARTSP_SUBPATH);
if (rtsp_path_match(vpath, reqmsg->hdrs.startline.reqline.uri.abspath)) {
rtsp_del_rtp_connection(cc, 0);
} else if (rtsp_path_match(apath, reqmsg->hdrs.startline.reqline.uri.abspath)) {
rtsp_del_rtp_connection(cc, 1);
} else {
rtsp_del_rtp_connection(cc, 0);
rtsp_del_rtp_connection(cc, 1);
}
if (!cc->vrtp && !cc->artp) {
cc->state = RTSP_CC_STATE_INIT;
}
return 0;
}
/*
rtsp 锟斤拷锟斤拷 rtsp锟斤拷锟斤拷锟斤拷
*/
static int rtsp_process_request(struct rtsp_client_connection *cc, const rtsp_msg_s *reqmsg, rtsp_msg_s *resmsg)
{
struct rtsp_demo *d = cc->demo;
struct rtsp_session *s = cc->session;
const char *path = reqmsg->hdrs.startline.reqline.uri.abspath;
uint32_t cseq = 0, session = 0;
rtsp_msg_set_response(resmsg, 200);
rtsp_msg_set_date(resmsg, NULL);
rtsp_msg_set_server(resmsg, "rtsp_demo");
if (rtsp_msg_get_cseq(reqmsg, &cseq) < 0) {
rtsp_msg_set_response(resmsg, 400);
warn("No CSeq field\n");
return 0;
}
rtsp_msg_set_cseq(resmsg, cseq);
if (cc->state != RTSP_CC_STATE_INIT) {
if (rtsp_msg_get_session(reqmsg, &session) < 0 || session != cc->session_id) {
warn("Invalid Session field\n");
rtsp_msg_set_response(resmsg, 454);
return 0;
}
rtsp_msg_set_session(resmsg, session);
}
if (s) {
if (rtsp_path_match(s->path, path) == 0) { // /live/chn0
warn("path is not matched %s (old:%s)\n", path, s->path);
rtsp_msg_set_response(resmsg, 451);
return 0;
}
} else if (reqmsg->hdrs.startline.reqline.method != RTSP_MSG_METHOD_OPTIONS) {
TAILQ_FOREACH(s, &d->sessions_qhead, demo_entry) {
if (rtsp_path_match(s->path, path)) {
break;
}
}
if (NULL == s) {
warn("Not found session path: %s\n", path);
rtsp_msg_set_response(resmsg, 454);
return 0;
}
__client_connection_bind_session(cc, s);
}
switch (reqmsg->hdrs.startline.reqline.method) {
case RTSP_MSG_METHOD_OPTIONS:
return rtsp_handle_OPTIONS(cc, reqmsg, resmsg);
case RTSP_MSG_METHOD_DESCRIBE:
return rtsp_handle_DESCRIBE(cc, reqmsg, resmsg);
case RTSP_MSG_METHOD_SETUP:
return rtsp_handle_SETUP(cc, reqmsg, resmsg);
case RTSP_MSG_METHOD_PAUSE:
return rtsp_handle_PAUSE(cc, reqmsg, resmsg);
case RTSP_MSG_METHOD_PLAY:
return rtsp_handle_PLAY(cc, reqmsg, resmsg);
case RTSP_MSG_METHOD_TEARDOWN:
return rtsp_handle_TEARDOWN(cc, reqmsg, resmsg);
default:
break;
}
rtsp_msg_set_response(resmsg, 501);
return 0;
}
static int rtsp_recv_msg(struct rtsp_client_connection *cc, rtsp_msg_s *msg)
{
int ret;
if (sizeof(cc->reqbuf) - cc->reqlen - 1 > 0) {
ret = recv(cc->sockfd, cc->reqbuf + cc->reqlen, sizeof(cc->reqbuf) - cc->reqlen - 1, MSG_DONTWAIT);
if (ret == 0) {
info("peer closed\n");
return -1;
}
if (ret == SOCKET_ERROR) {
if (sk_errno() != SK_EAGAIN && sk_errno() != SK_EINTR) {
err("recv data failed: %s\n", sk_strerror(sk_errno()));
return -1;
}
ret = 0;
}
cc->reqlen += ret;
cc->reqbuf[cc->reqlen] = 0;
}
if (cc->reqlen == 0) {
return 0;
}
ret = rtsp_msg_parse_from_array(msg, cc->reqbuf, cc->reqlen);
if (ret < 0) {
err("Invalid frame\n");
return -1;
}
if (ret == 0) {
return 0;
}
//dbg("recv %d bytes rtsp message from %s\n", ret, inet_ntoa(cc->peer_addr));
memmove(cc->reqbuf, cc->reqbuf + ret, cc->reqlen - ret);
cc->reqlen -= ret;
return ret;
}
static int rtsp_send_msg(struct rtsp_client_connection *cc, rtsp_msg_s *msg)
{
char szbuf[1024] = "";
int ret = rtsp_msg_build_to_array(msg, szbuf, sizeof(szbuf));
if (ret < 0) {
err("rtsp_msg_build_to_array failed\n");
return -1;
}
ret = send(cc->sockfd, szbuf, ret, 0);
if (ret == SOCKET_ERROR) {
err("rtsp response send failed: %s\n", sk_strerror(sk_errno()));
return -1;
}
//dbg("sent %d bytes rtsp message to %s\n", ret, inet_ntoa(cc->peer_addr));
return ret;
}
static int rtsp_recv_rtp_over_udp (struct rtsp_client_connection *cc, int isaudio)
{
struct rtp_connection *rtp = isaudio ? cc->artp : cc->vrtp;
struct sockaddr_in inaddr;
SOCKLEN addrlen = sizeof(inaddr);
char szbuf[128];
int len;
len = recvfrom(rtp->udp_sockfd[0], szbuf, sizeof(szbuf), MSG_DONTWAIT, (struct sockaddr*)&inaddr, &addrlen);
if (len == SOCKET_ERROR) {
if (sk_errno() != SK_EAGAIN && sk_errno() != SK_EINTR) {
warn("rtp over udp recv failed: %s\n", sk_strerror(sk_errno()));
return -1;
}
return 0;
}
//dbg("rtp over udp recv %d bytes from %s\n", len, inet_ntoa(rtp->peer_addr));
if (rtp->udp_peerport[0] != ntohs(inaddr.sin_port)) {
info("rtp over udp peer %s port change %u to %u\n", inet_ntoa(rtp->peer_addr),
rtp->udp_peerport[0], ntohs(inaddr.sin_port));
rtp->udp_peerport[0] = ntohs(inaddr.sin_port);
}
//TODO process rtp frame
return len;
}
static int rtsp_recv_rtcp_over_udp (struct rtsp_client_connection *cc, int isaudio)
{
struct rtp_connection *rtp = isaudio ? cc->artp : cc->vrtp;
struct sockaddr_in inaddr;
SOCKLEN addrlen = sizeof(inaddr);
char szbuf[128];
int len;
len = recvfrom(rtp->udp_sockfd[1], szbuf, sizeof(szbuf), MSG_DONTWAIT, (struct sockaddr*)&inaddr, &addrlen);
if (len == SOCKET_ERROR) {
if (sk_errno() != SK_EAGAIN && sk_errno() != SK_EINTR) {
warn("rtcp over udp recv failed: %s\n", sk_strerror(sk_errno()));
return -1;
}
return 0;
}
//dbg("rtcp over udp recv %d bytes from %s\n", len, inet_ntoa(rtp->peer_addr));
if (rtp->udp_peerport[1] != ntohs(inaddr.sin_port)) {
info("rtcp over udp peer %s port change %u to %u\n", inet_ntoa(rtp->peer_addr),
rtp->udp_peerport[1], ntohs(inaddr.sin_port));
rtp->udp_peerport[1] = ntohs(inaddr.sin_port);
}
//TODO process rtcp frame
return len;
}
static ssize_t send_with_poll(int socket_fd, const void *buf, size_t len, int flags)
{
ssize_t total_sent = 0;
size_t bytes_left = len;
const uint8_t *ptr = buf;
while (bytes_left > 0) {
ssize_t bytes_sent = send(socket_fd, ptr, bytes_left, flags);
if (bytes_sent > 0) {
total_sent += bytes_sent;
ptr += bytes_sent;
bytes_left -= bytes_sent;
} else if (bytes_sent == -1 && errno == EAGAIN) {
struct pollfd pfd;
pfd.fd = socket_fd;
pfd.events = POLLOUT;
pfd.revents = 0;
int ret = poll(&pfd, 1, 1000); // 1000 milliseconds timeout
if (ret == -1) {
perror("poll");
err("send poll error with some io error\n");
return -1;
} else if (ret == 0) {
err("send poll error with Timed out\n");
return total_sent > 0 ? total_sent : -1;
} else {
if ((pfd.revents & POLLOUT) == 0) {
perror("poll revents");
return -1;
}
}
} else {
perror("send");
err("send error with some io error\n");
return -1;
}
}
return total_sent;
}
static ssize_t sendto_with_poll(int socket_fd, const void *buf, size_t len, int flags, const struct sockaddr *dest_addr, socklen_t addrlen)
{
ssize_t total_sent = 0;
size_t bytes_left = len;
const uint8_t *ptr = buf;
while (bytes_left > 0) {
ssize_t bytes_sent = sendto(socket_fd, ptr, bytes_left, flags, dest_addr, addrlen);
if (bytes_sent > 0) {
total_sent += bytes_sent;
ptr += bytes_sent;
bytes_left -= bytes_sent;
} else if (bytes_sent == -1 && errno == EAGAIN) {
struct pollfd pfd;
pfd.fd = socket_fd;
pfd.events = POLLOUT;
pfd.revents = 0;
int ret = poll(&pfd, 1, 1000); // 1000 milliseconds timeout
if (ret == -1) {
perror("poll");
err("sendto poll error with some io error\n");
return -1;
} else if (ret == 0) {
err("sendto poll error with Timed out\n");
return total_sent > 0 ? total_sent : -1;
} else {
if ((pfd.revents & POLLOUT) == 0) {
perror("poll revents");
return -1;
}
}
} else {
perror("sendto");
err("sendto error with some io error\n");
return -1;
}
}
return total_sent;
}
static int rtp_tx_data(struct rtp_connection *c, const uint8_t *data, int size)
{
if (c->is_over_tcp) {
SOCKET sockfd = c->tcp_sockfd;
int ret = -1;
uint8_t szbuf[4] = {0};
sockfd = c->tcp_sockfd;
szbuf[0] = '$';
szbuf[1] = c->tcp_interleaved[0];
*((uint16_t*)&szbuf[2]) = htons(size);
// ret = send(sockfd, (const char*)szbuf, 4, 0);
ret = send_with_poll(sockfd, szbuf, 4, 0);
if (ret == SOCKET_ERROR) {
if (sk_errno() != SK_EAGAIN && sk_errno() != SK_EINTR) {
warn("rtp over tcp send interlaced frame to %s failed: %s\n",
inet_ntoa(c->peer_addr), sk_strerror(sk_errno()));
return -1;
}else{
err("rtp over tcp send interlaced frame to %s failed: %s\n",
inet_ntoa(c->peer_addr), sk_strerror(sk_errno()));
}
return 0;
}
ret = send_with_poll(sockfd, (const char*)data, size, 0);
// ret = send(sockfd, (const char*)data, size, 0);
if (ret == SOCKET_ERROR) {
if (sk_errno() != SK_EAGAIN && sk_errno() != SK_EINTR) {
warn("rtp over tcp send %d bytes to %s failed: %s\n", size, inet_ntoa(c->peer_addr), sk_strerror(sk_errno()));
return -1;
}else{
err("rtp over tcp send %d bytes to %s failed: %s\n", size, inet_ntoa(c->peer_addr), sk_strerror(sk_errno()));
}
return 0;
}
} else {
struct sockaddr_in inaddr;
SOCKET sockfd = c->udp_sockfd[0];
int ret = -1;
memset(&inaddr, 0, sizeof(inaddr));
inaddr.sin_family = AF_INET;
inaddr.sin_addr = c->peer_addr;
inaddr.sin_port = htons(c->udp_peerport[0]);
ret = sendto_with_poll(sockfd, (const char*)data, size, 0, (struct sockaddr*)&inaddr, sizeof(inaddr));
// ret = sendto(sockfd, (const char*)data, size, 0, (struct sockaddr*)&inaddr, sizeof(inaddr));
if (ret == SOCKET_ERROR) {
if (sk_errno() != SK_EAGAIN && sk_errno() != SK_EINTR) {
warn("rtp over udp send %d bytes to %s failed: %s\n", size, inet_ntoa(c->peer_addr), sk_strerror(sk_errno()));
return -1;
}else{
err("rtp over udp send %d bytes to %s failed: %s\n", size, inet_ntoa(c->peer_addr), sk_strerror(sk_errno()));
}
return 0;
}
}
return size;
}
static int rtsp_tx_video_packet (struct rtsp_client_connection *cc)
{
// struct rtsp_demo *d = cc->demo;
struct rtsp_session *s = cc->session;
struct stream_queue *q = s->vstreamq;
struct rtp_connection *rtp = cc->vrtp;
uint8_t *ppacket = NULL;
int *ppktlen = NULL;
int count = 0;
#if 0
dbg("index=%d head=%d tail=%d used=%d\n",
rtp->streamq_index,
streamq_head(q),
streamq_tail(q),
streamq_inused(q, rtp->streamq_index));
#endif
while (streamq_inused(q, rtp->streamq_index) > 0) {
streamq_query(q, rtp->streamq_index, (char**)&ppacket, &ppktlen);
// dbg("pktlen=%d\n", *ppktlen);
if (*ppktlen > 0) {
*((uint32_t*)(&ppacket[8])) = htonl(rtp->ssrc); //modify ssrc
int32_t nSend = rtp_tx_data(rtp, ppacket, *ppktlen);
if (nSend != *ppktlen) {
err("rtp_tx_data:[%d] != (%d)\n", nSend, *ppktlen);
break;
}
rtp->rtcp_packet_count ++;
rtp->rtcp_octet_count += *ppktlen - 12;//XXX
}
rtp->streamq_index = streamq_next(q, rtp->streamq_index);
count ++;
}
return count;
}
static int rtsp_tx_audio_packet (struct rtsp_client_connection *cc)
{
// struct rtsp_demo *d = cc->demo;
struct rtsp_session *s = cc->session;
struct stream_queue *q = s->astreamq;
struct rtp_connection *rtp = cc->artp;
uint8_t *ppacket = NULL;
int *ppktlen = NULL;
int count = 0;
while (streamq_inused(q, rtp->streamq_index) > 0) {
streamq_query(q, rtp->streamq_index, (char**)&ppacket, &ppktlen);
if (*ppktlen > 0) {
*((uint32_t*)(&ppacket[8])) = htonl(rtp->ssrc); //modify ssrc
if (rtp_tx_data(rtp, ppacket, *ppktlen) != *ppktlen) {
break;
}
rtp->rtcp_packet_count ++;
rtp->rtcp_octet_count += *ppktlen - 12;//XXX
}
rtp->streamq_index = streamq_next(q, rtp->streamq_index);
count ++;
}
return count;
}
static int rtsp_do_event(rtsp_demo_handle demo, rtspClientConnect connectCB, rtspClientDisconnect disconnectCB, void *userData)
{
struct rtsp_demo *d = (struct rtsp_demo*)demo;
struct rtsp_client_connection *cc = NULL;
struct timeval tv;
fd_set rfds;
fd_set wfds;
SOCKET maxfd;
int ret;
if (NULL == d) {
return -1;
}
FD_ZERO(&rfds);
FD_ZERO(&wfds);
FD_SET(d->sockfd, &rfds);
maxfd = d->sockfd;
TAILQ_FOREACH(cc, &d->connections_qhead, demo_entry) {
struct rtsp_session *s = cc->session;
struct rtp_connection *vrtp = cc->vrtp;
struct rtp_connection *artp = cc->artp;
FD_SET(cc->sockfd, &rfds);
if (cc->sockfd > maxfd)
maxfd = cc->sockfd;
if (cc->state != RTSP_CC_STATE_PLAYING)
continue;
if (vrtp && streamq_inused(s->vstreamq, vrtp->streamq_index) > 0) {
//add video rtp sock to wfds
if (vrtp->is_over_tcp) {
FD_SET(vrtp->tcp_sockfd, &wfds);
if (vrtp->tcp_sockfd > maxfd)
maxfd = vrtp->tcp_sockfd;
} else {
FD_SET(vrtp->udp_sockfd[0], &wfds);
if (vrtp->udp_sockfd[0] > maxfd)
maxfd = vrtp->udp_sockfd[0];
}
}
if (artp && streamq_inused(s->astreamq, artp->streamq_index) > 0) {
//add audio rtp sock to wfds
if (artp->is_over_tcp) {
FD_SET(artp->tcp_sockfd, &wfds);
if (artp->tcp_sockfd > maxfd)
maxfd = artp->tcp_sockfd;
} else {
FD_SET(artp->udp_sockfd[0], &wfds);
if (artp->udp_sockfd[0] > maxfd)
maxfd = artp->udp_sockfd[0];
}
}
if (vrtp && (!vrtp->is_over_tcp)) {
//add video rtcp sock to rfds
FD_SET(vrtp->udp_sockfd[0], &rfds);
FD_SET(vrtp->udp_sockfd[1], &rfds);
if (vrtp->udp_sockfd[0] > maxfd)
maxfd = vrtp->udp_sockfd[0];
if (vrtp->udp_sockfd[1] > maxfd)
maxfd = vrtp->udp_sockfd[1];
}
if (artp && (!artp->is_over_tcp)) {
//add audio rtcp sock to rfds
FD_SET(artp->udp_sockfd[0], &rfds);
FD_SET(artp->udp_sockfd[1], &rfds);
if (artp->udp_sockfd[0] > maxfd)
maxfd = artp->udp_sockfd[0];
if (artp->udp_sockfd[1] > maxfd)
maxfd = artp->udp_sockfd[1];
}
}
memset(&tv, 0, sizeof(tv));
tv.tv_sec = 0;
tv.tv_usec = 0;
//rtp rtcp 锟侥凤拷锟斤拷 锟斤拷锟斤拷占锟斤拷锟�
ret = select(maxfd + 1, &rfds, &wfds, NULL, &tv);
if (ret < 0) {
err("select failed : %s\n", strerror(errno));
return -1;
}
if (ret == 0) {
return 0;
}
if (FD_ISSET(d->sockfd, &rfds)) {
//new client_connection
rtsp_new_client_connection(connectCB, userData, d);
}
cc = TAILQ_FIRST(&d->connections_qhead); //NOTE do not use TAILQ_FOREACH
while (cc) {//锟斤拷锟斤拷rtsp_client_connection
struct rtsp_client_connection *cc1 = cc;
struct rtsp_session *s = cc1->session; //锟斤拷锟斤拷锟接碉拷rtsp session
struct rtp_connection *vrtp = cc1->vrtp;//锟斤拷锟斤拷锟接碉拷rtp session 锟斤拷频
struct rtp_connection *artp = cc1->artp;
cc = TAILQ_NEXT(cc, demo_entry);//锟斤拷一锟斤拷rtsp_client_connection
if (FD_ISSET(cc1->sockfd, &rfds)) {
do {
rtsp_msg_s reqmsg, resmsg;
rtsp_msg_init(&reqmsg);
rtsp_msg_init(&resmsg);
ret = rtsp_recv_msg(cc1, &reqmsg);
if (ret == 0)
break;
if (ret < 0) {
rtsp_del_client_connection(disconnectCB, userData, cc1);
cc1 = NULL;
break;
}
if (reqmsg.type == RTSP_MSG_TYPE_INTERLEAVED) {
//TODO process RTCP over TCP frame
rtsp_msg_free(&reqmsg);
continue;
}
if (reqmsg.type != RTSP_MSG_TYPE_REQUEST) {
err("not request frame.\n");
rtsp_msg_free(&reqmsg);
continue;
}
ret = rtsp_process_request(cc1, &reqmsg, &resmsg);
if (ret < 0) {
err("request internal err\n");
} else {
rtsp_send_msg(cc1, &resmsg);
}
rtsp_msg_free(&reqmsg);
rtsp_msg_free(&resmsg);
} while (cc1);
if (cc1 == NULL)
continue;
}
if (cc1->state != RTSP_CC_STATE_PLAYING)
continue;
if (vrtp && streamq_inused(s->vstreamq, vrtp->streamq_index) > 0) {
//send rtp video packet
if (vrtp->is_over_tcp && FD_ISSET(vrtp->tcp_sockfd, &wfds)) {
rtsp_tx_video_packet(cc1);
//printf("v");fflush(stdout);
} else if ((!vrtp->is_over_tcp) && FD_ISSET(vrtp->udp_sockfd[0], &wfds)) {
rtsp_tx_video_packet(cc1);
//printf("v");fflush(stdout);
}
}
if (artp && streamq_inused(s->astreamq, artp->streamq_index) > 0) {
//send rtp audio packet
if (artp->is_over_tcp && FD_ISSET(artp->tcp_sockfd, &wfds)) {
rtsp_tx_audio_packet(cc1);
//printf("a");fflush(stdout);
} else if (0 == artp->is_over_tcp && FD_ISSET(artp->udp_sockfd[0], &wfds)) {
rtsp_tx_audio_packet(cc1);
//printf("a");fflush(stdout);
}
}
if (vrtp && (!vrtp->is_over_tcp)) {
//process video rtcp socket
if (FD_ISSET(vrtp->udp_sockfd[0], &rfds)) {
rtsp_recv_rtp_over_udp(cc1, 0);
}
if (FD_ISSET(vrtp->udp_sockfd[1], &rfds)) {
rtsp_recv_rtcp_over_udp(cc1, 0);
}
}
if (artp && (!artp->is_over_tcp)) {
//process audio rtcp socket
if (FD_ISSET(artp->udp_sockfd[0], &rfds)) {
rtsp_recv_rtp_over_udp(cc1, 1);
}
if (FD_ISSET(artp->udp_sockfd[1], &rfds)) {
rtsp_recv_rtcp_over_udp(cc1, 1);
}
}
}
return 1;
}
static int rtcp_try_tx_sr (struct rtp_connection *c, uint64_t ntptime_of_zero_ts, uint64_t ts, uint32_t sample_rate);
static int rtsp_tx_video (rtsp_session_handle session, const uint8_t *frame, int len, uint64_t ts)
{
struct rtsp_session *s = (struct rtsp_session*) session;
struct stream_queue *q = NULL;
struct rtsp_client_connection *cc = NULL;
uint8_t *packets[VRTP_MAX_NBPKTS+1] = {NULL};
int pktsizs[VRTP_MAX_NBPKTS+1] = {0};
int *pktlens[VRTP_MAX_NBPKTS] = {NULL};
int i, index, count, start;
if (!s || !frame || s->vcodec_id == RTSP_CODEC_ID_NONE)
return -1;
//get free buffer
q = s->vstreamq;
index = streamq_tail(q);
for (i = 0; i < VRTP_MAX_NBPKTS; i++) {
if (streamq_next(q, index) == streamq_head(q))
streamq_pop(q);
streamq_query(q, index, (char**)&packets[i], &pktlens[i]);
pktsizs[i] = RTP_MAX_PKTSIZ;
index = streamq_next(q, index);
}
packets[i] = NULL;
pktsizs[i] = 0;
//move all slow rtp connections streamq_index to queue tail
TAILQ_FOREACH(cc, &s->connections_qhead, session_entry) {
struct rtp_connection *rtp = cc->vrtp;
if (cc->state != RTSP_CC_STATE_PLAYING || !rtp)
continue;
if (!streamq_inused(q, rtp->streamq_index) && rtp->streamq_index != streamq_tail(q)) {
rtp->streamq_index = streamq_head(q);
warn("client %s will lost video packet\n", inet_ntoa(cc->peer_addr));
}
}
switch (s->vcodec_id) {
case RTSP_CODEC_ID_VIDEO_H264:
if (s->vcodec_data.h264.pps_len == 0 || s->vcodec_data.h264.pps_len == 0) {
if (rtsp_codec_data_parse_from_frame_h264(frame, len, &s->vcodec_data.h264) < 0) {
warn("rtsp_codec_data_parse_from_frame_h264 failed\n");
} else {
// dbg("rtsp_codec_data_parse_from_frame_h264 sps:%u pps:%u success\n",
// s->vcodec_data.h264.sps_len, s->vcodec_data.h264.pps_len);
}
}
break;
case RTSP_CODEC_ID_VIDEO_H265:
if (s->vcodec_data.h265.pps_len == 0 || s->vcodec_data.h265.pps_len == 0 || s->vcodec_data.h265.vps_len == 0) {
if (rtsp_codec_data_parse_from_frame_h265(frame, len, &s->vcodec_data.h265) < 0) {
warn("rtsp_codec_data_parse_from_frame_h265 failed\n");
} else {
// dbg("rtsp_codec_data_parse_from_frame_h265 vps:%u sps:%u pps:%u success\n",
// s->vcodec_data.h265.vps, s->vcodec_data.h265.sps_len, s->vcodec_data.h265.pps_len);
}
}
break;
}
start = 0;
count = 0;
while (start < len && packets[count] && pktsizs[count] > 0) {
const uint8_t *p = NULL;
int size = 0;
int ret;
p = rtsp_find_h264_h265_nalu(frame + start, len - start, &size);
if (!p) {
warn("not found nal header\n");
break;
}
//dbg("size:%d\n", size);
switch (s->vcodec_id) {
case RTSP_CODEC_ID_VIDEO_H264:
ret = rtp_enc_h264(&s->vrtpe, p, size, ts, &packets[count], &pktsizs[count]);
if (ret <= 0) {
err("rtp_enc_h264 ret = %d\n", ret);
return -1;
}
break;
case RTSP_CODEC_ID_VIDEO_H265:
ret = rtp_enc_h265(&s->vrtpe, p, size, ts, &packets[count], &pktsizs[count]);
if (ret <= 0) {
err("rtp_enc_h265 ret = %d\n", ret);
return -1;
}
break;
}
count += ret;
start = p - frame + size;
}
for (i = 0; i < count; i++) {
*pktlens[i] = pktsizs[i];
streamq_push(q);
}
//first send
TAILQ_FOREACH(cc, &s->connections_qhead, session_entry) {
struct rtp_connection *rtp = cc->vrtp;
if (cc->state != RTSP_CC_STATE_PLAYING || !rtp)
continue;
rtcp_try_tx_sr(rtp, s->video_ntptime_of_zero_ts, ts, s->vrtpe.sample_rate);
rtsp_tx_video_packet(cc);
}
return len;
}
static int rtsp_sever_tx_video(const RtspHandle* pHandle, const uint8_t *frame, int len, uint64_t ts)
{
if(pHandle == NULL || pHandle->g_demo == NULL || pHandle->g_session == NULL){
err("Could not get the rtsp session\n");
return -1;
}
struct rtsp_session *s = (struct rtsp_session*) pHandle->g_session;
struct stream_queue *q = NULL;
struct rtsp_client_connection *cc = NULL;
uint8_t *packets[VRTP_MAX_NBPKTS+1] = {NULL};
int pktsizs[VRTP_MAX_NBPKTS+1] = {0};
int *pktlens[VRTP_MAX_NBPKTS] = {NULL};
int i, index, count, start;
if (!s || !frame || s->vcodec_id == RTSP_CODEC_ID_NONE)
return -1;
//get free buffer
q = s->vstreamq;
index = streamq_tail(q);
for (i = 0; i < VRTP_MAX_NBPKTS; i++) {
if (streamq_next(q, index) == streamq_head(q))
streamq_pop(q);
streamq_query(q, index, (char**)&packets[i], &pktlens[i]);
pktsizs[i] = RTP_MAX_PKTSIZ;
index = streamq_next(q, index);
}
packets[i] = NULL;
pktsizs[i] = 0;
//move all slow rtp connections streamq_index to queue tail
TAILQ_FOREACH(cc, &s->connections_qhead, session_entry) {
struct rtp_connection *rtp = cc->vrtp;
if (cc->state != RTSP_CC_STATE_PLAYING || !rtp)
continue;
if (!streamq_inused(q, rtp->streamq_index) && rtp->streamq_index != streamq_tail(q)) {
rtp->streamq_index = streamq_head(q);
warn("client %s will lost video packet\n", inet_ntoa(cc->peer_addr));
}
}
switch (s->vcodec_id) {
case RTSP_CODEC_ID_VIDEO_H264:
if (s->vcodec_data.h264.pps_len == 0 || s->vcodec_data.h264.pps_len == 0) {
if (rtsp_codec_data_parse_from_frame_h264(frame, len, &s->vcodec_data.h264) < 0) {
warn("rtsp_codec_data_parse_from_frame_h264 failed\n");
} else {
// dbg("rtsp_codec_data_parse_from_frame_h264 sps:%u pps:%u success\n",
// s->vcodec_data.h264.sps_len, s->vcodec_data.h264.pps_len);
}
}
break;
case RTSP_CODEC_ID_VIDEO_H265:
if (s->vcodec_data.h265.pps_len == 0 || s->vcodec_data.h265.pps_len == 0 || s->vcodec_data.h265.vps_len == 0) {
if (rtsp_codec_data_parse_from_frame_h265(frame, len, &s->vcodec_data.h265) < 0) {
warn("rtsp_codec_data_parse_from_frame_h265 failed\n");
} else {
// dbg("rtsp_codec_data_parse_from_frame_h265 vps:%u sps:%u pps:%u success\n",
// s->vcodec_data.h265.vps, s->vcodec_data.h265.sps_len, s->vcodec_data.h265.pps_len);
}
}
break;
}
start = 0;
count = 0;
while (start < len && packets[count] && pktsizs[count] > 0) {
const uint8_t *p = NULL;
int size = 0;
int ret = 0;
p = rtsp_find_h264_h265_nalu(frame + start, len - start, &size);
if (!p) {
warn("not found nal header\n");
break;
}
//dbg("size:%d\n", size);
switch (s->vcodec_id) {
case RTSP_CODEC_ID_VIDEO_H264:
ret = rtp_enc_h264(&s->vrtpe, p, size, ts, &packets[count], &pktsizs[count]);
if (ret <= 0) {
err("rtp_enc_h264 ret = %d\n", ret);
return -1;
}
break;
case RTSP_CODEC_ID_VIDEO_H265:
ret = rtp_enc_h265(&s->vrtpe, p, size, ts, &packets[count], &pktsizs[count]);
if (ret <= 0) {
err("rtp_enc_h265 ret = %d\n", ret);
return -1;
}
break;
}
count += ret;
start = p - frame + size;
}
for (i = 0; i < count; i++) {
*pktlens[i] = pktsizs[i];
streamq_push(q);
}
//first send
TAILQ_FOREACH(cc, &s->connections_qhead, session_entry) {
struct rtp_connection *rtp = cc->vrtp;
if (cc->state != RTSP_CC_STATE_PLAYING || !rtp)
continue;
rtcp_try_tx_sr(rtp, s->video_ntptime_of_zero_ts, ts, s->vrtpe.sample_rate);
rtsp_tx_video_packet(cc);
}
rtsp_do_event(pHandle->g_demo, pHandle->pRtspClientConnect, pHandle->pRtspClientDisconnect, pHandle->pUserData);
return len;
}
static int rtsp_tx_audio (rtsp_session_handle session, const uint8_t *frame, int len, uint64_t ts)
{
struct rtsp_session *s = (struct rtsp_session*) session;
struct stream_queue *q = NULL;
struct rtsp_client_connection *cc = NULL;
uint8_t *packets[ARTP_MAX_NBPKTS+1] = {NULL};
int pktsizs[ARTP_MAX_NBPKTS+1] = {0};
int *pktlens[ARTP_MAX_NBPKTS] = {NULL};
int i = 0, index = 0, count = 0;
if (!s || !frame || s->acodec_id == RTSP_CODEC_ID_NONE)
return -1;
//get free buffer
q = s->astreamq;
index = streamq_tail(q);
for (i = 0; i < ARTP_MAX_NBPKTS; i++) {
if (streamq_next(q, index) == streamq_head(q))
streamq_pop(q);
streamq_query(q, index, (char**)&packets[i], &pktlens[i]);
pktsizs[i] = RTP_MAX_PKTSIZ;
index = streamq_next(q, index);
}
packets[i] = NULL;
pktsizs[i] = 0;
//move all slow rtp connections streamq_index to queue tail
TAILQ_FOREACH(cc, &s->connections_qhead, session_entry) {
struct rtp_connection *rtp = cc->artp;
if (cc->state != RTSP_CC_STATE_PLAYING || !rtp)
continue;
if (!streamq_inused(q, rtp->streamq_index) && rtp->streamq_index != streamq_tail(q)) {
rtp->streamq_index = streamq_head(q);
warn("client %s will lost audio packet\n", inet_ntoa(cc->peer_addr));
}
}
switch (s->acodec_id) {
case RTSP_CODEC_ID_AUDIO_G711A:
case RTSP_CODEC_ID_AUDIO_G711U:
count = rtp_enc_g711(&s->artpe, frame, len, ts, packets, pktsizs);
if (count <= 0) {
err("rtp_enc_g711 ret = %d\n", count);
return -1;
}
break;
case RTSP_CODEC_ID_AUDIO_G726:
count = rtp_enc_g726(&s->artpe, frame, len, ts, packets, pktsizs);
if (count <= 0) {
err("rtp_enc_g726 ret = %d\n", count);
return -1;
}
break;
case RTSP_CODEC_ID_AUDIO_AAC:
if (s->acodec_data.aac.audio_specific_config_len == 0) {
if (rtsp_codec_data_parse_from_frame_aac(frame, len, &s->acodec_data.aac) < 0) {
// warn("rtsp_codec_data_parse_from_frame_aac failed\n");
} else {
// dbg("rtsp_codec_data_parse_from_frame_aac success\n");
s->artpe.sample_rate = s->acodec_data.aac.sample_rate;
}
}
count = rtp_enc_aac(&s->artpe, frame, len, ts, packets, pktsizs);
if (count <= 0) {
err("rtp_enc_aac ret = %d\n", count);
return -1;
}
break;
}
for (i = 0; i < count; i++) {
*pktlens[i] = pktsizs[i];
streamq_push(q);
}
//first send
TAILQ_FOREACH(cc, &s->connections_qhead, session_entry) {
struct rtp_connection *rtp = cc->artp;
if (cc->state != RTSP_CC_STATE_PLAYING || !rtp)
continue;
rtcp_try_tx_sr(rtp, s->audio_ntptime_of_zero_ts, ts, s->artpe.sample_rate);
rtsp_tx_audio_packet(cc);
}
return len;
}
//return us from system running
static uint64_t rtsp_get_reltime (void)
{
#ifdef __WIN32__
return (timeGetTime() * 1000ULL);
#endif
#ifdef __LINUX__
struct timespec tp;
clock_gettime(CLOCK_MONOTONIC, &tp);
return (tp.tv_sec * 1000000ULL + tp.tv_nsec / 1000ULL);
#endif
}
//return us from 1970/1/1 00:00:00
static uint64_t rtsp_get_abstime (void)
{
#ifdef __WIN32__
FILETIME ft;
uint64_t t;
GetSystemTimeAsFileTime(&ft);
t = (uint64_t)ft.dwHighDateTime << 32 | ft.dwLowDateTime;
return t / 10 - 11644473600000000ULL;
#endif
#ifdef __LINUX__
struct timeval tv;
gettimeofday(&tv, NULL);
return (tv.tv_sec * 1000000ULL + tv.tv_usec);
#endif
}
//return us from 1900/1/1 00:00:00
static uint64_t rtsp_get_ntptime (void)
{
#define NTP_OFFSET_US (2208988800000000ULL)
return (rtsp_get_abstime() + NTP_OFFSET_US);
}
static int rtsp_sync_video_ts (rtsp_session_handle session, uint64_t ts, uint64_t ntptime)
{
struct rtsp_session *s = (struct rtsp_session*) session;
if (!s || s->vcodec_id == RTSP_CODEC_ID_NONE)
return -1;
s->video_ntptime_of_zero_ts = ntptime - ts; //XXX
return 0;
}
static int rtsp_sync_audio_ts (rtsp_session_handle session, uint64_t ts, uint64_t ntptime)
{
struct rtsp_session *s = (struct rtsp_session*) session;
if (!s || s->acodec_id == RTSP_CODEC_ID_NONE)
return -1;
s->audio_ntptime_of_zero_ts = ntptime - ts; //XXX
return 0;
}
struct rtcp_sr {
#ifdef __BIG_ENDIAN__
uint16_t pt:8;
uint16_t v:2;
uint16_t p:1;
uint16_t rc:5;
#else
uint16_t rc:5;
uint16_t p:1;
uint16_t v:2;
uint16_t pt:8;
#endif
uint16_t length;
uint32_t ssrc;
uint32_t ntpts_msw;
uint32_t ntpts_lsw;
uint32_t rtp_ts;
uint32_t packet_count;
uint32_t octet_count;
};
static int rtcp_try_tx_sr (struct rtp_connection *c, uint64_t ntptime_of_zero_ts, uint64_t ts, uint32_t sample_rate)
{
struct rtcp_sr sr = {0};
uint32_t ntpts_msw = 0;
uint32_t ntpts_lsw = 0;
uint32_t rtp_ts = 0;
int size = sizeof(sr);
if (c->rtcp_last_ts && ts < c->rtcp_last_ts + 5000000ULL) {
return 0;
}
ntpts_msw = (uint32_t)((ntptime_of_zero_ts + ts) / 1000000ULL);
ntpts_lsw = (uint32_t)(((ntptime_of_zero_ts + ts) % 1000000ULL) * (1ULL << 32) / 1000000ULL);
rtp_ts = (uint32_t)(ts * sample_rate / 1000000ULL);
sr.v = 2;
sr.p = 0;
sr.rc = 0;
sr.pt = 200;
sr.length = htons(size / 4 - 1);
sr.ssrc = htonl(c->ssrc);
sr.ntpts_msw = htonl(ntpts_msw);
sr.ntpts_lsw = htonl(ntpts_lsw);
sr.rtp_ts = htonl(rtp_ts);
sr.packet_count = htonl(c->rtcp_packet_count);
sr.octet_count = htonl(c->rtcp_octet_count);
if (c->is_over_tcp) {
SOCKET sockfd = c->tcp_sockfd;
int ret = -1;
uint8_t szbuf[4 + sizeof(struct rtcp_sr)];
sockfd = c->tcp_sockfd;
szbuf[0] = '$';
szbuf[1] = c->tcp_interleaved[1];
*((uint16_t*)&szbuf[2]) = htons(size);
memcpy(&szbuf[4], &sr, size);
ret = send(sockfd, (const char*)szbuf, sizeof(szbuf), 0);
if (ret == SOCKET_ERROR) {
if (sk_errno() != SK_EAGAIN && sk_errno() != SK_EINTR) {
warn("rtcp over tcp send frame to %s failed: %s\n", inet_ntoa(c->peer_addr), sk_strerror(sk_errno()));
return -1;
}
return 0;
}
} else {
struct sockaddr_in inaddr;
SOCKET sockfd = c->udp_sockfd[1];
int ret = -1;
memset(&inaddr, 0, sizeof(inaddr));
inaddr.sin_family = AF_INET;
inaddr.sin_addr = c->peer_addr;
inaddr.sin_port = htons(c->udp_peerport[1]);
ret = sendto(sockfd, (const char*)&sr, size, 0, (struct sockaddr*)&inaddr, sizeof(inaddr));
if (ret == SOCKET_ERROR) {
if (sk_errno() != SK_EAGAIN && sk_errno() != SK_EINTR) {
warn("rtcp over udp send %d bytes to %s failed: %s\n", size, inet_ntoa(c->peer_addr), sk_strerror(sk_errno()));
return -1;
}
return 0;
}
}
c->rtcp_last_ts = ts;
return size;
}
static int get_random_free_port(int min_port, int max_port)
{
if (min_port > max_port) {
printf("Invalid port range\n");
return -1;
}
srand(time(NULL)); // 浠ユ椂闂翠綔涓虹瀛愬垵濮嬪寲闅忔満鏁扮敓鎴愬櫒
int port_range = max_port - min_port + 1;
for (int i = 0; i < port_range; ++i) {
int random_port = min_port + rand() % port_range;
int free_port = rtsp_find_free_port(random_port, 1);
if (free_port != -1) {
return free_port;
}
}
return -1; // 濡傛灉鎵句笉鍒扮┖闂茬鍙o紝杩斿洖-1
}
RtspHandle* rtsp_init(const rtsp_cfg *cfg, rtspClientConnect onConnect,rtspClientDisconnect onDisconnect, void *userData)
{
char ourIPAddr[22]={0};
if(cfg==NULL){
err("init_rtsp rtsp_cfg = NULL err\n");
return NULL;
}
RtspHandle *m_rtspHandle = (RtspHandle *)malloc(sizeof(RtspHandle));
if(m_rtspHandle == NULL){
err("init_rtsp malloc RtspHandle err\n");
return NULL;
}
memset(m_rtspHandle, 0, sizeof(RtspHandle));
m_rtspHandle->pRtspClientConnect = onConnect;
m_rtspHandle->pRtspClientDisconnect = onDisconnect;
m_rtspHandle->pUserData = userData;
m_rtspHandle->g_demo = NULL;
m_rtspHandle->g_session = NULL;
memset(ourIPAddr,0,sizeof(ourIPAddr));
if(rtsp_getOurIpAddr(ourIPAddr)!=0){
memset(ourIPAddr,0,sizeof(ourIPAddr));
strcpy(ourIPAddr,"127.0.0.1");
}
int dst_port =cfg->port;
if(dst_port==0){
dst_port = get_random_free_port(10000, 65535);
}
m_rtspHandle->g_demo = create_rtsp_demo(&dst_port);
if(m_rtspHandle->g_demo==NULL){
err("create_rtsp_demo err!\n");
return NULL;
}
m_rtspHandle->g_session = create_rtsp_session(m_rtspHandle->g_demo, (rtsp_codec_id)cfg->videoCodec, cfg->suffix);
if(m_rtspHandle->g_session==NULL){
err("create_rtsp_session err!\n");
return NULL;
}
snprintf(m_rtspHandle->outRtspUrl,sizeof(m_rtspHandle->outRtspUrl), "rtsp://%s:%d%s",ourIPAddr,dst_port,cfg->suffix);
return m_rtspHandle;
}
int rtsp_server_write_video(const RtspHandle* pHandle, const uint8_t *frame, int len, uint64_t ts)
{
int ret = 0;
if(pHandle == NULL || pHandle->g_demo==NULL || pHandle->g_session==NULL){
err("not init \n");
return RTSP_FAIL;
}
if(frame==NULL){
err("send frame null \n");
return RTSP_FAIL;
}
ret = rtsp_sever_tx_video(pHandle,frame,len,ts);
if(ret<0)return RTSP_FAIL;
return RTSP_SUCC;
}
int rtsp_server_write_audio(const RtspHandle* pHandle, const uint8_t *frame, int len, uint64_t ts)
{
int ret = 0;
if(pHandle == NULL || pHandle->g_demo==NULL || pHandle->g_session==NULL){
err("not init \n");
return RTSP_FAIL;
}
ret = rtsp_tx_audio(pHandle->g_session,frame,len,ts);
if(ret<0)return RTSP_FAIL;
return RTSP_SUCC;
}
int rtsp_uinit(RtspHandle* pHandle)
{
if(pHandle == NULL){
err("not init \n");
return RTSP_SUCC;
}
if(pHandle->g_session != NULL){
rtsp_del_session(pHandle->pRtspClientDisconnect, pHandle->pUserData, pHandle->g_session);
pHandle->g_session = NULL;
}
if(pHandle->g_demo != NULL){
rtsp_del_demo(pHandle->pRtspClientDisconnect, pHandle->pUserData, pHandle->g_demo);
pHandle->g_session = NULL;
}
pHandle->pUserData = NULL;
pHandle->pRtspClientConnect = NULL;
pHandle->pRtspClientDisconnect = NULL;
free(pHandle);
pHandle = NULL;
return RTSP_SUCC;
}