amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-22 10:40:05 +08:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Merge branch 'master' of https://gitee.com/xia-chu/ZLMediaKit into transcode2

Browse Source 
# Conflicts:
#	src/Common/config.cpp
#	src/Common/config.h
This commit is contained in:
cqm 2024-08-19 17:39:50 +08:00
commit d329f76edf
46 changed files with 382 additions and 333 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.github/workflows/issue_lint.yml vendored
View File

@ -42,6 +42,13 @@ jobs:
issue_number: context.issue.number, issue_number: context.issue.number,
body: '此issue由于不符合模板规范已经自动关闭，请重新按照模板规范确保包含模板中所有章节标题再提交\n', body: '此issue由于不符合模板规范已经自动关闭，请重新按照模板规范确保包含模板中所有章节标题再提交\n',
}); });
await github.rest.issues.addLabels({
owner: context.issue.owner,
repo: context.issue.repo,
issue_number: context.issue.number,
labels: ['自动关闭']
});
await github.rest.issues.update({ await github.rest.issues.update({
owner: context.issue.owner, owner: context.issue.owner,

2
3rdpart/ZLToolKit

@ -1 +1 @@
Subproject commit fb695d203421d906c473018022a736fa4a7a47e4 Subproject commit a6e30e41f0c52f9d36c41eb79ac69b50020a6ac9

6
api/include/mk_events_objects.h
View File

@ -159,9 +159,9 @@ API_EXPORT int API_CALL mk_media_source_seek_to(const mk_media_source ctx,uint32
*/ */
typedef void(API_CALL *on_mk_media_source_send_rtp_result)(void *user_data, uint16_t local_port, int err, const char *msg); typedef void(API_CALL *on_mk_media_source_send_rtp_result)(void *user_data, uint16_t local_port, int err, const char *msg);
//MediaSource::startSendRtp,请参考mk_media_start_send_rtp,注意ctx参数类型不一样 // MediaSource::startSendRtp,请参考mk_media_start_send_rtp,注意ctx参数类型不一样
API_EXPORT void API_CALL mk_media_source_start_send_rtp(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_source_send_rtp_result cb, void *user_data); API_EXPORT void API_CALL mk_media_source_start_send_rtp(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int con_type, on_mk_media_source_send_rtp_result cb, void *user_data);
API_EXPORT void API_CALL mk_media_source_start_send_rtp2(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_source_send_rtp_result cb, void *user_data, on_user_data_free user_data_free); API_EXPORT void API_CALL mk_media_source_start_send_rtp2(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int con_type, on_mk_media_source_send_rtp_result cb, void *user_data, on_user_data_free user_data_free);
//MediaSource::stopSendRtp，请参考mk_media_stop_send_rtp,注意ctx参数类型不一样 //MediaSource::stopSendRtp，请参考mk_media_stop_send_rtp,注意ctx参数类型不一样
API_EXPORT int API_CALL mk_media_source_stop_send_rtp(const mk_media_source ctx); API_EXPORT int API_CALL mk_media_source_stop_send_rtp(const mk_media_source ctx);

6
api/include/mk_media.h
View File

@ -269,12 +269,12 @@ typedef on_mk_media_source_send_rtp_result on_mk_media_send_rtp_result;
* @param dst_url 目标ip或域名 * @param dst_url 目标ip或域名
* @param dst_port 目标端口 * @param dst_port 目标端口
* @param ssrc rtp的ssrc，10进制的字符串打印 * @param ssrc rtp的ssrc，10进制的字符串打印
* @param is_udp 是否为udp * @param con_type 0: tcp主动，1：udp主动，2：tcp被动，3：udp被动
* @param cb 启动成功或失败回调 * @param cb 启动成功或失败回调
* @param user_data 回调用户指针 * @param user_data 回调用户指针
*/ */
API_EXPORT void API_CALL mk_media_start_send_rtp(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_send_rtp_result cb, void *user_data); API_EXPORT void API_CALL mk_media_start_send_rtp(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int con_type, on_mk_media_send_rtp_result cb, void *user_data);
API_EXPORT void API_CALL mk_media_start_send_rtp2(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_send_rtp_result cb, void *user_data, on_user_data_free user_data_free); API_EXPORT void API_CALL mk_media_start_send_rtp2(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int con_type, on_mk_media_send_rtp_result cb, void *user_data, on_user_data_free user_data_free);
/** /**
* 停止某路或全部ps-rtp发送，此api线程安全 * 停止某路或全部ps-rtp发送，此api线程安全

8
api/source/mk_events_objects.cpp
View File

@ -295,11 +295,11 @@ API_EXPORT int API_CALL mk_media_source_seek_to(const mk_media_source ctx,uint32
MediaSource *src = (MediaSource *)ctx; MediaSource *src = (MediaSource *)ctx;
return src->seekTo(stamp); return src->seekTo(stamp);
} }
API_EXPORT void API_CALL mk_media_source_start_send_rtp(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_source_send_rtp_result cb, void *user_data) { API_EXPORT void API_CALL mk_media_source_start_send_rtp(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int con_type, on_mk_media_source_send_rtp_result cb, void *user_data) {
mk_media_source_start_send_rtp2(ctx, dst_url, dst_port, ssrc, is_udp, cb, user_data, nullptr); mk_media_source_start_send_rtp2(ctx, dst_url, dst_port, ssrc, con_type, cb, user_data, nullptr);
} }
API_EXPORT void API_CALL mk_media_source_start_send_rtp2(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_source_send_rtp_result cb, void *user_data, on_user_data_free user_data_free){ API_EXPORT void API_CALL mk_media_source_start_send_rtp2(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int con_type, on_mk_media_source_send_rtp_result cb, void *user_data, on_user_data_free user_data_free){
assert(ctx && dst_url && ssrc); assert(ctx && dst_url && ssrc);
MediaSource *src = (MediaSource *)ctx; MediaSource *src = (MediaSource *)ctx;
@ -307,7 +307,7 @@ API_EXPORT void API_CALL mk_media_source_start_send_rtp2(const mk_media_source c
args.dst_url = dst_url; args.dst_url = dst_url;
args.dst_port = dst_port; args.dst_port = dst_port;
args.ssrc = ssrc; args.ssrc = ssrc;
args.is_udp = is_udp; args.con_type = (mediakit::MediaSourceEvent::SendRtpArgs::ConType)con_type;
std::shared_ptr<void> ptr(user_data, user_data_free ? user_data_free : [](void *) {}); std::shared_ptr<void> ptr(user_data, user_data_free ? user_data_free : [](void *) {});
src->startSendRtp(args, [cb, ptr](uint16_t local_port, const SockException &ex){ src->startSendRtp(args, [cb, ptr](uint16_t local_port, const SockException &ex){

2
api/source/mk_frame.cpp
View File

@ -15,7 +15,7 @@
using namespace mediakit; using namespace mediakit;
extern "C" { extern "C" {
#define XX(name, type, value, str, mpeg_id, mp4_id) API_EXPORT const int MK##name = value; #define XX(name, type, value, str, mpeg_id, mp4_id) const int MK##name = value;
CODEC_MAP(XX) CODEC_MAP(XX)
#undef XX #undef XX
} }

8
api/source/mk_media.cpp
View File

@ -282,11 +282,11 @@ API_EXPORT int API_CALL mk_media_input_audio(mk_media ctx, const void *data, int
return (*obj)->getChannel()->inputAudio((const char*)data, len, dts); return (*obj)->getChannel()->inputAudio((const char*)data, len, dts);
} }
API_EXPORT void API_CALL mk_media_start_send_rtp(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_send_rtp_result cb, void *user_data) { API_EXPORT void API_CALL mk_media_start_send_rtp(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int con_type, on_mk_media_send_rtp_result cb, void *user_data) {
mk_media_start_send_rtp2(ctx, dst_url, dst_port, ssrc, is_udp, cb, user_data, nullptr); mk_media_start_send_rtp2(ctx, dst_url, dst_port, ssrc, con_type, cb, user_data, nullptr);
} }
API_EXPORT void API_CALL mk_media_start_send_rtp2(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_send_rtp_result cb, void *user_data, API_EXPORT void API_CALL mk_media_start_send_rtp2(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int con_type, on_mk_media_send_rtp_result cb, void *user_data,
on_user_data_free user_data_free) { on_user_data_free user_data_free) {
assert(ctx && dst_url && ssrc); assert(ctx && dst_url && ssrc);
MediaHelper::Ptr* obj = (MediaHelper::Ptr*) ctx; MediaHelper::Ptr* obj = (MediaHelper::Ptr*) ctx;
@ -295,7 +295,7 @@ API_EXPORT void API_CALL mk_media_start_send_rtp2(mk_media ctx, const char *dst_
args.dst_url = dst_url; args.dst_url = dst_url;
args.dst_port = dst_port; args.dst_port = dst_port;
args.ssrc = ssrc; args.ssrc = ssrc;
args.is_udp = is_udp; args.con_type = (mediakit::MediaSourceEvent::SendRtpArgs::ConType)con_type;
// sender参数无用 // sender参数无用
auto ref = *obj; auto ref = *obj;

7
api/source/mk_rtp_server.cpp
View File

@ -14,17 +14,20 @@ using namespace toolkit;
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
#include "Rtp/RtpServer.h" #include "Rtp/RtpServer.h"
#include "Common/config.h"
using namespace mediakit; using namespace mediakit;
API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create(uint16_t port, int tcp_mode, const char *stream_id) { API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create(uint16_t port, int tcp_mode, const char *stream_id) {
RtpServer::Ptr *server = new RtpServer::Ptr(new RtpServer); RtpServer::Ptr *server = new RtpServer::Ptr(new RtpServer);
(*server)->start(port, MediaTuple { DEFAULT_VHOST, kRtpAppName, stream_id, "" }, (RtpServer::TcpMode)tcp_mode); GET_CONFIG(std::string, local_ip, General::kListenIP)
(*server)->start(port, local_ip.c_str(), MediaTuple { DEFAULT_VHOST, kRtpAppName, stream_id, "" }, (RtpServer::TcpMode)tcp_mode);
return (mk_rtp_server)server; return (mk_rtp_server)server;
} }
API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create2(uint16_t port, int tcp_mode, const char *vhost, const char *app, const char *stream_id) { API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create2(uint16_t port, int tcp_mode, const char *vhost, const char *app, const char *stream_id) {
RtpServer::Ptr *server = new RtpServer::Ptr(new RtpServer); RtpServer::Ptr *server = new RtpServer::Ptr(new RtpServer);
(*server)->start(port, MediaTuple { vhost, app, stream_id, "" }, (RtpServer::TcpMode)tcp_mode); GET_CONFIG(std::string, local_ip, General::kListenIP)
(*server)->start(port, local_ip.c_str(), MediaTuple { vhost, app, stream_id, "" }, (RtpServer::TcpMode)tcp_mode);
return (mk_rtp_server)server; return (mk_rtp_server)server;
} }

3
api/tests/h264_media_server.c
View File

@ -29,9 +29,8 @@ static void on_h264_frame(void *user_data, mk_h264_splitter splitter, const char
#else #else
usleep(40 * 1000); usleep(40 * 1000);
#endif #endif
static int dts = 0; uint64_t dts = mk_util_get_current_millisecond();
mk_frame frame = mk_frame_create(MKCodecH264, dts, dts, data, size, NULL, NULL); mk_frame frame = mk_frame_create(MKCodecH264, dts, dts, data, size, NULL, NULL);
dts += 40;
mk_media_input_frame((mk_media) user_data, frame); mk_media_input_frame((mk_media) user_data, frame);
mk_frame_unref(frame); mk_frame_unref(frame);
} }

10
api/tests/player_opencv.c
View File

@ -30,7 +30,13 @@ void API_CALL on_frame_decode(void *user_data, mk_frame_pix frame) {
int h = mk_get_av_frame_height(mk_frame_pix_get_av_frame(frame)); int h = mk_get_av_frame_height(mk_frame_pix_get_av_frame(frame));
#if 1 #if 1
uint8_t *brg24 = malloc(w * h * 3); int align = 32;
size_t pixel_size = 3;
size_t raw_linesize = w * pixel_size;
// 对齐后的宽度
size_t aligned_linesize = (raw_linesize + align - 1) & ~(align - 1);
size_t total_size = aligned_linesize * h;
uint8_t* brg24 = malloc(total_size);
mk_swscale_input_frame(ctx->swscale, frame, brg24); mk_swscale_input_frame(ctx->swscale, frame, brg24);
free(brg24); free(brg24);
#else #else
@ -106,4 +112,4 @@ int main(int argc, char *argv[]) {
mk_swscale_release(ctx.swscale); mk_swscale_release(ctx.swscale);
} }
return 0; return 0;
} }

14
conf/config.ini
View File

@ -140,6 +140,8 @@ wait_add_track_ms=3000
unready_frame_cache=100 unready_frame_cache=100
#是否启用观看人数变化事件广播，置1则启用，置0则关闭 #是否启用观看人数变化事件广播，置1则启用，置0则关闭
broadcast_player_count_changed=0 broadcast_player_count_changed=0
#绑定的本地网卡ip
listen_ip=::
[hls] [hls]
#hls写文件的buf大小，调整参数可以提高文件io性能 #hls写文件的buf大小，调整参数可以提高文件io性能
@ -389,13 +391,13 @@ start_bitrate=0
max_bitrate=0 max_bitrate=0
min_bitrate=0 min_bitrate=0
#nack接收端 #nack接收端, rtp发送端，zlm发送rtc流
#Nack缓存包最早时间间隔 #rtp重发缓存列队最大长度，单位毫秒
maxNackMS=5000 maxRtpCacheMS=5000
#Nack包检查间隔(包数量) #rtp重发缓存列队最大长度，单位个数
rtpCacheCheckInterval=100 maxRtpCacheSize=2048
#nack发送端 #nack发送端，rtp接收端，zlm接收rtc推流
#最大保留的rtp丢包状态个数 #最大保留的rtp丢包状态个数
nackMaxSize=2048 nackMaxSize=2048
#rtp丢包状态最长保留时间 #rtp丢包状态最长保留时间

2
ext-codec/AACRtp.cpp
View File

@ -86,7 +86,7 @@ bool AACRtpDecoder::inputRtp(const RtpPacket::Ptr &rtp, bool key_pos) {
} }
// 每个audio unit时间戳增量 // 每个audio unit时间戳增量
auto dts_inc = (stamp - _last_dts) / au_header_count; auto dts_inc = static_cast<int64_t>(stamp - _last_dts) / au_header_count;
if (dts_inc < 0 || dts_inc > 100) { if (dts_inc < 0 || dts_inc > 100) {
// 时间戳增量异常，忽略 // 时间戳增量异常，忽略
dts_inc = 0; dts_inc = 0;

2
ext-codec/H264Rtp.cpp
View File

@ -184,7 +184,7 @@ void H264RtpDecoder::outputFrame(const RtpPacket::Ptr &rtp, const H264Frame::Ptr
_gop_dropped = false; _gop_dropped = false;
InfoL << "new gop received, rtp:\r\n" << rtp->dumpString(); InfoL << "new gop received, rtp:\r\n" << rtp->dumpString();
} }
if (!_gop_dropped) { if (!_gop_dropped || frame->configFrame()) {
RtpCodec::inputFrame(frame); RtpCodec::inputFrame(frame);
} }
_frame = obtainFrame(); _frame = obtainFrame();

2
ext-codec/H265Rtp.cpp
View File

@ -240,7 +240,7 @@ void H265RtpDecoder::outputFrame(const RtpPacket::Ptr &rtp, const H265Frame::Ptr
_gop_dropped = false; _gop_dropped = false;
InfoL << "new gop received, rtp:\r\n" << rtp->dumpString(); InfoL << "new gop received, rtp:\r\n" << rtp->dumpString();
} }
if (!_gop_dropped) { if (!_gop_dropped || frame->configFrame()) {
RtpCodec::inputFrame(frame); RtpCodec::inputFrame(frame);
} }
_frame = obtainFrame(); _frame = obtainFrame();

2
ext-codec/JPEGRtp.cpp
View File

@ -598,7 +598,7 @@ void JPEGRtpEncoder::rtpSendJpeg(const uint8_t *buf, int size, uint64_t pts, uin
{ {
const uint8_t *qtables[4] = { NULL }; const uint8_t *qtables[4] = { NULL };
int nb_qtables = 0; int nb_qtables = 0;
uint8_t w, h; uint8_t w { 0 }, h { 0 };
uint8_t *p; uint8_t *p;
int off = 0; /* fragment offset of the current JPEG frame */ int off = 0; /* fragment offset of the current JPEG frame */
int len; int len;

20
postman/ZLMediaKit.postman_collection.json
View File

@ -1881,7 +1881,7 @@
"response": [] "response": []
}, },
{ {
"name": "开始发送rtp(startSendRtp)", "name": "开始active模式发送rtp(startSendRtp)",
"request": { "request": {
"method": "GET", "method": "GET",
"header": [], "header": [],
@ -1940,7 +1940,7 @@
{ {
"key": "is_udp", "key": "is_udp",
"value": "0", "value": "0",
"description": "是否为udp模式,否则为tcp模式" "description": "1:udp active模式, 0:tcp active模式"
}, },
{ {
"key": "src_port", "key": "src_port",
@ -1955,9 +1955,9 @@
"disabled": true "disabled": true
}, },
{ {
"key": "use_ps", "key": "type",
"value": "1", "value": "1",
"description": "rtp打包采用ps还是es模式，默认采用ps模式，该参数非必选参数", "description": "rtp打包模式，0:es, 1: ps, 2: ts",
"disabled": true "disabled": true
}, },
{ {
@ -1990,7 +1990,7 @@
"response": [] "response": []
}, },
{ {
"name": "开始tcp passive被动发送rtp(startSendRtpPassive)", "name": "开始passive模式发送rtp(startSendRtpPassive)",
"request": { "request": {
"method": "GET", "method": "GET",
"header": [], "header": [],
@ -2030,6 +2030,12 @@
"value": "1", "value": "1",
"description": "rtp推流的ssrc，ssrc不同时，可以推流到多个上级服务器" "description": "rtp推流的ssrc，ssrc不同时，可以推流到多个上级服务器"
}, },
{
"key": "is_udp",
"value": "0",
"disabled": true,
"description": "1:udp passive模式, 0:tcp passive模式"
},
{ {
"key": "src_port", "key": "src_port",
"value": "0", "value": "0",
@ -2043,9 +2049,9 @@
"disabled": true "disabled": true
}, },
{ {
"key": "use_ps", "key": "type",
"value": "1", "value": "1",
"description": "rtp打包采用ps还是es模式，默认采用ps模式，该参数非必选参数", "description": "rtp打包模式，0:es, 1: ps, 2: ts",
"disabled": true "disabled": true
}, },
{ {

92
server/WebApi.cpp
View File

@ -482,7 +482,7 @@ uint16_t openRtpServer(uint16_t local_port, const mediakit::MediaTuple &tuple, i
} }
auto server = s_rtp_server.makeWithAction(key, [&](RtpServer::Ptr server) { auto server = s_rtp_server.makeWithAction(key, [&](RtpServer::Ptr server) {
server->start(local_port, tuple, (RtpServer::TcpMode)tcp_mode, local_ip.c_str(), re_use_port, ssrc, only_track, multiplex); server->start(local_port, local_ip.c_str(), tuple, (RtpServer::TcpMode)tcp_mode, re_use_port, ssrc, only_track, multiplex);
}); });
server->setOnDetach([key](const SockException &ex) { server->setOnDetach([key](const SockException &ex) {
//设置rtp超时移除事件 //设置rtp超时移除事件
@ -1242,7 +1242,7 @@ void installWebApi() {
// 兼容老版本请求，新版本去除only_audio参数并新增only_track参数 // 兼容老版本请求，新版本去除only_audio参数并新增only_track参数
only_track = 1; only_track = 1;
} }
std::string local_ip = "::"; GET_CONFIG(std::string, local_ip, General::kListenIP)
if (!allArgs["local_ip"].empty()) { if (!allArgs["local_ip"].empty()) {
local_ip = allArgs["local_ip"]; local_ip = allArgs["local_ip"];
} }
@ -1377,10 +1377,7 @@ void installWebApi() {
} }
}); });
api_regist("/index/api/startSendRtp",[](API_ARGS_MAP_ASYNC){ static auto start_send_rtp = [] (bool passive, API_ARGS_MAP_ASYNC) {
CHECK_SECRET();
CHECK_ARGS("vhost", "app", "stream", "ssrc", "dst_url", "dst_port", "is_udp");
auto src = MediaSource::find(allArgs["vhost"], allArgs["app"], allArgs["stream"], allArgs["from_mp4"].as<int>()); auto src = MediaSource::find(allArgs["vhost"], allArgs["app"], allArgs["stream"], allArgs["from_mp4"].as<int>());
if (!src) { if (!src) {
throw ApiRetException("can not find the source stream", API::NotFound); throw ApiRetException("can not find the source stream", API::NotFound);
@ -1391,30 +1388,50 @@ void installWebApi() {
type = allArgs["use_ps"].as<int>(); type = allArgs["use_ps"].as<int>();
} }
MediaSourceEvent::SendRtpArgs args; MediaSourceEvent::SendRtpArgs args;
args.passive = false; if (passive) {
args.con_type = allArgs["is_udp"].as<bool>() ? mediakit::MediaSourceEvent::SendRtpArgs::kUdpPassive : mediakit::MediaSourceEvent::SendRtpArgs::kTcpPassive;
} else {
args.con_type = allArgs["is_udp"].as<bool>() ? mediakit::MediaSourceEvent::SendRtpArgs::kUdpActive : mediakit::MediaSourceEvent::SendRtpArgs::kTcpActive;
}
args.dst_url = allArgs["dst_url"]; args.dst_url = allArgs["dst_url"];
args.dst_port = allArgs["dst_port"]; args.dst_port = allArgs["dst_port"];
args.ssrc_multi_send = allArgs["ssrc_multi_send"].empty() ? false : allArgs["ssrc_multi_send"].as<bool>(); args.ssrc_multi_send = allArgs["ssrc_multi_send"].empty() ? false : allArgs["ssrc_multi_send"].as<bool>();
args.ssrc = allArgs["ssrc"]; args.ssrc = allArgs["ssrc"];
args.is_udp = allArgs["is_udp"];
args.src_port = allArgs["src_port"]; args.src_port = allArgs["src_port"];
args.pt = allArgs["pt"].empty() ? 96 : allArgs["pt"].as<int>(); args.pt = allArgs["pt"].empty() ? 96 : allArgs["pt"].as<int>();
args.type = (MediaSourceEvent::SendRtpArgs::Type)type; args.data_type = (MediaSourceEvent::SendRtpArgs::DataType)type;
args.only_audio = allArgs["only_audio"].as<bool>(); args.only_audio = allArgs["only_audio"].as<bool>();
args.udp_rtcp_timeout = allArgs["udp_rtcp_timeout"]; args.udp_rtcp_timeout = allArgs["udp_rtcp_timeout"];
args.recv_stream_id = allArgs["recv_stream_id"]; args.recv_stream_id = allArgs["recv_stream_id"];
TraceL << "startSendRtp, pt " << int(args.pt) << " rtp type " << type << " audio " << args.only_audio; args.close_delay_ms = allArgs["close_delay_ms"];
src->getOwnerPoller()->async([=]() mutable { src->getOwnerPoller()->async([=]() mutable {
src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable { try {
if (ex) { src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable {
val["code"] = API::OtherFailed; if (ex) {
val["msg"] = ex.what(); val["code"] = API::OtherFailed;
} val["msg"] = ex.what();
val["local_port"] = local_port; }
val["local_port"] = local_port;
invoker(200, headerOut, val.toStyledString());
});
} catch (std::exception &ex) {
val["code"] = API::Exception;
val["msg"] = ex.what();
invoker(200, headerOut, val.toStyledString()); invoker(200, headerOut, val.toStyledString());
}); }
}); });
};
api_regist("/index/api/startSendRtp",[](API_ARGS_MAP_ASYNC){
CHECK_SECRET();
CHECK_ARGS("vhost", "app", "stream", "ssrc", "dst_url", "dst_port", "is_udp");
start_send_rtp(false, API_ARGS_VALUE, invoker);
});
api_regist("/index/api/startSendRtpPassive",[](API_ARGS_MAP_ASYNC){
CHECK_SECRET();
CHECK_ARGS("vhost", "app", "stream", "ssrc");
start_send_rtp(true, API_ARGS_VALUE, invoker);
}); });
api_regist("/index/api/listRtpSender",[](API_ARGS_MAP_ASYNC){ api_regist("/index/api/listRtpSender",[](API_ARGS_MAP_ASYNC){
@ -1437,45 +1454,6 @@ void installWebApi() {
}); });
}); });
api_regist("/index/api/startSendRtpPassive",[](API_ARGS_MAP_ASYNC){
CHECK_SECRET();
CHECK_ARGS("vhost", "app", "stream", "ssrc");
auto src = MediaSource::find(allArgs["vhost"], allArgs["app"], allArgs["stream"], allArgs["from_mp4"].as<int>());
if (!src) {
throw ApiRetException("can not find the source stream", API::NotFound);
}
auto type = allArgs["type"].empty() ? (int)MediaSourceEvent::SendRtpArgs::kRtpPS : allArgs["type"].as<int>();
if (!allArgs["use_ps"].empty()) {
// 兼容之前的use_ps参数
type = allArgs["use_ps"].as<int>();
}
MediaSourceEvent::SendRtpArgs args;
args.passive = true;
args.ssrc = allArgs["ssrc"];
args.is_udp = false;
args.src_port = allArgs["src_port"];
args.pt = allArgs["pt"].empty() ? 96 : allArgs["pt"].as<int>();
args.type = (MediaSourceEvent::SendRtpArgs::Type)type;
args.only_audio = allArgs["only_audio"].as<bool>();
args.recv_stream_id = allArgs["recv_stream_id"];
//tcp被动服务器等待链接超时时间
args.tcp_passive_close_delay_ms = allArgs["close_delay_ms"];
TraceL << "startSendRtpPassive, pt " << int(args.pt) << " rtp type " << type << " audio " << args.only_audio;
src->getOwnerPoller()->async([=]() mutable {
src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable {
if (ex) {
val["code"] = API::OtherFailed;
val["msg"] = ex.what();
}
val["local_port"] = local_port;
invoker(200, headerOut, val.toStyledString());
});
});
});
api_regist("/index/api/stopSendRtp",[](API_ARGS_MAP_ASYNC){ api_regist("/index/api/stopSendRtp",[](API_ARGS_MAP_ASYNC){
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("vhost", "app", "stream"); CHECK_ARGS("vhost", "app", "stream");

23
server/main.cpp
View File

@ -281,6 +281,7 @@ int start_main(int argc,char *argv[]) {
}); });
} }
std::string listen_ip = mINI::Instance()[General::kListenIP];
uint16_t shellPort = mINI::Instance()[Shell::kPort]; uint16_t shellPort = mINI::Instance()[Shell::kPort];
uint16_t rtspPort = mINI::Instance()[Rtsp::kPort]; uint16_t rtspPort = mINI::Instance()[Rtsp::kPort];
uint16_t rtspsPort = mINI::Instance()[Rtsp::kSSLPort]; uint16_t rtspsPort = mINI::Instance()[Rtsp::kSSLPort];
@ -362,39 +363,39 @@ int start_main(int argc,char *argv[]) {
try { try {
//rtsp服务器，端口默认554 //rtsp服务器，端口默认554
if (rtspPort) { rtspSrv->start<RtspSession>(rtspPort); } if (rtspPort) { rtspSrv->start<RtspSession>(rtspPort, listen_ip); }
//rtsps服务器，端口默认322 //rtsps服务器，端口默认322
if (rtspsPort) { rtspSSLSrv->start<RtspSessionWithSSL>(rtspsPort); } if (rtspsPort) { rtspSSLSrv->start<RtspSessionWithSSL>(rtspsPort, listen_ip); }
//rtmp服务器，端口默认1935 //rtmp服务器，端口默认1935
if (rtmpPort) { rtmpSrv->start<RtmpSession>(rtmpPort); } if (rtmpPort) { rtmpSrv->start<RtmpSession>(rtmpPort, listen_ip); }
//rtmps服务器，端口默认19350 //rtmps服务器，端口默认19350
if (rtmpsPort) { rtmpsSrv->start<RtmpSessionWithSSL>(rtmpsPort); } if (rtmpsPort) { rtmpsSrv->start<RtmpSessionWithSSL>(rtmpsPort, listen_ip); }
//http服务器，端口默认80 //http服务器，端口默认80
if (httpPort) { httpSrv->start<HttpSession>(httpPort); } if (httpPort) { httpSrv->start<HttpSession>(httpPort, listen_ip); }
//https服务器，端口默认443 //https服务器，端口默认443
if (httpsPort) { httpsSrv->start<HttpsSession>(httpsPort); } if (httpsPort) { httpsSrv->start<HttpsSession>(httpsPort, listen_ip); }
//telnet远程调试服务器 //telnet远程调试服务器
if (shellPort) { shellSrv->start<ShellSession>(shellPort); } if (shellPort) { shellSrv->start<ShellSession>(shellPort, listen_ip); }
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
//创建rtp服务器 //创建rtp服务器
if (rtpPort) { rtpServer->start(rtpPort); } if (rtpPort) { rtpServer->start(rtpPort, listen_ip.c_str()); }
#endif//defined(ENABLE_RTPPROXY) #endif//defined(ENABLE_RTPPROXY)
#if defined(ENABLE_WEBRTC) #if defined(ENABLE_WEBRTC)
//webrtc udp服务器 //webrtc udp服务器
if (rtcPort) { rtcSrv_udp->start<WebRtcSession>(rtcPort);} if (rtcPort) { rtcSrv_udp->start<WebRtcSession>(rtcPort, listen_ip);}
if (rtcTcpPort) { rtcSrv_tcp->start<WebRtcSession>(rtcTcpPort);} if (rtcTcpPort) { rtcSrv_tcp->start<WebRtcSession>(rtcTcpPort, listen_ip);}
#endif//defined(ENABLE_WEBRTC) #endif//defined(ENABLE_WEBRTC)
#if defined(ENABLE_SRT) #if defined(ENABLE_SRT)
// srt udp服务器 // srt udp服务器
if (srtPort) { srtSrv->start<SRT::SrtSession>(srtPort); } if (srtPort) { srtSrv->start<SRT::SrtSession>(srtPort, listen_ip); }
#endif//defined(ENABLE_SRT) #endif//defined(ENABLE_SRT)
} catch (std::exception &ex) { } catch (std::exception &ex) {

1
src/Codec/Transcode.cpp
View File

@ -450,6 +450,7 @@ FFmpegDecoder::FFmpegDecoder(const Track::Ptr &track, int thread_num, const std:
if (codec->capabilities & AV_CODEC_CAP_TRUNCATED) { if (codec->capabilities & AV_CODEC_CAP_TRUNCATED) {
/* we do not send complete frames */ /* we do not send complete frames */
_context->flags |= AV_CODEC_FLAG_TRUNCATED; _context->flags |= AV_CODEC_FLAG_TRUNCATED;
_do_merger = false;
} else { } else {
// 此时业务层应该需要合帧 // 此时业务层应该需要合帧
_do_merger = true; _do_merger = true;

3
src/Codec/Transcode.h
View File

@ -132,7 +132,8 @@ private:
bool decodeFrame(const char *data, size_t size, uint64_t dts, uint64_t pts, bool live, bool key_frame); bool decodeFrame(const char *data, size_t size, uint64_t dts, uint64_t pts, bool live, bool key_frame);
private: private:
bool _do_merger = false; // default merge frame
bool _do_merger = true;
toolkit::Ticker _ticker; toolkit::Ticker _ticker;
onDec _cb; onDec _cb;
std::shared_ptr<AVCodecContext> _context; std::shared_ptr<AVCodecContext> _context;

39
src/Common/MediaSource.h
View File

@ -96,18 +96,29 @@ public:
class SendRtpArgs { class SendRtpArgs {
public: public:
enum Type { kRtpRAW = 0, kRtpPS = 1, kRtpTS = 2 }; enum DataType {
// 是否采用udp方式发送rtp kRtpES = 0, // 发送ES流
bool is_udp = true; kRtpPS = 1, // 发送PS流
kRtpTS = 2 // 发送TS流
};
enum ConType {
kTcpActive = 0, // tcp主动模式，tcp客户端主动连接对方并发送rtp
kUdpActive = 1, // udp主动模式，主动发送数据给对方
kTcpPassive = 2, // tcp被动模式，tcp服务器，等待对方连接并回复rtp
kUdpPassive = 3 // udp被动方式，等待对方发送nat打洞包，然后回复rtp至打洞包源地址
};
// rtp类型 // rtp类型
Type type = kRtpPS; DataType data_type = kRtpPS;
//发送es流时指定是否只发送纯音频流 // 连接类型
ConType con_type = kUdpActive;
// 发送es流时指定是否只发送纯音频流
bool only_audio = false; bool only_audio = false;
//tcp被动方式
bool passive = false;
// rtp payload type // rtp payload type
uint8_t pt = 96; uint8_t pt = 96;
//是否支持同ssrc多服务器发送 // 是否支持同ssrc多服务器发送
bool ssrc_multi_send = false; bool ssrc_multi_send = false;
// 指定rtp ssrc // 指定rtp ssrc
std::string ssrc; std::string ssrc;
@ -118,16 +129,16 @@ public:
// 发送目标主机地址，可以是ip或域名 // 发送目标主机地址，可以是ip或域名
std::string dst_url; std::string dst_url;
//udp发送时，是否开启rr rtcp接收超时判断 // udp发送时，是否开启rr rtcp接收超时判断
bool udp_rtcp_timeout = false; bool udp_rtcp_timeout = false;
//tcp被动发送服务器延时关闭事件，单位毫秒；设置为0时，则使用默认值5000ms // passive被动、tcp主动模式超时时间
uint32_t tcp_passive_close_delay_ms = 0; uint32_t close_delay_ms = 0;
//udp 发送时，rr rtcp包接收超时时间，单位毫秒 // udp 发送时，rr rtcp包接收超时时间，单位毫秒
uint32_t rtcp_timeout_ms = 30 * 1000; uint32_t rtcp_timeout_ms = 30 * 1000;
//udp 发送时，发送sr rtcp包间隔，单位毫秒 // udp 发送时，发送sr rtcp包间隔，单位毫秒
uint32_t rtcp_send_interval_ms = 5 * 1000; uint32_t rtcp_send_interval_ms = 5 * 1000;
//发送rtp同时接收，一般用于双向语言对讲, 如果不为空，说明开启接收 // 发送rtp同时接收，一般用于双向语言对讲, 如果不为空，说明开启接收
std::string recv_stream_id; std::string recv_stream_id;
}; };

21
src/Common/MultiMediaSourceMuxer.cpp
View File

@ -406,6 +406,17 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const MediaSourceE
weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this(); weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this();
rtp_sender->setOnClose([weak_self, ssrc](const toolkit::SockException &ex) {
if (auto strong_self = weak_self.lock()) {
// 可能归属线程发生变更
strong_self->getOwnerPoller(MediaSource::NullMediaSource())->async([=]() {
WarnL << "stream:" << strong_self->shortUrl() << " stop send rtp:" << ssrc << ", reason:" << ex;
strong_self->_rtp_sender.erase(ssrc);
NOTICE_EMIT(BroadcastSendRtpStoppedArgs, Broadcast::kBroadcastSendRtpStopped, *strong_self, ssrc, ex);
});
}
});
rtp_sender->startSend(args, [ssrc,ssrc_multi_send, weak_self, rtp_sender, cb, tracks, ring, poller](uint16_t local_port, const SockException &ex) mutable { rtp_sender->startSend(args, [ssrc,ssrc_multi_send, weak_self, rtp_sender, cb, tracks, ring, poller](uint16_t local_port, const SockException &ex) mutable {
cb(local_port, ex); cb(local_port, ex);
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
@ -417,16 +428,6 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const MediaSourceE
rtp_sender->addTrack(track); rtp_sender->addTrack(track);
} }
rtp_sender->addTrackCompleted(); rtp_sender->addTrackCompleted();
rtp_sender->setOnClose([weak_self, ssrc](const toolkit::SockException &ex) {
if (auto strong_self = weak_self.lock()) {
// 可能归属线程发生变更
strong_self->getOwnerPoller(MediaSource::NullMediaSource())->async([=]() {
WarnL << "stream:" << strong_self->shortUrl() << " stop send rtp:" << ssrc << ", reason:" << ex;
strong_self->_rtp_sender.erase(ssrc);
NOTICE_EMIT(BroadcastSendRtpStoppedArgs, Broadcast::kBroadcastSendRtpStopped, *strong_self, ssrc, ex);
});
}
});
auto reader = ring->attach(poller); auto reader = ring->attach(poller);
reader->setReadCB([rtp_sender](const Frame::Ptr &frame) { reader->setReadCB([rtp_sender](const Frame::Ptr &frame) {

2
src/Common/config.cpp
View File

@ -84,6 +84,7 @@ const string kWaitTrackReadyMS = GENERAL_FIELD "wait_track_ready_ms";
const string kWaitAddTrackMS = GENERAL_FIELD "wait_add_track_ms"; const string kWaitAddTrackMS = GENERAL_FIELD "wait_add_track_ms";
const string kUnreadyFrameCache = GENERAL_FIELD "unready_frame_cache"; const string kUnreadyFrameCache = GENERAL_FIELD "unready_frame_cache";
const string kBroadcastPlayerCountChanged = GENERAL_FIELD "broadcast_player_count_changed"; const string kBroadcastPlayerCountChanged = GENERAL_FIELD "broadcast_player_count_changed";
const string kListenIP = GENERAL_FIELD "listen_ip";
const string kOpusBitrate = GENERAL_FIELD"opusBitrate"; const string kOpusBitrate = GENERAL_FIELD"opusBitrate";
const string kAacBitrate = GENERAL_FIELD"aacBitrate"; const string kAacBitrate = GENERAL_FIELD"aacBitrate";
@ -103,6 +104,7 @@ static onceToken token([]() {
mINI::Instance()[kWaitAddTrackMS] = 3000; mINI::Instance()[kWaitAddTrackMS] = 3000;
mINI::Instance()[kUnreadyFrameCache] = 100; mINI::Instance()[kUnreadyFrameCache] = 100;
mINI::Instance()[kBroadcastPlayerCountChanged] = 0; mINI::Instance()[kBroadcastPlayerCountChanged] = 0;
mINI::Instance()[kListenIP] = "::";
}); });
} // namespace General } // namespace General

2
src/Common/config.h
View File

@ -202,6 +202,8 @@ extern const std::string kWaitAddTrackMS;
extern const std::string kUnreadyFrameCache; extern const std::string kUnreadyFrameCache;
// 是否启用观看人数变化事件广播，置1则启用，置0则关闭 // 是否启用观看人数变化事件广播，置1则启用，置0则关闭
extern const std::string kBroadcastPlayerCountChanged; extern const std::string kBroadcastPlayerCountChanged;
// 绑定的本地网卡ip
extern const std::string kListenIP;
extern const std::string kOpusBitrate; extern const std::string kOpusBitrate;
extern const std::string kAacBitrate; extern const std::string kAacBitrate;
} // namespace General } // namespace General

15
src/Common/macros.cpp
View File

@ -9,7 +9,6 @@
*/ */
#include "macros.h" #include "macros.h"
#include "Util/util.h"
using namespace toolkit; using namespace toolkit;
@ -17,20 +16,6 @@ using namespace toolkit;
#include "ZLMVersion.h" #include "ZLMVersion.h"
#endif #endif
extern "C" {
void Assert_Throw(int failed, const char *exp, const char *func, const char *file, int line, const char *str) {
if (failed) {
_StrPrinter printer;
printer << "Assertion failed: (" << exp ;
if(str && *str){
printer << ", " << str;
}
printer << "), function " << func << ", file " << file << ", line " << line << ".";
throw mediakit::AssertFailedException(printer);
}
}
}
namespace mediakit { namespace mediakit {
/** /**

21
src/Common/macros.h
View File

@ -11,9 +11,10 @@
#ifndef ZLMEDIAKIT_MACROS_H #ifndef ZLMEDIAKIT_MACROS_H
#define ZLMEDIAKIT_MACROS_H #define ZLMEDIAKIT_MACROS_H
#include "Util/logger.h"
#include <iostream>
#include <sstream> #include <sstream>
#include <iostream>
#include "Util/util.h"
#include "Util/logger.h"
#if defined(__MACH__) #if defined(__MACH__)
#include <arpa/inet.h> #include <arpa/inet.h>
#include <machine/endian.h> #include <machine/endian.h>
@ -40,7 +41,7 @@
#define CHECK_RET(...) \ #define CHECK_RET(...) \
try { \ try { \
CHECK(__VA_ARGS__); \ CHECK(__VA_ARGS__); \
} catch (AssertFailedException & ex) { \ } catch (toolkit::AssertFailedException & ex) { \
WarnL << ex.what(); \ WarnL << ex.what(); \
return; \ return; \
} }
@ -71,22 +72,8 @@
#define VHOST_KEY "vhost" #define VHOST_KEY "vhost"
#define DEFAULT_VHOST "__defaultVhost__" #define DEFAULT_VHOST "__defaultVhost__"
#ifdef __cplusplus
extern "C" {
#endif
extern void Assert_Throw(int failed, const char *exp, const char *func, const char *file, int line, const char *str);
#ifdef __cplusplus
}
#endif
namespace mediakit { namespace mediakit {
class AssertFailedException : public std::runtime_error {
public:
template<typename ...T>
AssertFailedException(T && ...args) : std::runtime_error(std::forward<T>(args)...) {}
};
extern const char kServerName[]; extern const char kServerName[];
template <typename... ARGS> template <typename... ARGS>

4
src/Player/PlayerProxy.cpp
View File

@ -191,13 +191,13 @@ void PlayerProxy::setDirectProxy() {
if (dynamic_pointer_cast<RtspPlayer>(_delegate)) { if (dynamic_pointer_cast<RtspPlayer>(_delegate)) {
// rtsp拉流 // rtsp拉流
GET_CONFIG(bool, directProxy, Rtsp::kDirectProxy); GET_CONFIG(bool, directProxy, Rtsp::kDirectProxy);
if (directProxy) { if (directProxy && _option.enable_rtsp) {
mediaSource = std::make_shared<RtspMediaSource>(_tuple); mediaSource = std::make_shared<RtspMediaSource>(_tuple);
} }
} else if (dynamic_pointer_cast<RtmpPlayer>(_delegate)) { } else if (dynamic_pointer_cast<RtmpPlayer>(_delegate)) {
// rtmp拉流 // rtmp拉流
GET_CONFIG(bool, directProxy, Rtmp::kDirectProxy); GET_CONFIG(bool, directProxy, Rtmp::kDirectProxy);
if (directProxy) { if (directProxy && _option.enable_rtmp) {
mediaSource = std::make_shared<RtmpMediaSource>(_tuple); mediaSource = std::make_shared<RtmpMediaSource>(_tuple);
} }
} }

2
src/Player/PlayerProxy.h
View File

@ -129,12 +129,12 @@ private:
void setTranslationInfo(); void setTranslationInfo();
private: private:
ProtocolOption _option;
int _retry_count; int _retry_count;
int _reconnect_delay_min; int _reconnect_delay_min;
int _reconnect_delay_max; int _reconnect_delay_max;
int _reconnect_delay_step; int _reconnect_delay_step;
MediaTuple _tuple; MediaTuple _tuple;
ProtocolOption _option;
std::string _pull_url; std::string _pull_url;
toolkit::Timer::Ptr _timer; toolkit::Timer::Ptr _timer;
std::function<void()> _on_disconnect; std::function<void()> _on_disconnect;

15
src/Record/MP4Muxer.cpp
View File

@ -99,6 +99,20 @@ bool MP4MuxerInterface::inputFrame(const Frame::Ptr &frame) {
_started = true; _started = true;
} }
// fmp4封装超过一定I帧间隔，强制刷新segment，防止内存上涨
if (frame->getTrackType() == TrackVideo && _mov_writter->fmp4) {
if (frame->keyFrame()) {
_non_iframe_video_count = 0;
} else {
_non_iframe_video_count++;
}
if (_non_iframe_video_count > 200) {
saveSegment();
_non_iframe_video_count = 0;
}
}
// mp4文件时间戳需要从0开始 // mp4文件时间戳需要从0开始
auto &track = it->second; auto &track = it->second;
switch (frame->getCodecId()) { switch (frame->getCodecId()) {
@ -164,6 +178,7 @@ bool MP4MuxerInterface::addTrack(const Track::Ptr &track) {
} }
_tracks[track->getIndex()].track_id = track_id; _tracks[track->getIndex()].track_id = track_id;
_have_video = true; _have_video = true;
_non_iframe_video_count = 0;
} else if (track->getTrackType() == TrackAudio) { } else if (track->getTrackType() == TrackAudio) {
auto audio_track = dynamic_pointer_cast<AudioTrack>(track); auto audio_track = dynamic_pointer_cast<AudioTrack>(track);
CHECK(audio_track); CHECK(audio_track);

1
src/Record/MP4Muxer.h
View File

@ -72,6 +72,7 @@ private:
bool _started = false; bool _started = false;
bool _have_video = false; bool _have_video = false;
MP4FileIO::Writer _mov_writter; MP4FileIO::Writer _mov_writter;
int _non_iframe_video_count; // 非I帧个数
class FrameMergerImp : public FrameMerger { class FrameMergerImp : public FrameMerger {
public: public:

2
src/Rtp/PSDecoder.cpp
View File

@ -63,7 +63,7 @@ const char *PSDecoder::onSearchPacketTail(const char *data, size_t len) {
//解析失败，丢弃所有数据 //解析失败，丢弃所有数据
return data + len; return data + len;
} catch (AssertFailedException &ex) { } catch (toolkit::AssertFailedException &ex) {
InfoL << "解析 ps 异常: bytes=" << len InfoL << "解析 ps 异常: bytes=" << len
<< ", exception=" << ex.what() << ", exception=" << ex.what()
<< ", hex=" << hexdump(data, MIN(len, 32)); << ", hex=" << hexdump(data, MIN(len, 32));

2
src/Rtp/RtpCache.h
View File

@ -41,7 +41,7 @@ private:
class RtpCachePS : public RtpCache, public PSEncoderImp { class RtpCachePS : public RtpCache, public PSEncoderImp {
public: public:
RtpCachePS(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96, bool ps_or_ts = true) : RtpCachePS(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96, bool ps_or_ts = true) :
RtpCache(std::move(cb)), PSEncoderImp(ssrc, ps_or_ts ? payload_type : Rtsp::PT_MP2T, ps_or_ts) {}; RtpCache(std::move(cb)), PSEncoderImp(ssrc, ps_or_ts ? payload_type : static_cast<int>(Rtsp::PT_MP2T), ps_or_ts) {};
void flush() override; void flush() override;

1
src/Rtp/RtpProcess.cpp
View File

@ -30,6 +30,7 @@ RtpProcess::Ptr RtpProcess::createProcess(const MediaTuple &tuple) {
} }
RtpProcess::RtpProcess(const MediaTuple &tuple) { RtpProcess::RtpProcess(const MediaTuple &tuple) {
_media_info.schema = "rtp";
static_cast<MediaTuple &>(_media_info) = tuple; static_cast<MediaTuple &>(_media_info) = tuple;
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir); GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);

254
src/Rtp/RtpSender.cpp
View File

@ -40,86 +40,99 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
if (!_interface) { if (!_interface) {
//重连时不重新创建对象 //重连时不重新创建对象
auto lam = [this](std::shared_ptr<List<Buffer::Ptr>> list) { onFlushRtpList(std::move(list)); }; auto lam = [this](std::shared_ptr<List<Buffer::Ptr>> list) { onFlushRtpList(std::move(list)); };
switch (args.type) { switch (args.data_type) {
case MediaSourceEvent::SendRtpArgs::kRtpPS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, true); break; case MediaSourceEvent::SendRtpArgs::kRtpPS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, true); break;
case MediaSourceEvent::SendRtpArgs::kRtpTS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, false); break; case MediaSourceEvent::SendRtpArgs::kRtpTS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, false); break;
case MediaSourceEvent::SendRtpArgs::kRtpRAW: _interface = std::make_shared<RtpCacheRaw>(lam, atoi(args.ssrc.data()), args.pt, args.only_audio); break; case MediaSourceEvent::SendRtpArgs::kRtpES: _interface = std::make_shared<RtpCacheRaw>(lam, atoi(args.ssrc.data()), args.pt, args.only_audio); break;
default: CHECK(0, "invalid rtp type:" + to_string(args.type)); break; default: CHECK(0, "invalid rtp type: " + to_string(args.data_type)); break;
} }
} }
auto delay_ms = _args.close_delay_ms ? _args.close_delay_ms : 5000;
weak_ptr<RtpSender> weak_self = shared_from_this(); weak_ptr<RtpSender> weak_self = shared_from_this();
if (args.passive) { if (args.con_type == MediaSourceEvent::SendRtpArgs::kTcpPassive) {
// tcp被动发流模式 auto tcp_listener = Socket::createSocket(_poller, false);
_args.is_udp = false; if (args.src_port) {
// 默认等待链接 // 指定端口
bool is_wait = true; if (!tcp_listener->listen(args.src_port)) {
try { throw std::invalid_argument(StrPrinter << "open tcp passive server failed on port: " << args.src_port << ", err: " << get_uv_errmsg(true));
auto tcp_listener = Socket::createSocket(_poller, false);
if (args.src_port) {
//指定端口
if (!tcp_listener->listen(args.src_port)) {
throw std::invalid_argument(StrPrinter << "open tcp passive server failed on port:" << args.src_port
<< ", err:" << get_uv_errmsg(true));
}
is_wait = true;
} else {
auto pr = std::make_pair(tcp_listener, Socket::createSocket(_poller, false));
//从端口池获取随机端口
makeSockPair(pr, "::", false, false);
// 随机端口不等待，保证调用者可以知道端口
is_wait = false;
} }
// tcp服务器默认开启5秒 } else {
auto delay = _args.tcp_passive_close_delay_ms ? _args.tcp_passive_close_delay_ms : 5000; auto pr = std::make_pair(tcp_listener, Socket::createSocket(_poller, false));
auto delay_task = _poller->doDelayTask(delay, [tcp_listener, cb, is_wait]() mutable { // 从端口池获取随机端口
if (is_wait) { makeSockPair(pr, "::", true, false);
cb(0, SockException(Err_timeout, "wait tcp connection timeout"));
}
tcp_listener = nullptr;
return 0;
});
tcp_listener->setOnAccept([weak_self, cb, delay_task,is_wait](Socket::Ptr &sock, std::shared_ptr<void> &complete) {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
//立即关闭tcp服务器
delay_task->cancel();
strong_self->_socket_rtp = sock;
strong_self->onConnect();
if (is_wait) {
cb(sock->get_local_port(), SockException());
}
InfoL << "accept connection from:" << sock->get_peer_ip() << ":" << sock->get_peer_port();
});
InfoL << "start tcp passive server on:" << tcp_listener->get_local_port();
if (!is_wait) {
// 随机端口马上返回端口，保证调用者知道端口
cb(tcp_listener->get_local_port(), SockException());
}
} catch (std::exception &ex) {
cb(0, SockException(Err_other, ex.what()));
return;
} }
return; // 定时器持有tcp_listener，保证超时时间内保持监听
} auto delay_task = _poller->doDelayTask(delay_ms, [weak_self, tcp_listener]() mutable {
if (args.is_udp) { // 防止循环引用
tcp_listener = nullptr;
if (auto strong_self = weak_self.lock()) {
strong_self->onClose(SockException(Err_timeout, "wait tcp connection timeout"));
}
return 0;
});
tcp_listener->setOnAccept([weak_self, delay_task](Socket::Ptr &sock, std::shared_ptr<void> &complete) {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
delay_task->cancel();
strong_self->_socket_rtp = sock;
strong_self->onConnect();
InfoL << "accept tcp connection from: " << sock->get_peer_ip() << ":" << sock->get_peer_port();
});
InfoL << "start tcp passive server on: " << tcp_listener->get_local_port();
cb(tcp_listener->get_local_port(), SockException());
} else if (args.con_type == MediaSourceEvent::SendRtpArgs::kUdpPassive) {
if (args.src_port) {
// 指定端口
if (!_socket_rtp->bindUdpSock(args.src_port, "::", true)) {
throw std::invalid_argument(StrPrinter << "open udp passive server failed on port: " << args.src_port << ", err: " << get_uv_errmsg(true));
}
} else {
auto pr = std::make_pair(_socket_rtp, Socket::createSocket(_poller, false));
// 从端口池获取随机端口
makeSockPair(pr, "::", true, true);
}
auto delay_task = _poller->doDelayTask(delay_ms, [weak_self]() mutable {
if (auto strong_self = weak_self.lock()) {
// 关闭端口
strong_self->_socket_rtp->closeSock();
strong_self->onClose(SockException(Err_timeout, "wait udp connection timeout"));
}
return 0;
});
_socket_rtp->setOnRead([weak_self, delay_task](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
delay_task->cancel();
strong_self->_socket_rtp->bindPeerAddr(addr, addr_len, true);
// 异步执行onConnect，防止在OnRead回调中调用setOnRead
strong_self->_poller->async([strong_self]() { strong_self->onConnect(); }, false);
InfoL << "accept udp connection from: " << strong_self->_socket_rtp->get_peer_ip() << ":" << strong_self->_socket_rtp->get_peer_port();
});
InfoL << "start udp passive server on: " << _socket_rtp->get_local_port();
cb(_socket_rtp->get_local_port(), SockException());
} else if (args.con_type == MediaSourceEvent::SendRtpArgs::kUdpActive) {
auto poller = _poller; auto poller = _poller;
WorkThreadPool::Instance().getPoller()->async([cb, args, weak_self, poller]() { WorkThreadPool::Instance().getPoller()->async([cb, args, weak_self, poller]() {
struct sockaddr_storage addr; struct sockaddr_storage addr;
//切换线程目的是为了dns解析放在后台线程执行 // 切换线程目的是为了dns解析放在后台线程执行
if (!SockUtil::getDomainIP(args.dst_url.data(), args.dst_port, addr, AF_INET, SOCK_DGRAM, IPPROTO_UDP)) { if (!SockUtil::getDomainIP(args.dst_url.data(), args.dst_port, addr, AF_INET, SOCK_DGRAM, IPPROTO_UDP)) {
poller->async([args, cb]() { poller->async([args, cb]() {
//切回自己的线程 // 切回自己的线程
cb(0, SockException(Err_dns, StrPrinter << "dns解析域名失败:" << args.dst_url)); cb(0, SockException(Err_dns, StrPrinter << "dns resolution failed: " << args.dst_url));
}); });
return; return;
} }
//dns解析成功 // dns解析成功
poller->async([args, addr, weak_self, cb]() { poller->async([args, addr, weak_self, cb]() {
//切回自己的线程 // 切回自己的线程
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
return; return;
@ -127,15 +140,14 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
string ifr_ip = addr.ss_family == AF_INET ? "0.0.0.0" : "::"; string ifr_ip = addr.ss_family == AF_INET ? "0.0.0.0" : "::";
try { try {
if (args.src_port) { if (args.src_port) {
//指定端口 // 指定端口
if (!strong_self->_socket_rtp->bindUdpSock(args.src_port, ifr_ip)) { if (!strong_self->_socket_rtp->bindUdpSock(args.src_port, ifr_ip, true)) {
throw std::invalid_argument(StrPrinter << "bindUdpSock failed on port:" << args.src_port throw std::invalid_argument(StrPrinter << "open udp active client failed on port: " << args.src_port << ", err: " << get_uv_errmsg(true));
<< ", err:" << get_uv_errmsg(true));
} }
} else { } else {
auto pr = std::make_pair(strong_self->_socket_rtp, Socket::createSocket(strong_self->_poller, false)); auto pr = std::make_pair(strong_self->_socket_rtp, Socket::createSocket(strong_self->_poller, false));
//从端口池获取随机端口 // 从端口池获取随机端口
makeSockPair(pr, ifr_ip, true); makeSockPair(pr, ifr_ip, true, true);
} }
} catch (std::exception &ex) { } catch (std::exception &ex) {
cb(0, SockException(Err_other, ex.what())); cb(0, SockException(Err_other, ex.what()));
@ -146,19 +158,24 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
cb(strong_self->_socket_rtp->get_local_port(), SockException()); cb(strong_self->_socket_rtp->get_local_port(), SockException());
}); });
}); });
} else { InfoL << "start udp active send rtp to: " << args.dst_url << ":" << args.dst_port;
_socket_rtp->connect(args.dst_url, args.dst_port, [cb, weak_self](const SockException &err) {
} else if (args.con_type == MediaSourceEvent::SendRtpArgs::kTcpActive) {
_socket_rtp->connect(args.dst_url, args.dst_port,[cb, weak_self](const SockException &err) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (strong_self) { if (strong_self) {
if (!err) { if (!err) {
//tcp连接成功 // tcp连接成功
strong_self->onConnect(); strong_self->onConnect();
} }
cb(strong_self->_socket_rtp->get_local_port(), err); cb(strong_self->_socket_rtp->get_local_port(), err);
} else { } else {
cb(0, err); cb(0, err);
} }
}, 5.0F, "::", args.src_port); }, delay_ms / 1000.0, "::", args.src_port);
InfoL << "start tcp active send rtp to: " << args.dst_url << ":" << args.dst_port;
} else {
CHECK(0, "invalid con type");
} }
} }
@ -168,8 +185,8 @@ void RtpSender::createRtcpSocket() {
} }
_socket_rtcp = Socket::createSocket(_socket_rtp->getPoller(), false); _socket_rtcp = Socket::createSocket(_socket_rtp->getPoller(), false);
//rtcp端口使用户rtp端口+1 //rtcp端口使用户rtp端口+1
if(!_socket_rtcp->bindUdpSock(_socket_rtp->get_local_port() + 1, _socket_rtp->get_local_ip(), false)){ if(!_socket_rtcp->bindUdpSock(_socket_rtp->get_local_port() + 1, _socket_rtp->get_local_ip(), true)){
WarnL << "bind rtcp udp socket failed:" << get_uv_errmsg(true); WarnL << "bind rtcp udp socket failed: " << get_uv_errmsg(true);
_socket_rtcp = nullptr; _socket_rtcp = nullptr;
return; return;
} }
@ -180,12 +197,18 @@ void RtpSender::createRtcpSocket() {
_rtcp_context = std::make_shared<RtcpContextForSend>(); _rtcp_context = std::make_shared<RtcpContextForSend>();
weak_ptr<RtpSender> weak_self = shared_from_this(); weak_ptr<RtpSender> weak_self = shared_from_this();
_socket_rtcp->setOnRead([weak_self](const Buffer::Ptr &buf, struct sockaddr *, int) { bool bind_addr = false;
_socket_rtcp->setOnRead([weak_self, bind_addr](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) mutable {
//接收receive report rtcp //接收receive report rtcp
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
return; return;
} }
if (!bind_addr) {
// 收到对方rtcp打洞包后，再回复rtcp
bind_addr = true;
strong_self->_socket_rtcp->bindPeerAddr(addr, addr_len, true);
}
auto rtcp_arr = RtcpHeader::loadFromBytes(buf->data(), buf->size()); auto rtcp_arr = RtcpHeader::loadFromBytes(buf->data(), buf->size());
for (auto &rtcp : rtcp_arr) { for (auto &rtcp : rtcp_arr) {
strong_self->onRecvRtcp(rtcp); strong_self->onRecvRtcp(rtcp);
@ -199,19 +222,19 @@ void RtpSender::onRecvRtcp(RtcpHeader *rtcp) {
_rtcp_recv_ticker.resetTime(); _rtcp_recv_ticker.resetTime();
} }
//连接建立成功事件 // 连接建立成功事件
void RtpSender::onConnect(){ void RtpSender::onConnect() {
_is_connect = true; _is_connect = true;
//加大发送缓存,防止udp丢包之类的问题 // 加大发送缓存,防止udp丢包之类的问题
SockUtil::setSendBuf(_socket_rtp->rawFD(), 4 * 1024 * 1024); SockUtil::setSendBuf(_socket_rtp->rawFD(), 4 * 1024 * 1024);
if (!_args.is_udp) { if (_args.con_type == MediaSourceEvent::SendRtpArgs::kTcpActive || _args.con_type == MediaSourceEvent::SendRtpArgs::kTcpPassive) {
//关闭tcp no_delay并开启MSG_MORE, 提高发送性能 // 关闭tcp no_delay并开启MSG_MORE, 提高发送性能
SockUtil::setNoDelay(_socket_rtp->rawFD(), false); SockUtil::setNoDelay(_socket_rtp->rawFD(), false);
_socket_rtp->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE); _socket_rtp->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
} else if (_args.udp_rtcp_timeout) { } else if (_args.udp_rtcp_timeout) {
createRtcpSocket(); createRtcpSocket();
} }
//连接建立成功事件 // 连接建立成功事件
weak_ptr<RtpSender> weak_self = shared_from_this(); weak_ptr<RtpSender> weak_self = shared_from_this();
if (!_args.recv_stream_id.empty()) { if (!_args.recv_stream_id.empty()) {
mINI ini; mINI ini;
@ -226,11 +249,13 @@ void RtpSender::onConnect(){
} }
try { try {
strong_self->_rtp_session->onRecv(buf); strong_self->_rtp_session->onRecv(buf);
} catch (std::exception &ex){ } catch (std::exception &ex) {
SockException err(toolkit::Err_shutdown, ex.what()); SockException err(toolkit::Err_shutdown, ex.what());
strong_self->_rtp_session->shutdown(err); strong_self->_rtp_session->shutdown(err);
} }
}); });
} else {
_socket_rtp->setOnRead(nullptr);
} }
_socket_rtp->setOnErr([weak_self](const SockException &err) { _socket_rtp->setOnErr([weak_self](const SockException &err) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
@ -238,12 +263,10 @@ void RtpSender::onConnect(){
strong_self->onErr(err); strong_self->onErr(err);
} }
}); });
//获取本地端口，断开重连后确保端口不变 InfoL << "startSend rtp success: " << _socket_rtp->get_peer_ip() << ":" << _socket_rtp->get_peer_port() << ", data_type: " << _args.data_type << ", con_type: " << _args.con_type;
_args.src_port = _socket_rtp->get_local_port();
InfoL << "开始发送 rtp:" << _socket_rtp->get_peer_ip() << ":" << _socket_rtp->get_peer_port() << ", 是否为udp方式:" << _args.is_udp;
} }
bool RtpSender::addTrack(const Track::Ptr &track){ bool RtpSender::addTrack(const Track::Ptr &track) {
if (_args.only_audio && track->getTrackType() == TrackVideo) { if (_args.only_audio && track->getTrackType() == TrackVideo) {
// 如果只发送音频则忽略视频 // 如果只发送音频则忽略视频
return false; return false;
@ -251,11 +274,11 @@ bool RtpSender::addTrack(const Track::Ptr &track){
return _interface->addTrack(track); return _interface->addTrack(track);
} }
void RtpSender::addTrackCompleted(){ void RtpSender::addTrackCompleted() {
_interface->addTrackCompleted(); _interface->addTrackCompleted();
} }
void RtpSender::resetTracks(){ void RtpSender::resetTracks() {
_interface->resetTracks(); _interface->resetTracks();
} }
@ -265,13 +288,12 @@ void RtpSender::flush() {
} }
} }
//此函数在其他线程执行
bool RtpSender::inputFrame(const Frame::Ptr &frame) { bool RtpSender::inputFrame(const Frame::Ptr &frame) {
if (_args.only_audio && frame->getTrackType() == TrackVideo) { if (_args.only_audio && frame->getTrackType() == TrackVideo) {
// 如果只发送音频则忽略视频 // 如果只发送音频则忽略视频
return false; return false;
} }
//连接成功后才做实质操作(节省cpu资源) // 连接成功后才做实质操作(节省cpu资源)
return _is_connect ? _interface->inputFrame(frame) : false; return _is_connect ? _interface->inputFrame(frame) : false;
} }
@ -283,20 +305,20 @@ void RtpSender::onSendRtpUdp(const toolkit::Buffer::Ptr &buf, bool check) {
_rtcp_context->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->ntp_stamp, 90000 /*not used*/, rtp->size()); _rtcp_context->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->ntp_stamp, 90000 /*not used*/, rtp->size());
if (!check) { if (!check) {
//减少判断次数 // 减少判断次数
return; return;
} }
//每5秒发送一次rtcp // 每5秒发送一次rtcp
if (_rtcp_send_ticker.elapsedTime() > _args.rtcp_send_interval_ms) { if (_rtcp_send_ticker.elapsedTime() > _args.rtcp_send_interval_ms) {
_rtcp_send_ticker.resetTime(); _rtcp_send_ticker.resetTime();
//rtcp ssrc为rtp ssrc + 1 // rtcp ssrc为rtp ssrc + 1
auto sr = _rtcp_context->createRtcpSR(atoi(_args.ssrc.data()) + 1); auto sr = _rtcp_context->createRtcpSR(atoi(_args.ssrc.data()) + 1);
//send sender report rtcp // send sender report rtcp
_socket_rtcp->send(sr); _socket_rtcp->send(sr);
} }
if (_rtcp_recv_ticker.elapsedTime() > _args.rtcp_timeout_ms) { if (_rtcp_recv_ticker.elapsedTime() > _args.rtcp_timeout_ms) {
//接收rr rtcp超时 // 接收rr rtcp超时
WarnL << "recv rr rtcp timeout"; WarnL << "recv rr rtcp timeout";
_rtcp_recv_ticker.resetTime(); _rtcp_recv_ticker.resetTime();
onClose(SockException(Err_timeout, "recv rr rtcp timeout")); onClose(SockException(Err_timeout, "recv rr rtcp timeout"));
@ -306,28 +328,36 @@ void RtpSender::onSendRtpUdp(const toolkit::Buffer::Ptr &buf, bool check) {
void RtpSender::onClose(const SockException &ex) { void RtpSender::onClose(const SockException &ex) {
auto cb = _on_close; auto cb = _on_close;
if (cb) { if (cb) {
//在下次循环时触发onClose，原因是防止遍历map时删除元素 // 在下次循环时触发onClose，原因是防止遍历map时删除元素
_poller->async([cb, ex]() { cb(ex); }, false); _poller->async([cb, ex]() { cb(ex); }, false);
} }
} }
//此函数在其他线程执行 // 此函数在其他线程执行
void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr> > rtp_list) { void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr>> rtp_list) {
if(!_is_connect){ if (!_is_connect) {
//连接成功后才能发送数据 // 连接成功后才能发送数据
return; return;
} }
size_t i = 0; size_t i = 0;
auto size = rtp_list->size(); auto size = rtp_list->size();
rtp_list->for_each([&](Buffer::Ptr &packet) { rtp_list->for_each([&](Buffer::Ptr &packet) {
if (_args.is_udp) { switch (_args.con_type) {
onSendRtpUdp(packet, i == 0); case MediaSourceEvent::SendRtpArgs::kUdpActive:
// udp模式，rtp over tcp前4个字节可以忽略 case MediaSourceEvent::SendRtpArgs::kUdpPassive: {
_socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), RtpPacket::kRtpTcpHeaderSize), nullptr, 0, ++i == size); onSendRtpUdp(packet, i == 0);
} else { // udp模式，rtp over tcp前4个字节可以忽略
// tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节 _socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), RtpPacket::kRtpTcpHeaderSize), nullptr, 0, ++i == size);
_socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size); break;
}
case MediaSourceEvent::SendRtpArgs::kTcpActive:
case MediaSourceEvent::SendRtpArgs::kTcpPassive: {
// tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节
_socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
break;
}
default: CHECK(0);
} }
}); });
} }
@ -338,9 +368,9 @@ void RtpSender::onErr(const SockException &ex) {
onClose(ex); onClose(ex);
} }
void RtpSender::setOnClose(std::function<void(const toolkit::SockException &ex)> on_close){ void RtpSender::setOnClose(std::function<void(const toolkit::SockException &ex)> on_close) {
_on_close = std::move(on_close); _on_close = std::move(on_close);
} }
}//namespace mediakit } // namespace mediakit
#endif// defined(ENABLE_RTPPROXY) #endif // defined(ENABLE_RTPPROXY)

3
src/Rtp/RtpSender.h
View File

@ -63,6 +63,9 @@ public:
*/ */
virtual void resetTracks() override; virtual void resetTracks() override;
/**
* 设置发送rtp停止回调
*/
void setOnClose(std::function<void(const toolkit::SockException &ex)> on_close); void setOnClose(std::function<void(const toolkit::SockException &ex)> on_close);
private: private:

2
src/Rtp/RtpServer.cpp
View File

@ -122,7 +122,7 @@ private:
std::shared_ptr<struct sockaddr_storage> _rtcp_addr; std::shared_ptr<struct sockaddr_storage> _rtcp_addr;
}; };
void RtpServer::start(uint16_t local_port, const MediaTuple &tuple, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) { void RtpServer::start(uint16_t local_port, const char *local_ip, const MediaTuple &tuple, TcpMode tcp_mode, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) {
//创建udp服务器 //创建udp服务器
auto poller = EventPollerPool::Instance().getPoller(); auto poller = EventPollerPool::Instance().getPoller();
Socket::Ptr rtp_socket = Socket::createSocket(poller, true); Socket::Ptr rtp_socket = Socket::createSocket(poller, true);

6
src/Rtp/RtpServer.h
View File

@ -36,15 +36,15 @@ public:
/** /**
* 开启服务器，可能抛异常 * 开启服务器，可能抛异常
* @param local_port 本地端口，0时为随机端口 * @param local_port 本地端口，0时为随机端口
* @param local_ip 绑定的本地网卡ip
* @param stream_id 流id，置空则使用ssrc * @param stream_id 流id，置空则使用ssrc
* @param tcp_mode tcp服务模式 * @param tcp_mode tcp服务模式
* @param local_ip 绑定的本地网卡ip
* @param re_use_port 是否设置socket为re_use属性 * @param re_use_port 是否设置socket为re_use属性
* @param ssrc 指定的ssrc * @param ssrc 指定的ssrc
* @param multiplex 多路复用 * @param multiplex 多路复用
*/ */
void start(uint16_t local_port, const MediaTuple &tuple = MediaTuple{DEFAULT_VHOST, kRtpAppName, "", ""}, TcpMode tcp_mode = PASSIVE, void start(uint16_t local_port, const char *local_ip = "::", const MediaTuple &tuple = MediaTuple{DEFAULT_VHOST, kRtpAppName, "", ""}, TcpMode tcp_mode = PASSIVE,
const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0, int only_track = 0, bool multiplex = false); bool re_use_port = true, uint32_t ssrc = 0, int only_track = 0, bool multiplex = false);
/** /**
* 连接到tcp服务(tcp主动模式) * 连接到tcp服务(tcp主动模式)

6
src/Rtp/RtpSplitter.cpp
View File

@ -46,7 +46,7 @@ const char *RtpSplitter::onSearchPacketTail(const char *data, size_t len) {
return nullptr; return nullptr;
} }
if ( _is_ehome ) { if (_check_ehome_count) {
if (isEhome(data, len)) { if (isEhome(data, len)) {
//是ehome协议 //是ehome协议
if (len < kEHOME_OFFSET + 4) { if (len < kEHOME_OFFSET + 4) {
@ -59,7 +59,7 @@ const char *RtpSplitter::onSearchPacketTail(const char *data, size_t len) {
//忽略ehome私有头 //忽略ehome私有头
return onSearchPacketTail_l(data + kEHOME_OFFSET + 2, len - kEHOME_OFFSET - 2); return onSearchPacketTail_l(data + kEHOME_OFFSET + 2, len - kEHOME_OFFSET - 2);
} }
_is_ehome = false; _check_ehome_count--;
} }
if ( _is_rtsp_interleaved ) { if ( _is_rtsp_interleaved ) {
@ -70,7 +70,7 @@ const char *RtpSplitter::onSearchPacketTail(const char *data, size_t len) {
} }
_is_rtsp_interleaved = false; _is_rtsp_interleaved = false;
} }
//两个字节的rtp头 //两个字节的rtp头
_offset = 2; _offset = 2;
return onSearchPacketTail_l(data, len); return onSearchPacketTail_l(data, len);

3
src/Rtp/RtpSplitter.h
View File

@ -31,7 +31,8 @@ protected:
const char *onSearchPacketTail_l(const char *data, size_t len); const char *onSearchPacketTail_l(const char *data, size_t len);
private: private:
bool _is_ehome = true; bool _is_ehome = false;
int _check_ehome_count = 3;
bool _is_rtsp_interleaved = true; bool _is_rtsp_interleaved = true;
size_t _offset = 0; size_t _offset = 0;
}; };

16
src/Rtsp/Rtsp.cpp
View File

@ -15,6 +15,7 @@
#include "Common/Parser.h" #include "Common/Parser.h"
#include "Common/config.h" #include "Common/config.h"
#include "Network/Socket.h" #include "Network/Socket.h"
#include "Extension/Factory.h"
using namespace std; using namespace std;
using namespace toolkit; using namespace toolkit;
@ -236,10 +237,6 @@ void SdpParser::load(const string &sdp) {
track._codec = codec; track._codec = codec;
track._samplerate = samplerate; track._samplerate = samplerate;
} }
if (!track._samplerate && track._type == TrackVideo) {
// 未设置视频采样率时，赋值为90000
track._samplerate = 90000;
}
++it; ++it;
} }
@ -260,6 +257,17 @@ void SdpParser::load(const string &sdp) {
if (it != track._attr.end()) { if (it != track._attr.end()) {
track._control = it->second; track._control = it->second;
} }
if (!track._samplerate && track._type == TrackVideo) {
// 未设置视频采样率时，赋值为90000
track._samplerate = 90000;
} else if (!track._samplerate && track._type == TrackAudio) {
// some rtsp sdp no sample rate but has fmt config to parser get sample rate
auto t = Factory::getTrackBySdp(track_ptr);
if (t) {
track._samplerate = std::static_pointer_cast<AudioTrack>(t)->getAudioSampleRate();
}
}
} }
} }

2
src/Rtsp/RtspSplitter.cpp
View File

@ -72,7 +72,7 @@ ssize_t RtspSplitter::onRecvHeader(const char *data, size_t len) {
} }
try { try {
_parser.parse(data, len); _parser.parse(data, len);
} catch (mediakit::AssertFailedException &ex){ } catch (toolkit::AssertFailedException &ex){
if (!_enableRecvRtp) { if (!_enableRecvRtp) {
// 还在握手中，直接中断握手 // 还在握手中，直接中断握手
throw; throw;

4
tests/test_bench_forward.cpp
View File

@ -223,8 +223,8 @@ int main(int argc, char *argv[]) {
option.enable_hls = false; option.enable_hls = false;
option.enable_mp4 = false; option.enable_mp4 = false;
option.modify_stamp = (int)ProtocolOption::kModifyStampRelative; option.modify_stamp = (int)ProtocolOption::kModifyStampRelative;
//添加拉流代理 //添加拉流代理
auto tuple = MediaTuple{DEFAULT_VHOST, "app", std::to_string(i)}; auto tuple = MediaTuple { DEFAULT_VHOST, "app", std::to_string(i), "" };
auto proxy = std::make_shared<PlayerProxy>(tuple, option, -1, nullptr, 1); auto proxy = std::make_shared<PlayerProxy>(tuple, option, -1, nullptr, 1);
//开始拉流代理 //开始拉流代理
proxy->play(input_urls[i]); proxy->play(input_urls[i]);

2
tests/test_bench_push.cpp
View File

@ -148,7 +148,7 @@ int main(int argc, char *argv[]) {
MediaSource::Ptr src = nullptr; MediaSource::Ptr src = nullptr;
PlayerProxy::Ptr proxy = nullptr;; PlayerProxy::Ptr proxy = nullptr;;
auto tuple = MediaTuple{DEFAULT_VHOST, app, stream}; auto tuple = MediaTuple { DEFAULT_VHOST, app, stream, "" };
if (end_with(in_url, ".mp4")) { if (end_with(in_url, ".mp4")) {
// create MediaSource from mp4file // create MediaSource from mp4file
auto reader = std::make_shared<MP4Reader>(tuple, in_url); auto reader = std::make_shared<MP4Reader>(tuple, in_url);

2
tests/test_rtp.cpp
View File

@ -75,7 +75,7 @@ static bool loadFile(const char *path, const EventPoller::Ptr &poller) {
return 0; return 0;
} }
auto diff = stamp - stamp_last; auto diff = static_cast<int64_t>(stamp - stamp_last);
if (diff < 0 || diff > 500) { if (diff < 0 || diff > 500) {
diff = 1; diff = 1;
} }

49
webrtc/Nack.cpp
View File

@ -20,12 +20,13 @@ namespace mediakit {
// RTC配置项目 // RTC配置项目
namespace Rtc { namespace Rtc {
#define RTC_FIELD "rtc." #define RTC_FIELD "rtc."
//~ nack接收端 //~ nack接收端, rtp发送端
// Nack缓存包最早时间间隔 // rtp重发缓存列队最大长度，单位毫秒
const string kMaxNackMS = RTC_FIELD "maxNackMS"; const string kMaxRtpCacheMS = RTC_FIELD "maxRtpCacheMS";
// Nack包检查间隔(包数量) // rtp重发缓存列队最大长度，单位个数
const string kRtpCacheCheckInterval = RTC_FIELD "rtpCacheCheckInterval"; const string kMaxRtpCacheSize = RTC_FIELD "maxRtpCacheSize";
//~ nack发送端
//~ nack发送端，rtp接收端
//最大保留的rtp丢包状态个数 //最大保留的rtp丢包状态个数
const string kNackMaxSize = RTC_FIELD "nackMaxSize"; const string kNackMaxSize = RTC_FIELD "nackMaxSize";
// rtp丢包状态最长保留时间 // rtp丢包状态最长保留时间
@ -38,8 +39,8 @@ const string kNackIntervalRatio = RTC_FIELD "nackIntervalRatio";
const string kNackRtpSize = RTC_FIELD "nackRtpSize"; const string kNackRtpSize = RTC_FIELD "nackRtpSize";
static onceToken token([]() { static onceToken token([]() {
mINI::Instance()[kMaxNackMS] = 5 * 1000; mINI::Instance()[kMaxRtpCacheMS] = 5 * 1000;
mINI::Instance()[kRtpCacheCheckInterval] = 100; mINI::Instance()[kMaxRtpCacheSize] = 2048;
mINI::Instance()[kNackMaxSize] = 2048; mINI::Instance()[kNackMaxSize] = 2048;
mINI::Instance()[kNackMaxMS] = 3 * 1000; mINI::Instance()[kNackMaxMS] = 3 * 1000;
mINI::Instance()[kNackMaxCount] = 15; mINI::Instance()[kNackMaxCount] = 15;
@ -50,17 +51,26 @@ static onceToken token([]() {
} // namespace Rtc } // namespace Rtc
void NackList::pushBack(RtpPacket::Ptr rtp) { void NackList::pushBack(RtpPacket::Ptr rtp) {
GET_CONFIG(uint32_t, max_rtp_cache_ms, Rtc::kMaxRtpCacheMS);
GET_CONFIG(uint32_t, max_rtp_cache_size, Rtc::kMaxRtpCacheSize);
// 记录rtp
auto seq = rtp->getSeq(); auto seq = rtp->getSeq();
_nack_cache_seq.emplace_back(seq); _nack_cache_seq.emplace_back(seq);
_nack_cache_pkt.emplace(seq, std::move(rtp)); _nack_cache_pkt.emplace(seq, std::move(rtp));
GET_CONFIG(uint32_t, rtpcache_checkinterval, Rtc::kRtpCacheCheckInterval);
if (++_cache_ms_check < rtpcache_checkinterval) { // 限制rtp缓存最大个数
if (_nack_cache_seq.size() > max_rtp_cache_size) {
popFront();
}
if (++_cache_ms_check < 100) {
// 每100个rtp包检测下缓存长度，节省cpu资源
return; return;
} }
_cache_ms_check = 0; _cache_ms_check = 0;
GET_CONFIG(uint32_t, maxnackms, Rtc::kMaxNackMS); // 限制rtp缓存最大时长
while (getCacheMS() >= maxnackms) { while (getCacheMS() >= max_rtp_cache_ms) {
// 需要清除部分nack缓存
popFront(); popFront();
} }
} }
@ -97,13 +107,13 @@ RtpPacket::Ptr *NackList::getRtp(uint16_t seq) {
uint32_t NackList::getCacheMS() { uint32_t NackList::getCacheMS() {
while (_nack_cache_seq.size() > 2) { while (_nack_cache_seq.size() > 2) {
auto back_stamp = getRtpStamp(_nack_cache_seq.back()); auto back_stamp = getNtpStamp(_nack_cache_seq.back());
if (back_stamp == -1) { if (back_stamp == -1) {
_nack_cache_seq.pop_back(); _nack_cache_seq.pop_back();
continue; continue;
} }
auto front_stamp = getRtpStamp(_nack_cache_seq.front()); auto front_stamp = getNtpStamp(_nack_cache_seq.front());
if (front_stamp == -1) { if (front_stamp == -1) {
_nack_cache_seq.pop_front(); _nack_cache_seq.pop_front();
continue; continue;
@ -112,18 +122,19 @@ uint32_t NackList::getCacheMS() {
if (back_stamp >= front_stamp) { if (back_stamp >= front_stamp) {
return back_stamp - front_stamp; return back_stamp - front_stamp;
} }
// 很有可能回环了 // ntp时间戳回退了，非法数据，丢掉
return back_stamp + (UINT32_MAX - front_stamp); _nack_cache_seq.pop_front();
} }
return 0; return 0;
} }
int64_t NackList::getRtpStamp(uint16_t seq) { int64_t NackList::getNtpStamp(uint16_t seq) {
auto it = _nack_cache_pkt.find(seq); auto it = _nack_cache_pkt.find(seq);
if (it == _nack_cache_pkt.end()) { if (it == _nack_cache_pkt.end()) {
return -1; return -1;
} }
return it->second->getStampMS(false); // 使用ntp时间戳，不会回退
return it->second->getStampMS(true);
} }
//////////////////////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////////////////////

19
webrtc/Nack.h
View File

@ -22,24 +22,11 @@ namespace mediakit {
// RTC配置项目 // RTC配置项目
namespace Rtc { namespace Rtc {
//~ nack发送端，rtp接收端
//~ nack接收端(rtp发送端)
// Nack缓存包最早时间间隔
extern const std::string kMaxNackMS;
// Nack包检查间隔(包数量)
extern const std::string kRtpCacheCheckInterval;
//~ nack发送端(rtp接收端)
// 最大保留的rtp丢包状态个数 // 最大保留的rtp丢包状态个数
extern const std::string kNackMaxSize; extern const std::string kNackMaxSize;
// rtp丢包状态最长保留时间 // rtp丢包状态最长保留时间
extern const std::string kNackMaxMS; extern const std::string kNackMaxMS;
// nack最多请求重传次数
extern const std::string kNackMaxCount;
// nack重传频率，rtt的倍数
extern const std::string kNackIntervalRatio;
// nack包中rtp个数，减小此值可以让nack包响应更灵敏
extern const std::string kNackRtpSize;
} // namespace Rtc } // namespace Rtc
class NackList { class NackList {
@ -50,7 +37,7 @@ public:
private: private:
void popFront(); void popFront();
uint32_t getCacheMS(); uint32_t getCacheMS();
int64_t getRtpStamp(uint16_t seq); int64_t getNtpStamp(uint16_t seq);
RtpPacket::Ptr *getRtp(uint16_t seq); RtpPacket::Ptr *getRtp(uint16_t seq);
private: private:
@ -88,7 +75,7 @@ private:
struct NackStatus { struct NackStatus {
uint64_t first_stamp; uint64_t first_stamp;
uint64_t update_stamp; uint64_t update_stamp;
int nack_count = 0; uint32_t nack_count = 0;
}; };
std::map<uint16_t /*seq*/, NackStatus> _nack_send_status; std::map<uint16_t /*seq*/, NackStatus> _nack_send_status;
}; };