external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

Merge branch 'feature/rtc' into develop

Browse source
This commit is contained in:
winlin 2020-05-03 20:20:15 +08:00
commit e2072a0bf6
29 changed files with 785 additions and 820 deletions
Download patch file Download diff file Expand all files Collapse all files

29
trunk/3rdparty/st-srs/io.c vendored
View file

@ -742,17 +742,16 @@ int st_sendmsg(_st_netfd_t *fd, const struct msghdr *msg, int flags, st_utime_t
return n; return n;
} }
int st_sendmmsg(st_netfd_t fd, struct st_mmsghdr *msgvec, unsigned int vlen, int flags, st_utime_t timeout)
#if defined(MD_HAVE_SENDMMSG) && defined(__linux__) && defined(_GNU_SOURCE)
int st_sendmmsg(st_netfd_t fd, struct mmsghdr *msgvec, unsigned int vlen, int flags, st_utime_t timeout)
{ {
#if defined(MD_HAVE_SENDMMSG) && defined(_GNU_SOURCE)
int n; int n;
int left; int left;
struct mmsghdr *p; struct mmsghdr *p;
left = (int)vlen; left = (int)vlen;
while (left > 0) { while (left > 0) {
p = msgvec + (vlen - left); p = (struct mmsghdr*)msgvec + (vlen - left);
if ((n = sendmmsg(fd->osfd, p, left, flags)) < 0) { if ((n = sendmmsg(fd->osfd, p, left, flags)) < 0) {
if (errno == EINTR) if (errno == EINTR)
@ -772,8 +771,30 @@ int st_sendmmsg(st_netfd_t fd, struct mmsghdr *msgvec, unsigned int vlen, int fl
return n; return n;
} }
return (int)vlen - left; return (int)vlen - left;
#else
struct st_mmsghdr *p;
int i, n;
// @see http://man7.org/linux/man-pages/man2/sendmmsg.2.html
for (i = 0; i < (int)vlen; ++i) {
p = msgvec + i;
n = st_sendmsg(fd, &p->msg_hdr, flags, timeout);
if (n < 0) {
// An error is returned only if no datagrams could be sent.
if (i == 0) {
return n;
} }
return i + 1;
}
p->msg_len = n;
}
// Returns the number of messages sent from msgvec; if this is less than vlen, the caller can retry with a
// further sendmmsg() call to send the remaining messages.
return vlen;
#endif #endif
}
/* /*

12
trunk/3rdparty/st-srs/public.h vendored
View file

@ -151,9 +151,15 @@ extern int st_recvfrom(st_netfd_t fd, void *buf, int len, struct sockaddr *from,
extern int st_sendto(st_netfd_t fd, const void *msg, int len, const struct sockaddr *to, int tolen, st_utime_t timeout); extern int st_sendto(st_netfd_t fd, const void *msg, int len, const struct sockaddr *to, int tolen, st_utime_t timeout);
extern int st_recvmsg(st_netfd_t fd, struct msghdr *msg, int flags, st_utime_t timeout); extern int st_recvmsg(st_netfd_t fd, struct msghdr *msg, int flags, st_utime_t timeout);
extern int st_sendmsg(st_netfd_t fd, const struct msghdr *msg, int flags, st_utime_t timeout); extern int st_sendmsg(st_netfd_t fd, const struct msghdr *msg, int flags, st_utime_t timeout);
#if defined(__linux__) && defined(_GNU_SOURCE)
extern int st_sendmmsg(st_netfd_t fd, struct mmsghdr *msgvec, unsigned int vlen, int flags, st_utime_t timeout); // @see http://man7.org/linux/man-pages/man2/sendmmsg.2.html
#endif #include <sys/socket.h>
struct st_mmsghdr {
struct msghdr msg_hdr; /* Message header */
unsigned int msg_len; /* Number of bytes transmitted */
};
extern int st_sendmmsg(st_netfd_t fd, struct st_mmsghdr *msgvec, unsigned int vlen, int flags, st_utime_t timeout);
extern st_netfd_t st_open(const char *path, int oflags, mode_t mode); extern st_netfd_t st_open(const char *path, int oflags, mode_t mode);
#ifdef DEBUG #ifdef DEBUG

5
trunk/auto/auto_headers.sh
View file

@ -165,11 +165,6 @@ if [ $SRS_SENDMMSG = YES ]; then
else else
srs_undefine_macro "SRS_SENDMMSG" $SRS_AUTO_HEADERS_H srs_undefine_macro "SRS_SENDMMSG" $SRS_AUTO_HEADERS_H
fi fi
if [ $SRS_HAS_SENDMMSG = YES ]; then
srs_define_macro "SRS_HAS_SENDMMSG" $SRS_AUTO_HEADERS_H
else
srs_undefine_macro "SRS_HAS_SENDMMSG" $SRS_AUTO_HEADERS_H
fi
if [ $SRS_DEBUG = YES ]; then if [ $SRS_DEBUG = YES ]; then
srs_define_macro "SRS_DEBUG" $SRS_AUTO_HEADERS_H srs_define_macro "SRS_DEBUG" $SRS_AUTO_HEADERS_H

2
trunk/auto/depends.sh
View file

@ -347,7 +347,7 @@ if [ $SRS_EXPORT_LIBRTMP_PROJECT = NO ]; then
_ST_MAKE=darwin-debug && _ST_EXTRA_CFLAGS="-DMD_HAVE_KQUEUE" && _ST_LD=${SRS_TOOL_CC} && _ST_OBJ="DARWIN_`uname -r`_DBG" _ST_MAKE=darwin-debug && _ST_EXTRA_CFLAGS="-DMD_HAVE_KQUEUE" && _ST_LD=${SRS_TOOL_CC} && _ST_OBJ="DARWIN_`uname -r`_DBG"
fi fi
# For UDP sendmmsg, disable it if not suppported. # For UDP sendmmsg, disable it if not suppported.
if [[ $SRS_HAS_SENDMMSG == YES ]]; then if [[ $SRS_SENDMMSG == YES ]]; then
echo "Build ST with UDP sendmmsg support." echo "Build ST with UDP sendmmsg support."
_ST_EXTRA_CFLAGS="$_ST_EXTRA_CFLAGS -DMD_HAVE_SENDMMSG -D_GNU_SOURCE" _ST_EXTRA_CFLAGS="$_ST_EXTRA_CFLAGS -DMD_HAVE_SENDMMSG -D_GNU_SOURCE"
else else

36
trunk/auto/options.sh
View file

@ -118,11 +118,9 @@ SRS_EXTRA_FLAGS=
# #
##################################################################################### #####################################################################################
# Performance optimize. # Performance optimize.
SRS_NASM=YES SRS_NASM=NO
SRS_SRTP_ASM=YES SRS_SRTP_ASM=NO
SRS_SENDMMSG=YES SRS_SENDMMSG=NO
SRS_HAS_SENDMMSG=YES
SRS_DETECT_SENDMMSG=YES
SRS_DEBUG=NO SRS_DEBUG=NO
##################################################################################### #####################################################################################
@ -257,8 +255,6 @@ function parse_user_option() {
--export-librtmp-project) SRS_EXPORT_LIBRTMP_PROJECT=${value} ;; --export-librtmp-project) SRS_EXPORT_LIBRTMP_PROJECT=${value} ;;
--export-librtmp-single) SRS_EXPORT_LIBRTMP_SINGLE=${value} ;; --export-librtmp-single) SRS_EXPORT_LIBRTMP_SINGLE=${value} ;;
--detect-sendmmsg) if [[ $value == off ]]; then SRS_DETECT_SENDMMSG=NO; else SRS_DETECT_SENDMMSG=YES; fi ;;
--has-sendmmsg) if [[ $value == off ]]; then SRS_HAS_SENDMMSG=NO; else SRS_HAS_SENDMMSG=YES; fi ;;
--sendmmsg) if [[ $value == off ]]; then SRS_SENDMMSG=NO; else SRS_SENDMMSG=YES; fi ;; --sendmmsg) if [[ $value == off ]]; then SRS_SENDMMSG=NO; else SRS_SENDMMSG=YES; fi ;;
--without-srtp-nasm) SRS_SRTP_ASM=NO ;; --without-srtp-nasm) SRS_SRTP_ASM=NO ;;
@ -613,30 +609,10 @@ function apply_user_detail_options() {
SRS_SRTP_ASM=NO SRS_SRTP_ASM=NO
fi fi
# Detect whether has sendmmsg. if [[ $SRS_OSX == YES && $SRS_SENDMMSG == YES ]]; then
# @see http://man7.org/linux/man-pages/man2/sendmmsg.2.html echo "Disable sendmmsg for OSX"
if [[ $SRS_DETECT_SENDMMSG == YES ]]; then
mkdir -p ${SRS_OBJS} &&
echo " #include <sys/socket.h> " > ${SRS_OBJS}/_tmp_sendmmsg_detect.c
echo " int main(int argc, char** argv) { " >> ${SRS_OBJS}/_tmp_sendmmsg_detect.c
echo " struct mmsghdr hdr; " >> ${SRS_OBJS}/_tmp_sendmmsg_detect.c
echo " hdr.msg_len = 0; " >> ${SRS_OBJS}/_tmp_sendmmsg_detect.c
echo " return 0; " >> ${SRS_OBJS}/_tmp_sendmmsg_detect.c
echo " } " >> ${SRS_OBJS}/_tmp_sendmmsg_detect.c
${SRS_TOOL_CC} -c ${SRS_OBJS}/_tmp_sendmmsg_detect.c -D_GNU_SOURCE -o /dev/null >/dev/null 2>&1
ret=$?; rm -f ${SRS_OBJS}/_tmp_sendmmsg_detect.c;
if [[ $ret -ne 0 ]]; then
SRS_HAS_SENDMMSG=NO
fi
fi
# If system has no sendmmsg, disable sendmmsg.
if [[ $SRS_HAS_SENDMMSG == NO ]]; then
if [[ $SRS_SENDMMSG == YES ]]; then
echo "Disable UDP sendmmsg automatically"
SRS_SENDMMSG=NO SRS_SENDMMSG=NO
fi fi
fi
} }
apply_user_detail_options apply_user_detail_options
@ -666,8 +642,6 @@ function regenerate_options() {
if [ $SRS_NASM = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --nasm=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --nasm=off"; fi if [ $SRS_NASM = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --nasm=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --nasm=off"; fi
if [ $SRS_SRTP_ASM = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --srtp-nasm=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --srtp-nasm=off"; fi if [ $SRS_SRTP_ASM = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --srtp-nasm=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --srtp-nasm=off"; fi
if [ $SRS_SENDMMSG = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --sendmmsg=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --sendmmsg=off"; fi if [ $SRS_SENDMMSG = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --sendmmsg=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --sendmmsg=off"; fi
if [ $SRS_DETECT_SENDMMSG = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --detect-sendmmsg=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --detect-sendmmsg=off"; fi
if [ $SRS_HAS_SENDMMSG = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --has-sendmmsg=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --has-sendmmsg=off"; fi
if [ $SRS_CLEAN = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --clean=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --clean=off"; fi if [ $SRS_CLEAN = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --clean=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --clean=off"; fi
if [ $SRS_GPERF = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --gperf=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --gperf=off"; fi if [ $SRS_GPERF = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --gperf=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --gperf=off"; fi
if [ $SRS_GPERF_MC = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --gmc=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --gmc=off"; fi if [ $SRS_GPERF_MC = YES ]; then SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --gmc=on"; else SRS_AUTO_CONFIGURE="${SRS_AUTO_CONFIGURE} --gmc=off"; fi

2
trunk/configure vendored
View file

@ -188,7 +188,7 @@ if [[ $SRS_GCOV == YES ]]; then
SrsLinkOptions="${SrsLinkOptions} ${SrsGcov}"; SrsLinkOptions="${SrsLinkOptions} ${SrsGcov}";
fi fi
# For FFMPEG/RTC. # For FFMPEG/RTC.
if [[ $SRS_RTC == YES && $SRS_NASM == NO ]]; then if [[ $SRS_RTC == YES && $SRS_NASM == NO && $SRS_OSX == NO ]]; then
SrsLinkOptions="${SrsLinkOptions} -lrt"; SrsLinkOptions="${SrsLinkOptions} -lrt";
fi fi

6
trunk/research/players/rtc_player.html
View file

@ -46,7 +46,7 @@
</div> </div>
<label></label> <label></label>
<video id="rtc_media_player" controls autoplay></video> <video id="rtc_media_player" width="320" controls autoplay></video>
<footer> <footer>
<p></p> <p></p>
@ -112,6 +112,9 @@
contentType:'application/json', dataType: 'json' contentType:'application/json', dataType: 'json'
}).done(function(data) { }).done(function(data) {
console.log("Got answer: ", data); console.log("Got answer: ", data);
if (data.code) {
reject(data); return;
}
resolve(data.sdp); resolve(data.sdp);
}).fail(function(reason){ }).fail(function(reason){
reject(reason); reject(reason);
@ -120,6 +123,7 @@
}).then(function(answer) { }).then(function(answer) {
return pc.setRemoteDescription(new RTCSessionDescription({type: 'answer', sdp: answer})); return pc.setRemoteDescription(new RTCSessionDescription({type: 'answer', sdp: answer}));
}).catch(function(reason) { }).catch(function(reason) {
pc.close(); $('#rtc_media_player').hide();
throw reason; throw reason;
}); });
}; };

17
trunk/research/players/rtc_publisher.html
View file

@ -46,7 +46,7 @@
</div> </div>
<label></label> <label></label>
<video id="rtc_media_player" autoplay muted></video> <video id="rtc_media_player" width="320" autoplay muted></video>
<footer> <footer>
<p></p> <p></p>
@ -71,10 +71,8 @@
pc.addTransceiver("video", {direction: "sendonly"}); pc.addTransceiver("video", {direction: "sendonly"});
var constraints = { var constraints = {
"audio": true, "video": { audio: true, video: {
"width": { "min": "480", "max": "1920" }, height: { max: 320 }
"height": { "min": "320", "max": "1080" },
"frameRate": { "min": "15", "max": "60" }
} }
}; };
navigator.mediaDevices.getUserMedia( navigator.mediaDevices.getUserMedia(
@ -124,6 +122,9 @@
contentType:'application/json', dataType: 'json' contentType:'application/json', dataType: 'json'
}).done(function(data) { }).done(function(data) {
console.log("Got answer: ", data); console.log("Got answer: ", data);
if (data.code) {
reject(data); return;
}
resolve(data.sdp); resolve(data.sdp);
}).fail(function(reason){ }).fail(function(reason){
reject(reason); reject(reason);
@ -132,6 +133,12 @@
}).then(function(answer) { }).then(function(answer) {
return pc.setRemoteDescription(new RTCSessionDescription({type: 'answer', sdp: answer})); return pc.setRemoteDescription(new RTCSessionDescription({type: 'answer', sdp: answer}));
}).catch(function(reason) { }).catch(function(reason) {
pc.getLocalStreams().forEach(function(stream){
stream.getTracks().forEach(function(track) {
track.stop();
});
});
pc.close(); $('#rtc_media_player').hide();
throw reason; throw reason;
}); });
}; };

4
trunk/src/app/srs_app_config.cpp
View file

@ -4679,7 +4679,7 @@ srs_utime_t SrsConfig::get_stream_caster_gb28181_sip_query_catalog_interval(SrsC
return (srs_utime_t)(::atoi(conf->arg0().c_str()) * SRS_UTIME_SECONDS); return (srs_utime_t)(::atoi(conf->arg0().c_str()) * SRS_UTIME_SECONDS);
} }
int SrsConfig::get_rtc_server_enabled() bool SrsConfig::get_rtc_server_enabled()
{ {
SrsConfDirective* conf = root->get("rtc_server"); SrsConfDirective* conf = root->get("rtc_server");
return get_rtc_server_enabled(conf); return get_rtc_server_enabled(conf);
@ -4781,7 +4781,7 @@ bool SrsConfig::get_rtc_server_encrypt()
int SrsConfig::get_rtc_server_sendmmsg() int SrsConfig::get_rtc_server_sendmmsg()
{ {
#if !defined(SRS_HAS_SENDMMSG) || !defined(SRS_SENDMMSG) #if !defined(SRS_SENDMMSG)
return 1; return 1;
#else #else
static int DEFAULT = 256; static int DEFAULT = 256;

2
trunk/src/app/srs_app_config.hpp
View file

@ -521,7 +521,7 @@ public:
// rtc section // rtc section
public: public:
virtual int get_rtc_server_enabled(); virtual bool get_rtc_server_enabled();
virtual bool get_rtc_server_enabled(SrsConfDirective* conf); virtual bool get_rtc_server_enabled(SrsConfDirective* conf);
virtual int get_rtc_server_listen(); virtual int get_rtc_server_listen();
virtual std::string get_rtc_server_candidates(); virtual std::string get_rtc_server_candidates();

119
trunk/src/app/srs_app_http_api.cpp
View file

@ -26,6 +26,7 @@
#include <sstream> #include <sstream>
#include <stdlib.h> #include <stdlib.h>
#include <signal.h> #include <signal.h>
#include <unistd.h>
using namespace std; using namespace std;
#include <srs_kernel_log.hpp> #include <srs_kernel_log.hpp>
@ -788,9 +789,9 @@ srs_error_t SrsGoApiStreams::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessa
#ifdef SRS_RTC #ifdef SRS_RTC
uint32_t SrsGoApiRtcPlay::ssrc_num = 0; uint32_t SrsGoApiRtcPlay::ssrc_num = 0;
SrsGoApiRtcPlay::SrsGoApiRtcPlay(SrsRtcServer* rtc_svr) SrsGoApiRtcPlay::SrsGoApiRtcPlay(SrsRtcServer* server)
{ {
rtc_server = rtc_svr; server_ = server;
} }
SrsGoApiRtcPlay::~SrsGoApiRtcPlay() SrsGoApiRtcPlay::~SrsGoApiRtcPlay()
@ -917,16 +918,26 @@ srs_error_t SrsGoApiRtcPlay::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMe
request.vhost = parsed_vhost->arg0(); request.vhost = parsed_vhost->arg0();
} }
// TODO: FIXME: Maybe need a better name? // Whether enabled.
bool server_enabled = _srs_config->get_rtc_server_enabled();
bool rtc_enabled = _srs_config->get_rtc_enabled(request.vhost);
if (server_enabled && !rtc_enabled) {
srs_warn("RTC disabled in vhost %s", request.vhost.c_str());
}
if (!server_enabled || !rtc_enabled) {
return srs_error_new(ERROR_RTC_DISABLED, "Disabled server=%d, rtc=%d, vhost=%s",
server_enabled, rtc_enabled, request.vhost.c_str());
}
// TODO: FIXME: When server enabled, but vhost disabled, should report error. // TODO: FIXME: When server enabled, but vhost disabled, should report error.
SrsRtcSession* rtc_session = NULL; SrsRtcSession* session = NULL;
if ((err = rtc_server->create_rtc_session(&request, remote_sdp, local_sdp, eip, false, &rtc_session)) != srs_success) { if ((err = server_->create_session(&request, remote_sdp, local_sdp, eip, false, &session)) != srs_success) {
return srs_error_wrap(err, "create session"); return srs_error_wrap(err, "create session");
} }
if (encrypt.empty()) { if (encrypt.empty()) {
rtc_session->set_encrypt(_srs_config->get_rtc_server_encrypt()); session->set_encrypt(_srs_config->get_rtc_server_encrypt());
} else { } else {
rtc_session->set_encrypt(encrypt != "false"); session->set_encrypt(encrypt != "false");
} }
ostringstream os; ostringstream os;
@ -944,9 +955,9 @@ srs_error_t SrsGoApiRtcPlay::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMe
// TODO: add candidates in response json? // TODO: add candidates in response json?
res->set("sdp", SrsJsonAny::str(local_sdp_str.c_str())); res->set("sdp", SrsJsonAny::str(local_sdp_str.c_str()));
res->set("sessionid", SrsJsonAny::str(rtc_session->id().c_str())); res->set("sessionid", SrsJsonAny::str(session->id().c_str()));
srs_trace("RTC sid=%s, offer=%dB, answer=%dB", rtc_session->id().c_str(), remote_sdp_str.length(), local_sdp_str.length()); srs_trace("RTC sid=%s, offer=%dB, answer=%dB", session->id().c_str(), remote_sdp_str.length(), local_sdp_str.length());
return err; return err;
} }
@ -994,7 +1005,7 @@ srs_error_t SrsGoApiRtcPlay::exchange_sdp(const std::string& app, const std::str
local_sdp.addrtype_ = "IP4"; local_sdp.addrtype_ = "IP4";
local_sdp.unicast_address_ = "0.0.0.0"; local_sdp.unicast_address_ = "0.0.0.0";
local_sdp.session_name_ = "live_play_session"; local_sdp.session_name_ = "SRSPlaySession";
local_sdp.msid_semantic_ = "WMS"; local_sdp.msid_semantic_ = "WMS";
local_sdp.msids_.push_back(app + "/" + stream); local_sdp.msids_.push_back(app + "/" + stream);
@ -1016,8 +1027,14 @@ srs_error_t SrsGoApiRtcPlay::exchange_sdp(const std::string& app, const std::str
// TODO: check opus format specific param // TODO: check opus format specific param
std::vector<SrsMediaPayloadType> payloads = remote_media_desc.find_media_with_encoding_name("opus"); std::vector<SrsMediaPayloadType> payloads = remote_media_desc.find_media_with_encoding_name("opus");
for (std::vector<SrsMediaPayloadType>::iterator iter = payloads.begin(); iter != payloads.end(); ++iter) { for (std::vector<SrsMediaPayloadType>::iterator iter = payloads.begin(); iter != payloads.end(); ++iter) {
// Only choose one match opus codec.
local_media_desc.payload_types_.push_back(*iter); local_media_desc.payload_types_.push_back(*iter);
SrsMediaPayloadType& payload_type = local_media_desc.payload_types_.back();
// TODO: FIXME: Only support some transport algorithms.
vector<string> rtcp_fb;
payload_type.rtcp_fb_.swap(rtcp_fb);
// Only choose one match opus codec.
break; break;
} }
@ -1039,8 +1056,14 @@ srs_error_t SrsGoApiRtcPlay::exchange_sdp(const std::string& app, const std::str
// Try to pick the "best match" H.264 payload type. // Try to pick the "best match" H.264 payload type.
if (h264_param.packetization_mode == "1" && h264_param.level_asymmerty_allow == "1") { if (h264_param.packetization_mode == "1" && h264_param.level_asymmerty_allow == "1") {
// Only choose first match H.264 payload type.
local_media_desc.payload_types_.push_back(*iter); local_media_desc.payload_types_.push_back(*iter);
SrsMediaPayloadType& payload_type = local_media_desc.payload_types_.back();
// TODO: FIXME: Only support some transport algorithms.
vector<string> rtcp_fb;
payload_type.rtcp_fb_.swap(rtcp_fb);
// Only choose first match H.264 payload type.
break; break;
} }
@ -1088,6 +1111,11 @@ srs_error_t SrsGoApiRtcPlay::exchange_sdp(const std::string& app, const std::str
local_media_desc.rtcp_mux_ = true; local_media_desc.rtcp_mux_ = true;
local_media_desc.rtcp_rsize_ = true; local_media_desc.rtcp_rsize_ = true;
// TODO: FIXME: Avoid SSRC collision.
if (!ssrc_num) {
ssrc_num = ::getpid() * 10000 + ::getpid() * 100 + ::getpid();
}
if (local_media_desc.sendonly_ || local_media_desc.sendrecv_) { if (local_media_desc.sendonly_ || local_media_desc.sendrecv_) {
SrsSSRCInfo ssrc_info; SrsSSRCInfo ssrc_info;
ssrc_info.ssrc_ = ++ssrc_num; ssrc_info.ssrc_ = ++ssrc_num;
@ -1102,9 +1130,9 @@ srs_error_t SrsGoApiRtcPlay::exchange_sdp(const std::string& app, const std::str
uint32_t SrsGoApiRtcPublish::ssrc_num = 0; uint32_t SrsGoApiRtcPublish::ssrc_num = 0;
SrsGoApiRtcPublish::SrsGoApiRtcPublish(SrsRtcServer* rtc_svr) SrsGoApiRtcPublish::SrsGoApiRtcPublish(SrsRtcServer* server)
{ {
rtc_server = rtc_svr; server_ = server;
} }
SrsGoApiRtcPublish::~SrsGoApiRtcPublish() SrsGoApiRtcPublish::~SrsGoApiRtcPublish()
@ -1226,10 +1254,20 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
request.vhost = parsed_vhost->arg0(); request.vhost = parsed_vhost->arg0();
} }
// TODO: FIXME: Maybe need a better name? // Whether enabled.
bool server_enabled = _srs_config->get_rtc_server_enabled();
bool rtc_enabled = _srs_config->get_rtc_enabled(request.vhost);
if (server_enabled && !rtc_enabled) {
srs_warn("RTC disabled in vhost %s", request.vhost.c_str());
}
if (!server_enabled || !rtc_enabled) {
return srs_error_new(ERROR_RTC_DISABLED, "Disabled server=%d, rtc=%d, vhost=%s",
server_enabled, rtc_enabled, request.vhost.c_str());
}
// TODO: FIXME: When server enabled, but vhost disabled, should report error. // TODO: FIXME: When server enabled, but vhost disabled, should report error.
SrsRtcSession* rtc_session = NULL; SrsRtcSession* session = NULL;
if ((err = rtc_server->create_rtc_session(&request, remote_sdp, local_sdp, eip, true, &rtc_session)) != srs_success) { if ((err = server_->create_session(&request, remote_sdp, local_sdp, eip, true, &session)) != srs_success) {
return srs_error_wrap(err, "create session"); return srs_error_wrap(err, "create session");
} }
@ -1240,7 +1278,7 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
string local_sdp_str = os.str(); string local_sdp_str = os.str();
srs_trace("local_sdp=%s", local_sdp_str.c_str()); srs_verbose("local_sdp=%s", local_sdp_str.c_str());
res->set("code", SrsJsonAny::integer(ERROR_SUCCESS)); res->set("code", SrsJsonAny::integer(ERROR_SUCCESS));
res->set("server", SrsJsonAny::integer(SrsStatistic::instance()->server_id())); res->set("server", SrsJsonAny::integer(SrsStatistic::instance()->server_id()));
@ -1248,9 +1286,9 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
// TODO: add candidates in response json? // TODO: add candidates in response json?
res->set("sdp", SrsJsonAny::str(local_sdp_str.c_str())); res->set("sdp", SrsJsonAny::str(local_sdp_str.c_str()));
res->set("sessionid", SrsJsonAny::str(rtc_session->id().c_str())); res->set("sessionid", SrsJsonAny::str(session->id().c_str()));
srs_trace("RTC sid=%s, offer=%dB, answer=%dB", rtc_session->id().c_str(), remote_sdp_str.length(), local_sdp_str.length()); srs_trace("RTC sid=%s, offer=%dB, answer=%dB", session->id().c_str(), remote_sdp_str.length(), local_sdp_str.length());
return err; return err;
} }
@ -1298,7 +1336,7 @@ srs_error_t SrsGoApiRtcPublish::exchange_sdp(const std::string& app, const std::
local_sdp.addrtype_ = "IP4"; local_sdp.addrtype_ = "IP4";
local_sdp.unicast_address_ = "0.0.0.0"; local_sdp.unicast_address_ = "0.0.0.0";
local_sdp.session_name_ = "live_publish_session"; local_sdp.session_name_ = "SRSPublishSession";
local_sdp.msid_semantic_ = "WMS"; local_sdp.msid_semantic_ = "WMS";
local_sdp.msids_.push_back(app + "/" + stream); local_sdp.msids_.push_back(app + "/" + stream);
@ -1320,8 +1358,14 @@ srs_error_t SrsGoApiRtcPublish::exchange_sdp(const std::string& app, const std::
// TODO: check opus format specific param // TODO: check opus format specific param
std::vector<SrsMediaPayloadType> payloads = remote_media_desc.find_media_with_encoding_name("opus"); std::vector<SrsMediaPayloadType> payloads = remote_media_desc.find_media_with_encoding_name("opus");
for (std::vector<SrsMediaPayloadType>::iterator iter = payloads.begin(); iter != payloads.end(); ++iter) { for (std::vector<SrsMediaPayloadType>::iterator iter = payloads.begin(); iter != payloads.end(); ++iter) {
// Only choose one match opus codec.
local_media_desc.payload_types_.push_back(*iter); local_media_desc.payload_types_.push_back(*iter);
SrsMediaPayloadType& payload_type = local_media_desc.payload_types_.back();
// TODO: FIXME: Only support some transport algorithms.
vector<string> rtcp_fb;
payload_type.rtcp_fb_.swap(rtcp_fb);
// Only choose one match opus codec.
break; break;
} }
@ -1344,8 +1388,14 @@ srs_error_t SrsGoApiRtcPublish::exchange_sdp(const std::string& app, const std::
// Try to pick the "best match" H.264 payload type. // Try to pick the "best match" H.264 payload type.
if (h264_param.packetization_mode == "1" && h264_param.level_asymmerty_allow == "1") { if (h264_param.packetization_mode == "1" && h264_param.level_asymmerty_allow == "1") {
// Only choose first match H.264 payload type.
local_media_desc.payload_types_.push_back(*iter); local_media_desc.payload_types_.push_back(*iter);
SrsMediaPayloadType& payload_type = local_media_desc.payload_types_.back();
// TODO: FIXME: Only support some transport algorithms.
vector<string> rtcp_fb;
payload_type.rtcp_fb_.swap(rtcp_fb);
// Only choose first match H.264 payload type.
break; break;
} }
@ -1362,7 +1412,8 @@ srs_error_t SrsGoApiRtcPublish::exchange_sdp(const std::string& app, const std::
return srs_error_new(ERROR_RTC_SDP_EXCHANGE, "no found valid H.264 payload type"); return srs_error_new(ERROR_RTC_SDP_EXCHANGE, "no found valid H.264 payload type");
} }
local_media_desc.payload_types_.back().rtcp_fb_.push_back("rrtr"); // TODO: FIXME: Support RRTR?
//local_media_desc.payload_types_.back().rtcp_fb_.push_back("rrtr");
} }
local_media_desc.mid_ = remote_media_desc.mid_; local_media_desc.mid_ = remote_media_desc.mid_;
@ -1384,24 +1435,12 @@ srs_error_t SrsGoApiRtcPublish::exchange_sdp(const std::string& app, const std::
local_media_desc.session_info_.setup_ = "passive"; local_media_desc.session_info_.setup_ = "passive";
} }
local_sdp.media_descs_.back().session_info_.ice_options_ = "trickle";
if (remote_media_desc.sendonly_) {
local_media_desc.recvonly_ = true;
} else if (remote_media_desc.recvonly_) {
local_media_desc.sendonly_ = true;
} else if (remote_media_desc.sendrecv_) {
local_media_desc.sendrecv_ = true;
}
local_media_desc.rtcp_mux_ = true; local_media_desc.rtcp_mux_ = true;
if (local_media_desc.recvonly_ || local_media_desc.sendrecv_) { // For publisher, we are always sendonly.
SrsSSRCInfo ssrc_info; local_media_desc.sendonly_ = false;
ssrc_info.ssrc_ = ++ssrc_num; local_media_desc.recvonly_ = true;
ssrc_info.cname_ = "test_sdp_cname"; local_media_desc.sendrecv_ = false;
local_media_desc.ssrc_infos_.push_back(ssrc_info);
}
} }
return err; return err;

8
trunk/src/app/srs_app_http_api.hpp
View file

@ -173,9 +173,9 @@ class SrsGoApiRtcPlay : public ISrsHttpHandler
public: public:
static uint32_t ssrc_num; static uint32_t ssrc_num;
private: private:
SrsRtcServer* rtc_server; SrsRtcServer* server_;
public: public:
SrsGoApiRtcPlay(SrsRtcServer* rtc_svr); SrsGoApiRtcPlay(SrsRtcServer* server);
virtual ~SrsGoApiRtcPlay(); virtual ~SrsGoApiRtcPlay();
public: public:
virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
@ -190,9 +190,9 @@ class SrsGoApiRtcPublish : public ISrsHttpHandler
public: public:
static uint32_t ssrc_num; static uint32_t ssrc_num;
private: private:
SrsRtcServer* rtc_server; SrsRtcServer* server_;
public: public:
SrsGoApiRtcPublish(SrsRtcServer* rtc_svr); SrsGoApiRtcPublish(SrsRtcServer* server);
virtual ~SrsGoApiRtcPublish(); virtual ~SrsGoApiRtcPublish();
public: public:
virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);

15
trunk/src/app/srs_app_http_stream.cpp
View file

@ -128,10 +128,13 @@ srs_error_t SrsBufferCache::cycle()
// the stream cache will create consumer to cache stream, // the stream cache will create consumer to cache stream,
// which will trigger to fetch stream from origin for edge. // which will trigger to fetch stream from origin for edge.
SrsConsumer* consumer = NULL; SrsConsumer* consumer = NULL;
if ((err = source->create_consumer(NULL, consumer, false, false, true)) != srs_success) { SrsAutoFree(SrsConsumer, consumer);
if ((err = source->create_consumer(NULL, consumer)) != srs_success) {
return srs_error_wrap(err, "create consumer"); return srs_error_wrap(err, "create consumer");
} }
SrsAutoFree(SrsConsumer, consumer); if ((err = source->consumer_dumps(consumer, false, false, true)) != srs_success) {
return srs_error_wrap(err, "dumps consumer");
}
SrsPithyPrint* pprint = SrsPithyPrint::create_http_stream_cache(); SrsPithyPrint* pprint = SrsPithyPrint::create_http_stream_cache();
SrsAutoFree(SrsPithyPrint, pprint); SrsAutoFree(SrsPithyPrint, pprint);
@ -583,11 +586,13 @@ srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess
// create consumer of souce, ignore gop cache, use the audio gop cache. // create consumer of souce, ignore gop cache, use the audio gop cache.
SrsConsumer* consumer = NULL; SrsConsumer* consumer = NULL;
if ((err = source->create_consumer(NULL, consumer, true, true, !enc->has_cache())) != srs_success) { SrsAutoFree(SrsConsumer, consumer);
if ((err = source->create_consumer(NULL, consumer)) != srs_success) {
return srs_error_wrap(err, "create consumer"); return srs_error_wrap(err, "create consumer");
} }
SrsAutoFree(SrsConsumer, consumer); if ((err = source->consumer_dumps(consumer, true, true, !enc->has_cache())) != srs_success) {
srs_verbose("http: consumer created success."); return srs_error_wrap(err, "dumps consumer");
}
SrsPithyPrint* pprint = SrsPithyPrint::create_http_stream(); SrsPithyPrint* pprint = SrsPithyPrint::create_http_stream();
SrsAutoFree(SrsPithyPrint, pprint); SrsAutoFree(SrsPithyPrint, pprint);

4
trunk/src/app/srs_app_listener.hpp
View file

@ -138,9 +138,9 @@ public:
virtual ~ISrsUdpSender(); virtual ~ISrsUdpSender();
public: public:
// Fetch a mmsghdr from sender's cache. // Fetch a mmsghdr from sender's cache.
virtual srs_error_t fetch(mmsghdr** pphdr) = 0; virtual srs_error_t fetch(srs_mmsghdr** pphdr) = 0;
// Notify the sender to send out the msg. // Notify the sender to send out the msg.
virtual srs_error_t sendmmsg(mmsghdr* hdr) = 0; virtual srs_error_t sendmmsg(srs_mmsghdr* hdr) = 0;
// Whether sender exceed the max queue, that is, overflow. // Whether sender exceed the max queue, that is, overflow.
virtual bool overflow() = 0; virtual bool overflow() = 0;
// Set the queue extra ratio, for example, when mw_msgs > 0, we need larger queue. // Set the queue extra ratio, for example, when mw_msgs > 0, we need larger queue.

514
trunk/src/app/srs_app_rtc_conn.cpp
View file

File diff suppressed because it is too large Load diff

97
trunk/src/app/srs_app_rtc_conn.hpp
View file

@ -53,11 +53,12 @@ class SrsSource;
class SrsRtpPacket2; class SrsRtpPacket2;
class ISrsUdpSender; class ISrsUdpSender;
class SrsRtpQueue; class SrsRtpQueue;
class SrsRtpH264Demuxer; class SrsRtpAudioQueue;
class SrsRtpOpusDemuxer; class SrsRtpVideoQueue;
class SrsRtpPacket2; class SrsRtpPacket2;
class ISrsCodec; class ISrsCodec;
class SrsRtpNackForReceiver; class SrsRtpNackForReceiver;
class SrsRtpIncommingVideoFrame;
const uint8_t kSR = 200; const uint8_t kSR = 200;
const uint8_t kRR = 201; const uint8_t kRR = 201;
@ -103,10 +104,10 @@ enum SrsRtcSessionStateType
CLOSED = 4, CLOSED = 4,
}; };
class SrsDtlsSession class SrsRtcDtls
{ {
private: private:
SrsRtcSession* rtc_session; SrsRtcSession* session_;
SSL* dtls; SSL* dtls;
BIO* bio_in; BIO* bio_in;
@ -121,8 +122,8 @@ private:
bool handshake_done; bool handshake_done;
public: public:
SrsDtlsSession(SrsRtcSession* s); SrsRtcDtls(SrsRtcSession* s);
virtual ~SrsDtlsSession(); virtual ~SrsRtcDtls();
srs_error_t initialize(SrsRequest* r); srs_error_t initialize(SrsRequest* r);
@ -143,8 +144,8 @@ private:
srs_error_t srtp_recv_init(); srs_error_t srtp_recv_init();
}; };
// A group of RTP packets. // A group of RTP packets for outgoing(send to players).
class SrsRtcPackets class SrsRtcOutgoingPackets
{ {
public: public:
bool use_gso; bool use_gso;
@ -183,8 +184,8 @@ private:
int nn_cache; int nn_cache;
SrsRtpPacket2* cache; SrsRtpPacket2* cache;
public: public:
SrsRtcPackets(int nn_cache_max); SrsRtcOutgoingPackets(int nn_cache_max);
virtual ~SrsRtcPackets(); virtual ~SrsRtcOutgoingPackets();
public: public:
void reset(bool gso, bool merge_nalus); void reset(bool gso, bool merge_nalus);
SrsRtpPacket2* fetch(); SrsRtpPacket2* fetch();
@ -194,14 +195,13 @@ public:
SrsRtpPacket2* at(int index); SrsRtpPacket2* at(int index);
}; };
// TODO: FIXME: Rename to RTC player or subscriber. class SrsRtcPlayer : virtual public ISrsCoroutineHandler, virtual public ISrsReloadHandler
class SrsRtcSenderThread : virtual public ISrsCoroutineHandler, virtual public ISrsReloadHandler
{ {
protected: protected:
SrsCoroutine* trd; SrsCoroutine* trd;
int _parent_cid; int _parent_cid;
private: private:
SrsRtcSession* rtc_session; SrsRtcSession* session_;
uint32_t video_ssrc; uint32_t video_ssrc;
uint32_t audio_ssrc; uint32_t audio_ssrc;
uint16_t video_payload_type; uint16_t video_payload_type;
@ -221,8 +221,8 @@ private:
int mw_msgs; int mw_msgs;
bool realtime; bool realtime;
public: public:
SrsRtcSenderThread(SrsRtcSession* s, int parent_cid); SrsRtcPlayer(SrsRtcSession* s, int parent_cid);
virtual ~SrsRtcSenderThread(); virtual ~SrsRtcPlayer();
public: public:
srs_error_t initialize(const uint32_t& vssrc, const uint32_t& assrc, const uint16_t& v_pt, const uint16_t& a_pt); srs_error_t initialize(const uint32_t& vssrc, const uint32_t& assrc, const uint16_t& v_pt, const uint16_t& a_pt);
// interface ISrsReloadHandler // interface ISrsReloadHandler
@ -239,17 +239,17 @@ public:
public: public:
virtual srs_error_t cycle(); virtual srs_error_t cycle();
private: private:
srs_error_t send_messages(SrsSource* source, SrsSharedPtrMessage** msgs, int nb_msgs, SrsRtcPackets& packets); srs_error_t send_messages(SrsSource* source, SrsSharedPtrMessage** msgs, int nb_msgs, SrsRtcOutgoingPackets& packets);
srs_error_t messages_to_packets(SrsSource* source, SrsSharedPtrMessage** msgs, int nb_msgs, SrsRtcPackets& packets); srs_error_t messages_to_packets(SrsSource* source, SrsSharedPtrMessage** msgs, int nb_msgs, SrsRtcOutgoingPackets& packets);
srs_error_t send_packets(SrsRtcPackets& packets); srs_error_t send_packets(SrsRtcOutgoingPackets& packets);
srs_error_t send_packets_gso(SrsRtcPackets& packets); srs_error_t send_packets_gso(SrsRtcOutgoingPackets& packets);
private: private:
srs_error_t package_opus(SrsSample* sample, SrsRtcPackets& packets, int nn_max_payload); srs_error_t package_opus(SrsSample* sample, SrsRtcOutgoingPackets& packets, int nn_max_payload);
private: private:
srs_error_t package_fu_a(SrsSharedPtrMessage* msg, SrsSample* sample, int fu_payload_size, SrsRtcPackets& packets); srs_error_t package_fu_a(SrsSharedPtrMessage* msg, SrsSample* sample, int fu_payload_size, SrsRtcOutgoingPackets& packets);
srs_error_t package_nalus(SrsSharedPtrMessage* msg, SrsRtcPackets& packets); srs_error_t package_nalus(SrsSharedPtrMessage* msg, SrsRtcOutgoingPackets& packets);
srs_error_t package_single_nalu(SrsSharedPtrMessage* msg, SrsSample* sample, SrsRtcPackets& packets); srs_error_t package_single_nalu(SrsSharedPtrMessage* msg, SrsSample* sample, SrsRtcOutgoingPackets& packets);
srs_error_t package_stap_a(SrsSource* source, SrsSharedPtrMessage* msg, SrsRtcPackets& packets); srs_error_t package_stap_a(SrsSource* source, SrsSharedPtrMessage* msg, SrsRtcOutgoingPackets& packets);
}; };
class SrsRtcPublisher : virtual public ISrsHourGlass, virtual public ISrsRtpPacketDecodeHandler class SrsRtcPublisher : virtual public ISrsHourGlass, virtual public ISrsRtpPacketDecodeHandler
@ -257,13 +257,13 @@ class SrsRtcPublisher : virtual public ISrsHourGlass, virtual public ISrsRtpPack
private: private:
SrsHourGlass* report_timer; SrsHourGlass* report_timer;
private: private:
SrsRtcSession* rtc_session; SrsRtcSession* session_;
uint32_t video_ssrc; uint32_t video_ssrc;
uint32_t audio_ssrc; uint32_t audio_ssrc;
private: private:
SrsRtpQueue* video_queue_; SrsRtpVideoQueue* video_queue_;
SrsRtpNackForReceiver* video_nack_; SrsRtpNackForReceiver* video_nack_;
SrsRtpQueue* audio_queue_; SrsRtpAudioQueue* audio_queue_;
SrsRtpNackForReceiver* audio_nack_; SrsRtpNackForReceiver* audio_nack_;
private: private:
SrsRequest* req; SrsRequest* req;
@ -288,11 +288,9 @@ public:
virtual void on_before_decode_payload(SrsRtpPacket2* pkt, SrsBuffer* buf, ISrsCodec** ppayload); virtual void on_before_decode_payload(SrsRtpPacket2* pkt, SrsBuffer* buf, ISrsCodec** ppayload);
private: private:
srs_error_t on_audio(SrsRtpPacket2* pkt); srs_error_t on_audio(SrsRtpPacket2* pkt);
srs_error_t collect_audio_frames(); srs_error_t on_audio_frame(SrsRtpPacket2* frame);
srs_error_t do_collect_audio_frame(SrsRtpPacket2* packet);
srs_error_t on_video(SrsRtpPacket2* pkt); srs_error_t on_video(SrsRtpPacket2* pkt);
srs_error_t collect_video_frames(); srs_error_t on_video_frame(SrsRtpPacket2* frame);
srs_error_t do_collect_video_frame(std::vector<SrsRtpPacket2*>& packets);
public: public:
void request_keyframe(); void request_keyframe();
// interface ISrsHourGlass // interface ISrsHourGlass
@ -302,15 +300,16 @@ public:
class SrsRtcSession class SrsRtcSession
{ {
friend class SrsDtlsSession; friend class SrsRtcDtls;
friend class SrsRtcSenderThread; friend class SrsRtcPlayer;
friend class SrsRtcPublisher; friend class SrsRtcPublisher;
private: private:
SrsRtcServer* rtc_server; SrsRtcServer* server_;
SrsRtcSessionStateType session_state; SrsRtcSessionStateType session_state;
SrsDtlsSession* dtls_session; SrsRtcDtls* dtls_;
SrsRtcSenderThread* sender; SrsRtcPlayer* player_;
SrsRtcPublisher* publisher; SrsRtcPublisher* publisher_;
bool is_publisher_;
private: private:
SrsUdpMuxSocket* sendonly_skt; SrsUdpMuxSocket* sendonly_skt;
std::string username; std::string username;
@ -328,7 +327,7 @@ private:
// TODO: FIXME: Support reload. // TODO: FIXME: Support reload.
bool encrypt; bool encrypt;
SrsRequest* req; SrsRequest* req;
SrsSource* source; SrsSource* source_;
SrsSdp remote_sdp; SrsSdp remote_sdp;
SrsSdp local_sdp; SrsSdp local_sdp;
private: private:
@ -336,7 +335,7 @@ private:
sockaddr_in* blackhole_addr; sockaddr_in* blackhole_addr;
srs_netfd_t blackhole_stfd; srs_netfd_t blackhole_stfd;
public: public:
SrsRtcSession(SrsRtcServer* s, SrsRequest* r, const std::string& un, int context_id); SrsRtcSession(SrsRtcServer* s);
virtual ~SrsRtcSession(); virtual ~SrsRtcSession();
public: public:
SrsSdp* get_local_sdp(); SrsSdp* get_local_sdp();
@ -352,7 +351,7 @@ public:
void switch_to_context(); void switch_to_context();
int context_id(); int context_id();
public: public:
srs_error_t initialize(); srs_error_t initialize(SrsSource* source, SrsRequest* r, bool is_publisher, const std::string& un, int context_id);
// The peer address may change, we can identify that by STUN messages. // The peer address may change, we can identify that by STUN messages.
srs_error_t on_stun(SrsUdpMuxSocket* skt, SrsStunPacket* r); srs_error_t on_stun(SrsUdpMuxSocket* skt, SrsStunPacket* r);
srs_error_t on_dtls(char* data, int nb_data); srs_error_t on_dtls(char* data, int nb_data);
@ -387,9 +386,9 @@ private:
private: private:
// Hotspot msgs, we are working on it. // Hotspot msgs, we are working on it.
// @remark We will wait util all messages are ready. // @remark We will wait util all messages are ready.
std::vector<mmsghdr> hotspot; std::vector<srs_mmsghdr> hotspot;
// Cache msgs, for other coroutines to fill it. // Cache msgs, for other coroutines to fill it.
std::vector<mmsghdr> cache; std::vector<srs_mmsghdr> cache;
int cache_pos; int cache_pos;
// The max number of messages for sendmmsg. If 1, we use sendmsg to send. // The max number of messages for sendmmsg. If 1, we use sendmsg to send.
int max_sendmmsg; int max_sendmmsg;
@ -404,10 +403,10 @@ public:
public: public:
virtual srs_error_t initialize(srs_netfd_t fd, int senders); virtual srs_error_t initialize(srs_netfd_t fd, int senders);
private: private:
void free_mhdrs(std::vector<mmsghdr>& mhdrs); void free_mhdrs(std::vector<srs_mmsghdr>& mhdrs);
public: public:
virtual srs_error_t fetch(mmsghdr** pphdr); virtual srs_error_t fetch(srs_mmsghdr** pphdr);
virtual srs_error_t sendmmsg(mmsghdr* hdr); virtual srs_error_t sendmmsg(srs_mmsghdr* hdr);
virtual bool overflow(); virtual bool overflow();
virtual void set_extra_ratio(int r); virtual void set_extra_ratio(int r);
public: public:
@ -438,16 +437,16 @@ public:
virtual srs_error_t on_udp_packet(SrsUdpMuxSocket* skt); virtual srs_error_t on_udp_packet(SrsUdpMuxSocket* skt);
public: public:
virtual srs_error_t listen_api(); virtual srs_error_t listen_api();
srs_error_t create_rtc_session( srs_error_t create_session(
SrsRequest* req, const SrsSdp& remote_sdp, SrsSdp& local_sdp, const std::string& mock_eip, bool publish, SrsRequest* req, const SrsSdp& remote_sdp, SrsSdp& local_sdp, const std::string& mock_eip, bool publish,
SrsRtcSession** psession SrsRtcSession** psession
); );
bool insert_into_id_sessions(const std::string& peer_id, SrsRtcSession* rtc_session); bool insert_into_id_sessions(const std::string& peer_id, SrsRtcSession* session);
void check_and_clean_timeout_session(); void check_and_clean_timeout_session();
int nn_sessions(); int nn_sessions();
private: private:
SrsRtcSession* find_rtc_session_by_username(const std::string& ufrag); SrsRtcSession* find_session_by_username(const std::string& ufrag);
SrsRtcSession* find_rtc_session_by_peer_id(const std::string& peer_id); SrsRtcSession* find_session_by_peer_id(const std::string& peer_id);
// interface ISrsHourGlass // interface ISrsHourGlass
public: public:
virtual srs_error_t notify(int type, srs_utime_t interval, srs_utime_t tick); virtual srs_error_t notify(int type, srs_utime_t interval, srs_utime_t tick);

5
trunk/src/app/srs_app_rtmp_conn.cpp
View file

@ -654,10 +654,13 @@ srs_error_t SrsRtmpConn::playing(SrsSource* source)
// Create a consumer of source. // Create a consumer of source.
SrsConsumer* consumer = NULL; SrsConsumer* consumer = NULL;
SrsAutoFree(SrsConsumer, consumer);
if ((err = source->create_consumer(this, consumer)) != srs_success) { if ((err = source->create_consumer(this, consumer)) != srs_success) {
return srs_error_wrap(err, "rtmp: create consumer"); return srs_error_wrap(err, "rtmp: create consumer");
} }
SrsAutoFree(SrsConsumer, consumer); if ((err = source->consumer_dumps(consumer)) != srs_success) {
return srs_error_wrap(err, "rtmp: dumps consumer");
}
// Use receiving thread to receive packets from peer. // Use receiving thread to receive packets from peer.
// @see: https://github.com/ossrs/srs/issues/217 // @see: https://github.com/ossrs/srs/issues/217

447
trunk/src/app/srs_app_rtp_queue.cpp
View file

@ -131,7 +131,7 @@ void SrsRtpNackForReceiver::update_rtt(int rtt)
SrsRtpRingBuffer::SrsRtpRingBuffer(int capacity) SrsRtpRingBuffer::SrsRtpRingBuffer(int capacity)
{ {
nn_seq_flip_backs = 0; nn_seq_flip_backs = 0;
high_ = low_ = 0; begin = end = 0;
capacity_ = (uint16_t)capacity; capacity_ = (uint16_t)capacity;
initialized_ = false; initialized_ = false;
@ -144,19 +144,21 @@ SrsRtpRingBuffer::~SrsRtpRingBuffer()
srs_freepa(queue_); srs_freepa(queue_);
} }
uint16_t SrsRtpRingBuffer::low() bool SrsRtpRingBuffer::empty()
{ {
return low_; return begin == end;
} }
uint16_t SrsRtpRingBuffer::high() int SrsRtpRingBuffer::size()
{ {
return high_; int size = srs_rtp_seq_distance(begin, end);
srs_assert(size >= 0);
return size;
} }
void SrsRtpRingBuffer::advance_to(uint16_t seq) void SrsRtpRingBuffer::advance_to(uint16_t seq)
{ {
low_ = seq; begin = seq;
} }
void SrsRtpRingBuffer::set(uint16_t at, SrsRtpPacket2* pkt) void SrsRtpRingBuffer::set(uint16_t at, SrsRtpPacket2* pkt)
@ -175,92 +177,54 @@ void SrsRtpRingBuffer::remove(uint16_t at)
set(at, NULL); set(at, NULL);
} }
void SrsRtpRingBuffer::reset(uint16_t low, uint16_t high) void SrsRtpRingBuffer::reset(uint16_t first, uint16_t last)
{ {
for (uint16_t s = low; s != high; ++s) { for (uint16_t s = first; s != last; ++s) {
queue_[s % capacity_] = NULL; queue_[s % capacity_] = NULL;
} }
} }
bool SrsRtpRingBuffer::overflow() bool SrsRtpRingBuffer::overflow()
{ {
return high_ - low_ >= capacity_; return srs_rtp_seq_distance(begin, end) >= capacity_;
}
bool SrsRtpRingBuffer::is_heavy()
{
return high_ - low_ >= capacity_ / 2;
}
uint16_t SrsRtpRingBuffer::next_start_of_frame()
{
if (low_ == high_) {
return low_;
}
for (uint16_t s = low_ + 1 ; s != high_; ++s) {
SrsRtpPacket2*& pkt = queue_[s % capacity_];
if (pkt && pkt->is_first_packet_of_frame) {
return s;
}
}
return low_;
}
uint16_t SrsRtpRingBuffer::next_keyframe()
{
if (low_ == high_) {
return low_;
}
for (uint16_t s = low_ + 1 ; s != high_; ++s) {
SrsRtpPacket2*& pkt = queue_[s % capacity_];
if (pkt && pkt->is_key_frame && pkt->is_first_packet_of_frame) {
return s;
}
}
return low_;
} }
uint32_t SrsRtpRingBuffer::get_extended_highest_sequence() uint32_t SrsRtpRingBuffer::get_extended_highest_sequence()
{ {
return nn_seq_flip_backs * 65536 + high_; return nn_seq_flip_backs * 65536 + end - 1;
} }
void SrsRtpRingBuffer::update(uint16_t seq, bool startup, uint16_t& nack_low, uint16_t& nack_high) void SrsRtpRingBuffer::update(uint16_t seq, uint16_t& nack_first, uint16_t& nack_last)
{ {
if (!initialized_) { if (!initialized_) {
initialized_ = true; initialized_ = true;
low_ = high_ = seq; begin = seq;
end = seq + 1;
return; return;
} }
// Normal sequence, seq follows high_. // Normal sequence, seq follows high_.
if (srs_rtp_seq_distance(high_, seq)) { if (srs_rtp_seq_distance(end, seq) >= 0) {
nack_low = high_ + 1; nack_first = end + 1;
nack_high = seq; nack_last = seq + 1;
// When distance(seq,high_)>0 and seq<high_, seq must flip back, // When distance(seq,high_)>0 and seq<high_, seq must flip back,
// for example, high_=65535, seq=1, distance(65535,1)>0 and 1<65535. // for example, high_=65535, seq=1, distance(65535,1)>0 and 1<65535.
if (seq < high_) { // TODO: FIXME: The first flip may be dropped.
if (seq < end) {
++nn_seq_flip_backs; ++nn_seq_flip_backs;
} }
high_ = seq; end = seq + 1;
return; return;
} }
// Out-of-order sequence, seq before low_. // Out-of-order sequence, seq before low_.
if (srs_rtp_seq_distance(seq, low_)) { if (srs_rtp_seq_distance(seq, begin) > 0) {
// When startup, we may receive packets in chaos order. // When startup, we may receive packets in chaos order.
// Because we don't know the ISN(initiazlie sequence number), the first packet // Because we don't know the ISN(initiazlie sequence number), the first packet
// we received maybe no the first packet client sent. // we received maybe no the first packet client sent.
if (startup) { // @remark We only log a warning, because it seems ok for publisher.
nack_low = seq + 1; srs_warn("too old seq %u, range [%u, %u]", seq, begin, end);
nack_high = low_;
low_ = seq;
}
} }
} }
@ -269,9 +233,8 @@ SrsRtpPacket2* SrsRtpRingBuffer::at(uint16_t seq)
return queue_[seq % capacity_]; return queue_[seq % capacity_];
} }
SrsRtpQueue::SrsRtpQueue(const char* tag, int capacity) SrsRtpQueue::SrsRtpQueue(int capacity)
{ {
nn_collected_frames = 0;
queue_ = new SrsRtpRingBuffer(capacity); queue_ = new SrsRtpRingBuffer(capacity);
jitter_ = 0; jitter_ = 0;
@ -282,9 +245,6 @@ SrsRtpQueue::SrsRtpQueue(const char* tag, int capacity)
num_of_packet_received_ = 0; num_of_packet_received_ = 0;
number_of_packet_lossed_ = 0; number_of_packet_lossed_ = 0;
request_key_frame_ = false;
tag_ = tag;
} }
SrsRtpQueue::~SrsRtpQueue() SrsRtpQueue::~SrsRtpQueue()
@ -327,89 +287,20 @@ srs_error_t SrsRtpQueue::consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt
// OK, we got one new RTP packet, which is not in NACK. // OK, we got one new RTP packet, which is not in NACK.
if (!nack_info) { if (!nack_info) {
++num_of_packet_received_; ++num_of_packet_received_;
uint16_t nack_low = 0, nack_high = 0; uint16_t nack_first = 0, nack_last = 0;
queue_->update(seq, !nn_collected_frames, nack_low, nack_high); queue_->update(seq, nack_first, nack_last);
if (srs_rtp_seq_distance(nack_low, nack_high)) { if (srs_rtp_seq_distance(nack_first, nack_last) > 0) {
srs_trace("%s update nack seq=%u, startup=%d, range [%u, %u]", tag_, seq, !nn_collected_frames, nack_low, nack_high); srs_trace("update seq=%u, nack range [%u, %u]", seq, nack_first, nack_last);
insert_into_nack_list(nack, nack_low, nack_high); insert_into_nack_list(nack, nack_first, nack_last);
} }
} }
// When packets overflow, collect frame and move head to next frame start.
if (queue_->overflow()) {
collect_packet(nack);
uint16_t next = queue_->next_start_of_frame();
// Note that low_ mean not found, clear queue util one packet.
if (next == queue_->low()) {
next = queue_->high() - 1;
}
srs_trace("%s seq out of range [%u, %u]", tag_, queue_->low(), next);
for (uint16_t s = queue_->low(); s != next; ++s) {
nack->remove(s);
queue_->remove(s);
}
srs_trace("%s force update seq %u to %u", tag_, queue_->low(), next + 1);
queue_->advance_to(next + 1);
}
// Save packet at the position seq. // Save packet at the position seq.
queue_->set(seq, pkt); queue_->set(seq, pkt);
return err; return err;
} }
void SrsRtpQueue::collect_frames(std::vector<std::vector<SrsRtpPacket2*> >& frames)
{
frames.swap(frames_);
}
bool SrsRtpQueue::should_request_key_frame()
{
if (request_key_frame_) {
request_key_frame_ = false;
return true;
}
return request_key_frame_;
}
void SrsRtpQueue::notify_drop_seq(uint16_t seq)
{
uint16_t next = queue_->next_start_of_frame();
// Note that low_ mean not found, clear queue util one packet.
if (next == queue_->low()) {
next = queue_->high() - 1;
}
// When NACK is timeout, move to the next start of frame.
srs_trace("%s nack drop seq=%u, drop range [%u, %u]", tag_, seq, queue_->low(), next + 1);
queue_->advance_to(next + 1);
}
void SrsRtpQueue::notify_nack_list_full()
{
uint16_t next = queue_->next_keyframe();
// Note that low_ mean not found, clear queue util one packet.
if (next == queue_->low()) {
next = queue_->high() - 1;
}
// When NACK is overflow, move to the next keyframe.
srs_trace("%s nack overflow drop range [%u, %u]", tag_, queue_->low(), next + 1);
queue_->advance_to(next + 1);
}
void SrsRtpQueue::request_keyframe()
{
request_key_frame_ = true;
}
uint32_t SrsRtpQueue::get_extended_highest_sequence() uint32_t SrsRtpQueue::get_extended_highest_sequence()
{ {
return queue_->get_extended_highest_sequence(); return queue_->get_extended_highest_sequence();
@ -439,9 +330,9 @@ uint32_t SrsRtpQueue::get_interarrival_jitter()
return static_cast<uint32_t>(jitter_); return static_cast<uint32_t>(jitter_);
} }
void SrsRtpQueue::insert_into_nack_list(SrsRtpNackForReceiver* nack, uint16_t seq_start, uint16_t seq_end) void SrsRtpQueue::insert_into_nack_list(SrsRtpNackForReceiver* nack, uint16_t first, uint16_t last)
{ {
for (uint16_t s = seq_start; s != seq_end; ++s) { for (uint16_t s = first; s != last; ++s) {
nack->insert(s); nack->insert(s);
++number_of_packet_lossed_; ++number_of_packet_lossed_;
} }
@ -449,7 +340,7 @@ void SrsRtpQueue::insert_into_nack_list(SrsRtpNackForReceiver* nack, uint16_t se
nack->check_queue_size(); nack->check_queue_size();
} }
SrsRtpAudioQueue::SrsRtpAudioQueue(int capacity) : SrsRtpQueue("audio", capacity) SrsRtpAudioQueue::SrsRtpAudioQueue(int capacity) : SrsRtpQueue(capacity)
{ {
} }
@ -457,114 +348,181 @@ SrsRtpAudioQueue::~SrsRtpAudioQueue()
{ {
} }
srs_error_t SrsRtpAudioQueue::consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt) void SrsRtpAudioQueue::notify_drop_seq(uint16_t seq)
{ {
srs_error_t err = srs_success; uint16_t next = seq + 1;
if (srs_rtp_seq_distance(queue_->end, seq) > 0) {
seq = queue_->end;
}
srs_trace("nack drop seq=%u, drop range [%u, %u, %u]", seq, queue_->begin, next, queue_->end);
if ((err = SrsRtpQueue::consume(nack, pkt)) != srs_success) { queue_->advance_to(next);
return srs_error_wrap(err, "audio queue");
} }
// For audio, always try to collect frame, because each packet is a frame. void SrsRtpAudioQueue::notify_nack_list_full()
collect_packet(nack); {
// TODO: FIXME: Maybe we should not drop all packets.
return err; queue_->advance_to(queue_->end);
} }
void SrsRtpAudioQueue::collect_packet(SrsRtpNackForReceiver* nack) void SrsRtpAudioQueue::collect_frames(SrsRtpNackForReceiver* nack, vector<SrsRtpPacket2*>& frames)
{ {
// When done, s point to the next available packet. // When done, next point to the next available packet.
uint16_t next = queue_->low(); uint16_t next = queue_->begin;
for (; next != queue_->end; ++next) {
for (; next != queue_->high(); ++next) {
SrsRtpPacket2* pkt = queue_->at(next); SrsRtpPacket2* pkt = queue_->at(next);
// Not found or in NACK, stop collecting frame. // Not found or in NACK, stop collecting frame.
if (!pkt || nack->find(next) != NULL) { if (!pkt || nack->find(next) != NULL) {
srs_trace("%s wait for nack seq=%u", tag_, next); srs_trace("wait for nack seq=%u", next);
break; break;
} }
// OK, collect packet to frame. frames.push_back(pkt);
vector<SrsRtpPacket2*> frame;
frame.push_back(pkt);
// Done, we got the last packet of frame.
nn_collected_frames++;
frames_.push_back(frame);
} }
if (queue_->low() != next) { if (next != queue_->begin) {
// Reset the range of packets to NULL in buffer. // Reset the range of packets to NULL in buffer.
queue_->reset(queue_->low(), next); queue_->reset(queue_->begin, next);
srs_verbose("%s collect on frame, update head seq=%u t %u", tag_, queue_->low(), next); srs_verbose("RTC collect audio [%u, %u, %u]", queue_->begin, next, queue_->end);
queue_->advance_to(next); queue_->advance_to(next);
} }
// For audio, if overflow, clear all packets.
if (queue_->overflow()) {
queue_->advance_to(queue_->end);
}
} }
SrsRtpVideoQueue::SrsRtpVideoQueue(int capacity) : SrsRtpQueue("video", capacity) SrsRtpVideoQueue::SrsRtpVideoQueue(int capacity) : SrsRtpQueue(capacity)
{ {
request_key_frame_ = false;
} }
SrsRtpVideoQueue::~SrsRtpVideoQueue() SrsRtpVideoQueue::~SrsRtpVideoQueue()
{ {
} }
void SrsRtpVideoQueue::notify_drop_seq(uint16_t seq)
{
// If not found start frame, return the end, and we will clear queue.
uint16_t next = next_start_of_frame(seq);
srs_trace("nack drop seq=%u, drop range [%u, %u, %u]", seq, queue_->begin, next, queue_->end);
queue_->advance_to(next);
}
void SrsRtpVideoQueue::notify_nack_list_full()
{
// If not found start frame, return the end, and we will clear queue.
uint16_t next = next_keyframe();
srs_trace("nack overflow, drop range [%u, %u, %u]", queue_->begin, next, queue_->end);
queue_->advance_to(next);
}
srs_error_t SrsRtpVideoQueue::consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt) srs_error_t SrsRtpVideoQueue::consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt)
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
if ((err = SrsRtpQueue::consume(nack, pkt)) != srs_success) { uint8_t v = (uint8_t)pkt->nalu_type;
return srs_error_wrap(err, "video queue"); if (v == kFuA) {
SrsRtpFUAPayload2* payload = dynamic_cast<SrsRtpFUAPayload2*>(pkt->payload);
if (!payload) {
srs_freep(pkt);
return srs_error_new(ERROR_RTC_RTP_MUXER, "FU-A payload");
} }
// Collect packets to frame when: pkt->video_is_first_packet = payload->start;
// 1. Marker bit means the last packet of frame received. pkt->video_is_last_packet = payload->end;
// 2. Queue has lots of packets, the load is heavy. pkt->video_is_idr = (payload->nalu_type == SrsAvcNaluTypeIDR);
// TODO: FIMXE: For real-time, we should collect each frame ASAP. } else {
if (pkt->rtp_header.get_marker() || queue_->is_heavy()) { pkt->video_is_first_packet = true;
collect_packet(nack); pkt->video_is_last_packet = true;
if (v == kStapA) {
pkt->video_is_idr = true;
} else {
pkt->video_is_idr = (pkt->nalu_type == SrsAvcNaluTypeIDR);
}
}
if ((err = SrsRtpQueue::consume(nack, pkt)) != srs_success) {
return srs_error_wrap(err, "video consume");
} }
return err; return err;
} }
void SrsRtpVideoQueue::collect_packet(SrsRtpNackForReceiver* nack) void SrsRtpVideoQueue::collect_frames(SrsRtpNackForReceiver* nack, std::vector<SrsRtpPacket2*>& frames)
{ {
while (queue_->low() != queue_->high()) { while (true) {
vector<SrsRtpPacket2*> frame; SrsRtpPacket2* pkt = NULL;
do_collect_packet(nack, frame); collect_frame(nack, &pkt);
if (frame.empty()) { if (!pkt) {
return; break;
} }
nn_collected_frames++; frames.push_back(pkt);
frames_.push_back(frame);
} }
if (queue_->overflow()) {
on_overflow(nack);
}
}
bool SrsRtpVideoQueue::should_request_key_frame()
{
if (request_key_frame_) {
request_key_frame_ = false;
return true;
}
return request_key_frame_;
}
void SrsRtpVideoQueue::request_keyframe()
{
request_key_frame_ = true;
}
void SrsRtpVideoQueue::on_overflow(SrsRtpNackForReceiver* nack)
{
// If not found start frame, return the end, and we will clear queue.
uint16_t next = next_start_of_frame(queue_->begin);
srs_trace("on overflow, remove range [%u, %u, %u]", queue_->begin, next, queue_->end);
for (uint16_t s = queue_->begin; s != next; ++s) {
nack->remove(s);
queue_->remove(s);
}
queue_->advance_to(next);
} }
// TODO: FIXME: Should refer to the FU-A original video frame, to avoid finding for each packet. // TODO: FIXME: Should refer to the FU-A original video frame, to avoid finding for each packet.
void SrsRtpVideoQueue::do_collect_packet(SrsRtpNackForReceiver* nack, vector<SrsRtpPacket2*>& frame) void SrsRtpVideoQueue::collect_frame(SrsRtpNackForReceiver* nack, SrsRtpPacket2** ppkt)
{ {
// When done, s point to the next available packet.
uint16_t next = queue_->low();
bool found = false; bool found = false;
for (; next != queue_->high(); ++next) { vector<SrsRtpPacket2*> frame;
// When done, next point to the next available packet.
uint16_t next = queue_->begin;
for (; next != queue_->end; ++next) {
SrsRtpPacket2* pkt = queue_->at(next); SrsRtpPacket2* pkt = queue_->at(next);
// Not found or in NACK, stop collecting frame. // Not found or in NACK, stop collecting frame.
if (!pkt || nack->find(next) != NULL) { if (!pkt || nack->find(next) != NULL) {
srs_trace("%s wait for nack seq=%u", tag_, next); srs_trace("wait for nack seq=%u", next);
break; return;
} }
// Ignore when the first packet not the start. // Ignore when the first packet not the start.
if (next == queue_->low() && !pkt->is_first_packet_of_frame) { if (next == queue_->begin && !pkt->video_is_first_packet) {
break; return;
} }
// OK, collect packet to frame. // OK, collect packet to frame.
@ -572,24 +530,115 @@ void SrsRtpVideoQueue::do_collect_packet(SrsRtpNackForReceiver* nack, vector<Srs
// Done, we got the last packet of frame. // Done, we got the last packet of frame.
// @remark Note that the STAP-A is marker false and it's the last packet. // @remark Note that the STAP-A is marker false and it's the last packet.
if (pkt->rtp_header.get_marker() || pkt->is_last_packet_of_frame) { if (pkt->rtp_header.get_marker() || pkt->video_is_last_packet) {
found = true; found = true;
next++; next++;
break; break;
} }
} }
if (!found) { if (!found || frame.empty()) {
frame.clear(); return;
} }
uint16_t cur = next - 1; if (next != queue_->begin) {
if (found && cur != queue_->high()) {
// Reset the range of packets to NULL in buffer. // Reset the range of packets to NULL in buffer.
queue_->reset(queue_->low(), next); queue_->reset(queue_->begin, next);
srs_verbose("%s collect on frame, update head seq=%u t %u", tag_, queue_->low(), next); srs_verbose("RTC collect video [%u, %u, %u]", queue_->begin, next, queue_->end);
queue_->advance_to(next); queue_->advance_to(next);
} }
// Merge packets to one packet.
covert_frame(frame, ppkt);
return;
}
void SrsRtpVideoQueue::covert_frame(std::vector<SrsRtpPacket2*>& frame, SrsRtpPacket2** ppkt)
{
if (frame.size() == 1) {
*ppkt = frame[0];
return;
}
// If more than one packet in a frame, it must be FU-A.
SrsRtpPacket2* head = frame.at(0);
SrsAvcNaluType nalu_type = head->nalu_type;
// Covert FU-A to one RAW RTP packet.
int nn_nalus = 0;
for (size_t i = 0; i < frame.size(); ++i) {
SrsRtpPacket2* pkt = frame[i];
SrsRtpFUAPayload2* payload = dynamic_cast<SrsRtpFUAPayload2*>(pkt->payload);
if (!payload) {
nn_nalus = 0;
break;
}
nn_nalus += payload->size;
}
// Invalid packets, ignore.
if (nalu_type != (SrsAvcNaluType)kFuA || !nn_nalus) {
for (int i = 0; i < (int)frame.size(); i++) {
SrsRtpPacket2* pkt = frame[i];
srs_freep(pkt);
}
return;
}
// Merge to one RAW RTP packet.
// TODO: FIXME: Should covert to multiple NALU RTP packet to avoid copying.
SrsRtpPacket2* pkt = new SrsRtpPacket2();
pkt->rtp_header = head->rtp_header;
SrsRtpFUAPayload2* head_payload = dynamic_cast<SrsRtpFUAPayload2*>(head->payload);
pkt->nalu_type = head_payload->nalu_type;
SrsRtpRawPayload* payload = pkt->reuse_raw();
payload->nn_payload = nn_nalus + 1;
payload->payload = new char[payload->nn_payload];
SrsBuffer buf(payload->payload, payload->nn_payload);
buf.write_1bytes(head_payload->nri | head_payload->nalu_type); // NALU header.
for (size_t i = 0; i < frame.size(); ++i) {
SrsRtpPacket2* pkt = frame[i];
SrsRtpFUAPayload2* payload = dynamic_cast<SrsRtpFUAPayload2*>(pkt->payload);
buf.write_bytes(payload->payload, payload->size);
}
*ppkt = pkt;
}
uint16_t SrsRtpVideoQueue::next_start_of_frame(uint16_t seq)
{
uint16_t s = seq;
if (srs_rtp_seq_distance(seq, queue_->begin) >= 0) {
s = queue_->begin + 1;
}
for (; s != queue_->end; ++s) {
SrsRtpPacket2* pkt = queue_->at(s);
if (pkt && pkt->video_is_first_packet) {
return s;
}
}
return queue_->end;
}
uint16_t SrsRtpVideoQueue::next_keyframe()
{
uint16_t s = queue_->begin + 1;
for (; s != queue_->end; ++s) {
SrsRtpPacket2* pkt = queue_->at(s);
if (pkt && pkt->video_is_idr && pkt->video_is_first_packet) {
return s;
}
}
return queue_->end;
} }

81
trunk/src/app/srs_app_rtp_queue.hpp
View file

@ -66,9 +66,9 @@ struct SrsRtpNackInfo
// distance(low=3, high=65534) === (int16_t)(uint16_t)((uint16_t)3-(uint16_t)65534) === 5 // distance(low=3, high=65534) === (int16_t)(uint16_t)((uint16_t)3-(uint16_t)65534) === 5
// distance(low=65532, high=65534) === (int16_t)(uint16_t)((uint16_t)65532-(uint16_t)65534) === -2 // distance(low=65532, high=65534) === (int16_t)(uint16_t)((uint16_t)65532-(uint16_t)65534) === -2
// For RTP sequence, it's only uint16 and may flip back, so 3 maybe 3+0xffff. // For RTP sequence, it's only uint16 and may flip back, so 3 maybe 3+0xffff.
inline bool srs_rtp_seq_distance(const uint16_t& low, const uint16_t& high) inline int16_t srs_rtp_seq_distance(const uint16_t& low, const uint16_t& high)
{ {
return ((int16_t)(high - low)) > 0; return (int16_t)(high - low);
} }
class SrsRtpNackForReceiver class SrsRtpNackForReceiver
@ -76,7 +76,7 @@ class SrsRtpNackForReceiver
private: private:
struct SeqComp { struct SeqComp {
bool operator()(const uint16_t& low, const uint16_t& high) const { bool operator()(const uint16_t& low, const uint16_t& high) const {
return srs_rtp_seq_distance(low, high); return srs_rtp_seq_distance(low, high) > 0;
} }
}; };
private: private:
@ -124,37 +124,34 @@ private:
uint64_t nn_seq_flip_backs; uint64_t nn_seq_flip_backs;
// Whether initialized, because we use uint16 so we can't use -1. // Whether initialized, because we use uint16 so we can't use -1.
bool initialized_; bool initialized_;
private: public:
// Current position we are working at. // The begin iterator for ring buffer.
uint16_t low_; // For example, when got 1 elems, the begin is 0.
uint16_t high_; uint16_t begin;
// The end iterator for ring buffer.
// For example, when got 1 elems, the end is 1.
uint16_t end;
public: public:
SrsRtpRingBuffer(int capacity); SrsRtpRingBuffer(int capacity);
virtual ~SrsRtpRingBuffer(); virtual ~SrsRtpRingBuffer();
public: public:
// Move the position of buffer. // Whether the ring buffer is empty.
uint16_t low(); bool empty();
uint16_t high(); // Get the count of elems in ring buffer.
int size();
// Move the low position of buffer to seq.
void advance_to(uint16_t seq); void advance_to(uint16_t seq);
// Free the packet at position. // Free the packet at position.
void set(uint16_t at, SrsRtpPacket2* pkt); void set(uint16_t at, SrsRtpPacket2* pkt);
void remove(uint16_t at); void remove(uint16_t at);
// Directly reset range [low, high] to NULL. // Directly reset range [first, last) to NULL.
void reset(uint16_t low, uint16_t high); void reset(uint16_t first, uint16_t last);
// Whether queue overflow or heavy(too many packets and need clear). // Whether queue overflow or heavy(too many packets and need clear).
bool overflow(); bool overflow();
bool is_heavy();
// Get the next start packet of frame.
// @remark If not found, return the low_, which should never be the "next" one,
// because it MAY or NOT current start packet of frame but never be the next.
uint16_t next_start_of_frame();
// Get the next seq of keyframe.
// @remark Return low_ if not found.
uint16_t next_keyframe();
// The highest sequence number, calculate the flip back base. // The highest sequence number, calculate the flip back base.
uint32_t get_extended_highest_sequence(); uint32_t get_extended_highest_sequence();
// Update the sequence, got the nack range by [low, high]. // Update the sequence, got the nack range by [first, last).
void update(uint16_t seq, bool startup, uint16_t& nack_low, uint16_t& nack_high); void update(uint16_t seq, uint16_t& nack_first, uint16_t& nack_last);
// Get the packet by seq. // Get the packet by seq.
SrsRtpPacket2* at(uint16_t seq); SrsRtpPacket2* at(uint16_t seq);
}; };
@ -171,31 +168,20 @@ private:
uint64_t number_of_packet_lossed_; uint64_t number_of_packet_lossed_;
protected: protected:
SrsRtpRingBuffer* queue_; SrsRtpRingBuffer* queue_;
uint64_t nn_collected_frames;
std::vector<std::vector<SrsRtpPacket2*> > frames_;
const char* tag_;
private:
bool request_key_frame_;
public: public:
SrsRtpQueue(const char* tag, int capacity); SrsRtpQueue(int capacity);
virtual ~SrsRtpQueue(); virtual ~SrsRtpQueue();
public: public:
virtual srs_error_t consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt); virtual srs_error_t consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt);
// TODO: FIXME: Should merge FU-A to RAW, then we can return RAW payloads. virtual void notify_drop_seq(uint16_t seq) = 0;
void collect_frames(std::vector<std::vector<SrsRtpPacket2*> >& frames); virtual void notify_nack_list_full() = 0;
bool should_request_key_frame();
void notify_drop_seq(uint16_t seq);
void notify_nack_list_full();
void request_keyframe();
public: public:
uint32_t get_extended_highest_sequence(); uint32_t get_extended_highest_sequence();
uint8_t get_fraction_lost(); uint8_t get_fraction_lost();
uint32_t get_cumulative_number_of_packets_lost(); uint32_t get_cumulative_number_of_packets_lost();
uint32_t get_interarrival_jitter(); uint32_t get_interarrival_jitter();
private: private:
void insert_into_nack_list(SrsRtpNackForReceiver* nack, uint16_t seq_start, uint16_t seq_end); void insert_into_nack_list(SrsRtpNackForReceiver* nack, uint16_t first, uint16_t last);
protected:
virtual void collect_packet(SrsRtpNackForReceiver* nack) = 0;
}; };
class SrsRtpAudioQueue : public SrsRtpQueue class SrsRtpAudioQueue : public SrsRtpQueue
@ -204,22 +190,31 @@ public:
SrsRtpAudioQueue(int capacity); SrsRtpAudioQueue(int capacity);
virtual ~SrsRtpAudioQueue(); virtual ~SrsRtpAudioQueue();
public: public:
virtual srs_error_t consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt); virtual void notify_drop_seq(uint16_t seq);
protected: virtual void notify_nack_list_full();
virtual void collect_packet(SrsRtpNackForReceiver* nack); virtual void collect_frames(SrsRtpNackForReceiver* nack, std::vector<SrsRtpPacket2*>& frames);
}; };
class SrsRtpVideoQueue : public SrsRtpQueue class SrsRtpVideoQueue : public SrsRtpQueue
{ {
private:
bool request_key_frame_;
public: public:
SrsRtpVideoQueue(int capacity); SrsRtpVideoQueue(int capacity);
virtual ~SrsRtpVideoQueue(); virtual ~SrsRtpVideoQueue();
public: public:
virtual void notify_drop_seq(uint16_t seq);
virtual void notify_nack_list_full();
virtual srs_error_t consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt); virtual srs_error_t consume(SrsRtpNackForReceiver* nack, SrsRtpPacket2* pkt);
protected: virtual void collect_frames(SrsRtpNackForReceiver* nack, std::vector<SrsRtpPacket2*>& frame);
virtual void collect_packet(SrsRtpNackForReceiver* nack); bool should_request_key_frame();
void request_keyframe();
private: private:
virtual void do_collect_packet(SrsRtpNackForReceiver* nack, std::vector<SrsRtpPacket2*>& frame); virtual void on_overflow(SrsRtpNackForReceiver* nack);
virtual void collect_frame(SrsRtpNackForReceiver* nack, SrsRtpPacket2** ppkt);
virtual void covert_frame(std::vector<SrsRtpPacket2*>& frame, SrsRtpPacket2** ppkt);
uint16_t next_start_of_frame(uint16_t seq);
uint16_t next_keyframe();
}; };
#endif #endif

16
trunk/src/app/srs_app_sdp.cpp
View file

@ -111,6 +111,7 @@ SrsSessionInfo::~SrsSessionInfo()
srs_error_t SrsSessionInfo::parse_attribute(const std::string& attribute, const std::string& value) srs_error_t SrsSessionInfo::parse_attribute(const std::string& attribute, const std::string& value)
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
if (attribute == "ice-ufrag") { if (attribute == "ice-ufrag") {
ice_ufrag_ = value; ice_ufrag_ = value;
} else if (attribute == "ice-pwd") { } else if (attribute == "ice-pwd") {
@ -134,22 +135,24 @@ srs_error_t SrsSessionInfo::parse_attribute(const std::string& attribute, const
srs_error_t SrsSessionInfo::encode(std::ostringstream& os) srs_error_t SrsSessionInfo::encode(std::ostringstream& os)
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
if (!ice_ufrag_.empty()) { if (!ice_ufrag_.empty()) {
os << "a=ice-ufrag:" << ice_ufrag_ << kCRLF; os << "a=ice-ufrag:" << ice_ufrag_ << kCRLF;
} }
if (!ice_pwd_.empty()) { if (!ice_pwd_.empty()) {
os << "a=ice-pwd:" << ice_pwd_ << kCRLF; os << "a=ice-pwd:" << ice_pwd_ << kCRLF;
} }
// For ICE-lite, we never set the trickle.
if (!ice_options_.empty()) { if (!ice_options_.empty()) {
os << "a=ice-options:" << ice_options_ << kCRLF; os << "a=ice-options:" << ice_options_ << kCRLF;
} else {
// @see: https://webrtcglossary.com/trickle-ice/
// Trickle ICE is an optimization of the ICE specification for NAT traversal.
os << "a=ice-options:trickle" << kCRLF;
} }
if (!fingerprint_algo_.empty() && ! fingerprint_.empty()) { if (!fingerprint_algo_.empty() && ! fingerprint_.empty()) {
os << "a=fingerprint:" << fingerprint_algo_ << " " << fingerprint_ << kCRLF; os << "a=fingerprint:" << fingerprint_algo_ << " " << fingerprint_ << kCRLF;
} }
if (!setup_.empty()) { if (!setup_.empty()) {
os << "a=setup:" << setup_ << kCRLF; os << "a=setup:" << setup_ << kCRLF;
} }
@ -179,6 +182,7 @@ SrsSSRCInfo::~SrsSSRCInfo()
srs_error_t SrsSSRCInfo::encode(std::ostringstream& os) srs_error_t SrsSSRCInfo::encode(std::ostringstream& os)
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
if (ssrc_ == 0) { if (ssrc_ == 0) {
return srs_error_new(ERROR_RTC_SDP_DECODE, "invalid ssrc"); return srs_error_new(ERROR_RTC_SDP_DECODE, "invalid ssrc");
} }
@ -373,7 +377,7 @@ srs_error_t SrsMediaDesc::encode(std::ostringstream& os)
<< " typ " << iter->type_ << " typ " << iter->type_
<< " generation 0" << kCRLF; << " generation 0" << kCRLF;
srs_trace("local SDP candidate line=%s", os.str().c_str()); srs_verbose("local SDP candidate line=%s", os.str().c_str());
} }
return err; return err;
@ -528,7 +532,7 @@ srs_error_t SrsMediaDesc::parse_attr_mid(const std::string& value)
std::istringstream is(value); std::istringstream is(value);
// mid_ means m-line id // mid_ means m-line id
FETCH(is, mid_); FETCH(is, mid_);
srs_trace("mid=%s", mid_.c_str()); srs_verbose("mid=%s", mid_.c_str());
return err; return err;
} }

35
trunk/src/app/srs_app_source.cpp
View file

@ -1921,7 +1921,7 @@ SrsSource::SrsSource()
atc = false; atc = false;
#ifdef SRS_RTC #ifdef SRS_RTC
rtc_publisher = NULL; rtc_publisher_ = NULL;
#endif #endif
} }
@ -2594,13 +2594,28 @@ void SrsSource::on_unpublish()
} }
} }
srs_error_t SrsSource::create_consumer(SrsConnection* conn, SrsConsumer*& consumer, bool ds, bool dm, bool dg) srs_error_t SrsSource::create_consumer(SrsConnection* conn, SrsConsumer*& consumer)
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
consumer = new SrsConsumer(this, conn); consumer = new SrsConsumer(this, conn);
consumers.push_back(consumer); consumers.push_back(consumer);
// for edge, when play edge stream, check the state
if (_srs_config->get_vhost_is_edge(req->vhost)) {
// notice edge to start for the first client.
if ((err = play_edge->on_client_play()) != srs_success) {
return srs_error_wrap(err, "play edge");
}
}
return err;
}
srs_error_t SrsSource::consumer_dumps(SrsConsumer* consumer, bool ds, bool dm, bool dg)
{
srs_error_t err = srs_success;
srs_utime_t queue_size = _srs_config->get_queue_length(req->vhost); srs_utime_t queue_size = _srs_config->get_queue_length(req->vhost);
consumer->set_queue_size(queue_size); consumer->set_queue_size(queue_size);
@ -2637,14 +2652,6 @@ srs_error_t SrsSource::create_consumer(SrsConnection* conn, SrsConsumer*& consum
srs_trace("create consumer, active=%d, ignore gop cache, jitter=%d", hub->active(), jitter_algorithm); srs_trace("create consumer, active=%d, ignore gop cache, jitter=%d", hub->active(), jitter_algorithm);
} }
// for edge, when play edge stream, check the state
if (_srs_config->get_vhost_is_edge(req->vhost)) {
// notice edge to start for the first client.
if ((err = play_edge->on_client_play()) != srs_success) {
return srs_error_wrap(err, "play edge");
}
}
return err; return err;
} }
@ -2699,16 +2706,14 @@ SrsMetaCache* SrsSource::cached_meta()
return meta; return meta;
} }
void SrsSource::request_keyframe() SrsRtcPublisher* SrsSource::rtc_publisher()
{ {
if (rtc_publisher) { return rtc_publisher_;
rtc_publisher->request_keyframe();
}
} }
void SrsSource::set_rtc_publisher(SrsRtcPublisher* v) void SrsSource::set_rtc_publisher(SrsRtcPublisher* v)
{ {
rtc_publisher = v; rtc_publisher_ = v;
} }
srs_error_t SrsSource::on_rtc_audio(SrsSharedPtrMessage* audio) srs_error_t SrsSource::on_rtc_audio(SrsSharedPtrMessage* audio)

16
trunk/src/app/srs_app_source.hpp
View file

@ -551,7 +551,7 @@ private:
srs_utime_t die_at; srs_utime_t die_at;
#ifdef SRS_RTC #ifdef SRS_RTC
private: private:
SrsRtcPublisher* rtc_publisher; SrsRtcPublisher* rtc_publisher_;
#endif #endif
public: public:
SrsSource(); SrsSource();
@ -599,12 +599,14 @@ public:
virtual srs_error_t on_publish(); virtual srs_error_t on_publish();
virtual void on_unpublish(); virtual void on_unpublish();
public: public:
// Create consumer and dumps packets in cache. // Create consumer
// @param consumer, output the create consumer. // @param consumer, output the create consumer.
virtual srs_error_t create_consumer(SrsConnection* conn, SrsConsumer*& consumer);
// Dumps packets in cache to consumer.
// @param ds, whether dumps the sequence header. // @param ds, whether dumps the sequence header.
// @param dm, whether dumps the metadata. // @param dm, whether dumps the metadata.
// @param dg, whether dumps the gop cache. // @param dg, whether dumps the gop cache.
virtual srs_error_t create_consumer(SrsConnection* conn, SrsConsumer*& consumer, bool ds = true, bool dm = true, bool dg = true); virtual srs_error_t consumer_dumps(SrsConsumer* consumer, bool ds = true, bool dm = true, bool dg = true);
virtual void on_consumer_destroy(SrsConsumer* consumer); virtual void on_consumer_destroy(SrsConsumer* consumer);
virtual void set_cache(bool enabled); virtual void set_cache(bool enabled);
virtual SrsRtmpJitterAlgorithm jitter(); virtual SrsRtmpJitterAlgorithm jitter();
@ -619,12 +621,10 @@ public:
virtual std::string get_curr_origin(); virtual std::string get_curr_origin();
#ifdef SRS_RTC #ifdef SRS_RTC
public: public:
// Get the cached meta, as such the sps/pps. // For RTC, we need to package SPS/PPS(in cached meta) before each IDR.
SrsMetaCache* cached_meta(); SrsMetaCache* cached_meta();
// Request keyframe for new client. // Get and set the publisher, passed to consumer to process requests such as PLI.
// TODO: FIXME: Maybe we could cache the keyframe. SrsRtcPublisher* rtc_publisher();
// TODO: FIXME: Maybe we should only response for the new clients.
void request_keyframe();
void set_rtc_publisher(SrsRtcPublisher* v); void set_rtc_publisher(SrsRtcPublisher* v);
// When got RTC audio message, which is encoded in opus. // When got RTC audio message, which is encoded in opus.
// TODO: FIXME: Merge with on_audio. // TODO: FIXME: Merge with on_audio.

2
trunk/src/core/srs_core_version4.hpp
View file

@ -24,6 +24,6 @@
#ifndef SRS_CORE_VERSION4_HPP #ifndef SRS_CORE_VERSION4_HPP
#define SRS_CORE_VERSION4_HPP #define SRS_CORE_VERSION4_HPP
#define SRS_VERSION4_REVISION 24 #define SRS_VERSION4_REVISION 25
#endif #endif

1
trunk/src/kernel/srs_kernel_error.hpp
View file

@ -349,6 +349,7 @@
#define ERROR_RTC_SDP_EXCHANGE 5018 #define ERROR_RTC_SDP_EXCHANGE 5018
#define ERROR_RTC_API_BODY 5019 #define ERROR_RTC_API_BODY 5019
#define ERROR_RTC_SOURCE_BUSY 5020 #define ERROR_RTC_SOURCE_BUSY 5020
#define ERROR_RTC_DISABLED 5021
/////////////////////////////////////////////////////// ///////////////////////////////////////////////////////
// GB28181 API error. // GB28181 API error.

6
trunk/src/kernel/srs_kernel_rtp.cpp
View file

@ -277,9 +277,9 @@ SrsRtpPacket2::SrsRtpPacket2()
payload = NULL; payload = NULL;
decode_handler = NULL; decode_handler = NULL;
is_first_packet_of_frame = false; video_is_first_packet = false;
is_last_packet_of_frame = false; video_is_last_packet = false;
is_key_frame = false; video_is_idr = false;
nalu_type = SrsAvcNaluTypeReserved; nalu_type = SrsAvcNaluTypeReserved;
cache_raw = new SrsRtpRawPayload(); cache_raw = new SrsRtpRawPayload();

8
trunk/src/kernel/srs_kernel_rtp.hpp
View file

@ -114,10 +114,10 @@ public:
int padding; int padding;
// Decoder helper. // Decoder helper.
public: public:
// Helper information for decoder. // Helper information for video decoder only.
bool is_first_packet_of_frame; bool video_is_first_packet;
bool is_last_packet_of_frame; bool video_is_last_packet;
bool is_key_frame; bool video_is_idr;
// The first byte as nalu type, for video decoder only. // The first byte as nalu type, for video decoder only.
SrsAvcNaluType nalu_type; SrsAvcNaluType nalu_type;
// The original payload bytes length. // The original payload bytes length.

6
trunk/src/protocol/srs_sip_stack.cpp
View file

@ -635,9 +635,9 @@ srs_error_t SrsSipStack::do_parse_request(SrsSipRequest* req, const char* recv_m
std::vector<std::string> vec_device_status = srs_string_split(body_map["Response@DeviceList@Item@Status"], ","); std::vector<std::string> vec_device_status = srs_string_split(body_map["Response@DeviceList@Item@Status"], ",");
//map key:devicd_id value:status //map key:devicd_id value:status
for(int i=0 ; i<vec_device_id.size(); i++){ for(int i=0 ; i< (int)vec_device_id.size(); i++){
std::string status = ""; std::string status = "";
if (vec_device_id.size() > i) { if ((int)vec_device_id.size() > i) {
status = vec_device_status.at(i); status = vec_device_status.at(i);
} }
@ -653,7 +653,7 @@ srs_error_t SrsSipStack::do_parse_request(SrsSipRequest* req, const char* recv_m
if (cmdtype == "Keepalive"){ if (cmdtype == "Keepalive"){
//TODO: ???? //TODO: ????
std::vector<std::string> vec_device_id = srs_string_split(body_map["Notify@Info@DeviceID"], ","); std::vector<std::string> vec_device_id = srs_string_split(body_map["Notify@Info@DeviceID"], ",");
for(int i=0; i<vec_device_id.size(); i++){ for(int i=0; i< (int)vec_device_id.size(); i++){
//req->device_list_map[vec_device_id.at(i)] = "OFF"; //req->device_list_map[vec_device_id.at(i)] = "OFF";
} }
}else{ }else{

48
trunk/src/service/srs_service_st.cpp
View file

@ -412,53 +412,9 @@ int srs_sendmsg(srs_netfd_t stfd, const struct msghdr *msg, int flags, srs_utime
return st_sendmsg((st_netfd_t)stfd, msg, flags, (st_utime_t)timeout); return st_sendmsg((st_netfd_t)stfd, msg, flags, (st_utime_t)timeout);
} }
int srs_sendmmsg(srs_netfd_t stfd, struct mmsghdr *msgvec, unsigned int vlen, int flags, srs_utime_t timeout) int srs_sendmmsg(srs_netfd_t stfd, struct srs_mmsghdr *msgvec, unsigned int vlen, int flags, srs_utime_t timeout)
{ {
#if !defined(SRS_HAS_SENDMMSG) || !defined(SRS_SENDMMSG) return st_sendmmsg((st_netfd_t)stfd, (struct st_mmsghdr*)msgvec, vlen, flags, (st_utime_t)timeout);
// @see http://man7.org/linux/man-pages/man2/sendmmsg.2.html
for (int i = 0; i < (int)vlen; ++i) {
struct mmsghdr* p = msgvec + i;
int n = srs_sendmsg(stfd, &p->msg_hdr, flags, timeout);
if (n < 0) {
// An error is returned only if no datagrams could be sent.
if (i == 0) {
return n;
}
return i + 1;
}
p->msg_len = n;
}
// Returns the number of messages sent from msgvec; if this is less than vlen, the caller can retry with a
// further sendmmsg() call to send the remaining messages.
return vlen;
#else
if (vlen == 1) {
#if 1
int r0 = srs_sendmsg(stfd, &msgvec->msg_hdr, flags, timeout);
if (r0 < 0) {
return r0;
}
msgvec->msg_len = r0;
#else
msgvec->msg_len = 0;
int tolen = (int)msgvec->msg_hdr.msg_namelen;
const struct sockaddr* to = (const struct sockaddr*)msgvec->msg_hdr.msg_name;
for (int i = 0; i < (int)msgvec->msg_hdr.msg_iovlen; i++) {
iovec* iov = msgvec->msg_hdr.msg_iov + i;
int r0 = srs_sendto(stfd, (void*)iov->iov_base, (int)iov->iov_len, to, tolen, timeout);
if (r0 < 0) {
return r0;
}
msgvec->msg_len += r0;
}
#endif
return 1;
}
return st_sendmmsg((st_netfd_t)stfd, msgvec, vlen, flags, (st_utime_t)timeout);
#endif
} }
srs_netfd_t srs_accept(srs_netfd_t stfd, struct sockaddr *addr, int *addrlen, srs_utime_t timeout) srs_netfd_t srs_accept(srs_netfd_t stfd, struct sockaddr *addr, int *addrlen, srs_utime_t timeout)

6
trunk/src/service/srs_service_st.hpp
View file

@ -92,15 +92,13 @@ extern int srs_sendto(srs_netfd_t stfd, void *buf, int len, const struct sockadd
extern int srs_recvmsg(srs_netfd_t stfd, struct msghdr *msg, int flags, srs_utime_t timeout); extern int srs_recvmsg(srs_netfd_t stfd, struct msghdr *msg, int flags, srs_utime_t timeout);
extern int srs_sendmsg(srs_netfd_t stfd, const struct msghdr *msg, int flags, srs_utime_t timeout); extern int srs_sendmsg(srs_netfd_t stfd, const struct msghdr *msg, int flags, srs_utime_t timeout);
#if !defined(SRS_HAS_SENDMMSG)
// @see http://man7.org/linux/man-pages/man2/sendmmsg.2.html // @see http://man7.org/linux/man-pages/man2/sendmmsg.2.html
#include <sys/socket.h> #include <sys/socket.h>
struct mmsghdr { struct srs_mmsghdr {
struct msghdr msg_hdr; /* Message header */ struct msghdr msg_hdr; /* Message header */
unsigned int msg_len; /* Number of bytes transmitted */ unsigned int msg_len; /* Number of bytes transmitted */
}; };
#endif extern int srs_sendmmsg(srs_netfd_t stfd, struct srs_mmsghdr *msgvec, unsigned int vlen, int flags, srs_utime_t timeout);
extern int srs_sendmmsg(srs_netfd_t stfd, struct mmsghdr *msgvec, unsigned int vlen, int flags, srs_utime_t timeout);
extern srs_netfd_t srs_accept(srs_netfd_t stfd, struct sockaddr *addr, int *addrlen, srs_utime_t timeout); extern srs_netfd_t srs_accept(srs_netfd_t stfd, struct sockaddr *addr, int *addrlen, srs_utime_t timeout);