THENG
/
ZLMediaKit
mirror of https://gitcode.com/gh_mirrors/zlme/ZLMediaKit.git
6
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/ixingqiao/ZLMediaKit

This commit is contained in:
xingqiao 2024-03-07 11:59:02 +08:00
parent faf0febc54 79b2aa6adc
commit 3df1841232
57 changed files with 872 additions and 310 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
3rdpart/ZLToolKit

@ -1 +1 @@
Subproject commit e5535a7164f55eb9062213f40ddc68c0294e6f57 Subproject commit 04d1c47d2568f5ce1ff84260cefaf2754e514a5e

2
3rdpart/media-server

@ -1 +1 @@
Subproject commit a8a80e0738b052aa5671ef82a295ef388bd28e13 Subproject commit 8bc32a516b279414f749d0dead8bdc2837d3c527

6
AUTHORS
View File

@ -98,3 +98,9 @@ WuPeng <wp@zafu.edu.cn>
[zjx94](https://github.com/zjx94) [zjx94](https://github.com/zjx94)
[LeiZhi.Mai ](https://github.com/blueskiner) [LeiZhi.Mai ](https://github.com/blueskiner)
[JiaHao](https://github.com/nashiracn) [JiaHao](https://github.com/nashiracn)
[chdahuzi](https://github.com/chdahuzi)
[snysmtx](https://github.com/snysmtx)
[SetoKaiba](https://github.com/SetoKaiba)
[sandro-qiang](https://github.com/sandro-qiang)
[Paul Philippov](https://github.com/themactep)
[张传峰](https://github.com/zhang-chuanfeng)

6
README.md
View File

@ -352,6 +352,12 @@ bash build_docker_images.sh
[zjx94](https://github.com/zjx94) [zjx94](https://github.com/zjx94)
[LeiZhi.Mai ](https://github.com/blueskiner) [LeiZhi.Mai ](https://github.com/blueskiner)
[JiaHao](https://github.com/nashiracn) [JiaHao](https://github.com/nashiracn)
[chdahuzi](https://github.com/chdahuzi)
[snysmtx](https://github.com/snysmtx)
[SetoKaiba](https://github.com/SetoKaiba)
[sandro-qiang](https://github.com/sandro-qiang)
[Paul Philippov](https://github.com/themactep)
[张传峰](https://github.com/zhang-chuanfeng)
同时感谢JetBrains对开源项目的支持，本项目使用CLion开发与调试： 同时感谢JetBrains对开源项目的支持，本项目使用CLion开发与调试：

12
README_en.md
View File

@ -279,12 +279,12 @@ git submodule update --init
return; return;
} }
auto viedoTrack = strongPlayer->getTrack(TrackVideo); auto videoTrack = strongPlayer->getTrack(TrackVideo);
if (!viedoTrack) { if (!videoTrack) {
WarnL << "No video Track!"; WarnL << "No video Track!";
return; return;
} }
viedoTrack->addDelegate([](const Frame::Ptr &frame) { videoTrack->addDelegate([](const Frame::Ptr &frame) {
//please decode video here //please decode video here
}); });
}); });
@ -510,6 +510,12 @@ Thanks to all those who have supported this project in various ways, including b
[zjx94](https://github.com/zjx94) [zjx94](https://github.com/zjx94)
[LeiZhi.Mai ](https://github.com/blueskiner) [LeiZhi.Mai ](https://github.com/blueskiner)
[JiaHao](https://github.com/nashiracn) [JiaHao](https://github.com/nashiracn)
[chdahuzi](https://github.com/chdahuzi)
[snysmtx](https://github.com/snysmtx)
[SetoKaiba](https://github.com/SetoKaiba)
[sandro-qiang](https://github.com/sandro-qiang)
[Paul Philippov](https://github.com/themactep)
[张传峰](https://github.com/zhang-chuanfeng)
Also thank to JetBrains for their support for open source project, we developed and debugged zlmediakit with CLion: Also thank to JetBrains for their support for open source project, we developed and debugged zlmediakit with CLion:

27
api/include/mk_events.h
View File

@ -177,6 +177,33 @@ typedef struct {
*/ */
void(API_CALL *on_mk_media_send_rtp_stop)(const char *vhost, const char *app, const char *stream, const char *ssrc, int err, const char *msg); void(API_CALL *on_mk_media_send_rtp_stop)(const char *vhost, const char *app, const char *stream, const char *ssrc, int err, const char *msg);
/**
* rtc sctp连接中/完成/失败/关闭回调
* @param rtc_transport 数据通道对象
*/
void(API_CALL *on_mk_rtc_sctp_connecting)(mk_rtc_transport rtc_transport);
void(API_CALL *on_mk_rtc_sctp_connected)(mk_rtc_transport rtc_transport);
void(API_CALL *on_mk_rtc_sctp_failed)(mk_rtc_transport rtc_transport);
void(API_CALL *on_mk_rtc_sctp_closed)(mk_rtc_transport rtc_transport);
/**
* rtc数据通道发送数据回调
* @param rtc_transport 数据通道对象
* @param msg 数据
* @param len 数据长度
*/
void(API_CALL *on_mk_rtc_sctp_send)(mk_rtc_transport rtc_transport, const uint8_t *msg, size_t len);
/**
* rtc数据通道接收数据回调
* @param rtc_transport 数据通道对象
* @param streamId 流id
* @param ppid 协议id
* @param msg 数据
* @param len 数据长度
*/
void(API_CALL *on_mk_rtc_sctp_received)(mk_rtc_transport rtc_transport, uint16_t streamId, uint32_t ppid, const uint8_t *msg, size_t len);
} mk_events; } mk_events;

14
api/include/mk_events_objects.h
View File

@ -352,6 +352,20 @@ API_EXPORT mk_auth_invoker API_CALL mk_auth_invoker_clone(const mk_auth_invoker
*/ */
API_EXPORT void API_CALL mk_auth_invoker_clone_release(const mk_auth_invoker ctx); API_EXPORT void API_CALL mk_auth_invoker_clone_release(const mk_auth_invoker ctx);
///////////////////////////////////////////WebRtcTransport/////////////////////////////////////////////
//WebRtcTransport对象的C映射
typedef struct mk_rtc_transport_t *mk_rtc_transport;
/**
* 发送rtc数据通道
* @param ctx 数据通道对象
* @param streamId 流id
* @param ppid 协议id
* @param msg 数据
* @param len 数据长度
*/
API_EXPORT void API_CALL mk_rtc_send_datachannel(const mk_rtc_transport ctx, uint16_t streamId, uint32_t ppid, const char* msg, size_t len);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

12
api/include/mk_proxyplayer.h
View File

@ -12,6 +12,7 @@
#define MK_PROXY_PLAYER_H_ #define MK_PROXY_PLAYER_H_
#include "mk_common.h" #include "mk_common.h"
#include "mk_util.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
@ -31,6 +32,17 @@ typedef struct mk_proxy_player_t *mk_proxy_player;
*/ */
API_EXPORT mk_proxy_player API_CALL mk_proxy_player_create(const char *vhost, const char *app, const char *stream, int hls_enabled, int mp4_enabled); API_EXPORT mk_proxy_player API_CALL mk_proxy_player_create(const char *vhost, const char *app, const char *stream, int hls_enabled, int mp4_enabled);
/**
* 创建一个代理播放器
* @param vhost 虚拟主机名，一般为__defaultVhost__
* @param app 应用名
* @param stream 流名
* @param option ProtocolOption相关配置
* @return 对象指针
*/
API_EXPORT mk_proxy_player API_CALL mk_proxy_player_create2(const char *vhost, const char *app, const char *stream, mk_ini option);
/** /**
* 销毁代理播放器 * 销毁代理播放器
* @param ctx 对象指针 * @param ctx 对象指针

37
api/source/mk_events.cpp
View File

@ -14,6 +14,7 @@
#include "Http/HttpSession.h" #include "Http/HttpSession.h"
#include "Rtsp/RtspSession.h" #include "Rtsp/RtspSession.h"
#include "Record/MP4Recorder.h" #include "Record/MP4Recorder.h"
#include "webrtc/WebRtcTransport.h"
using namespace toolkit; using namespace toolkit;
using namespace mediakit; using namespace mediakit;
@ -167,6 +168,42 @@ API_EXPORT void API_CALL mk_events_listen(const mk_events *events){
sender.getMediaTuple().stream.c_str(), ssrc.c_str(), ex.getErrCode(), ex.what()); sender.getMediaTuple().stream.c_str(), ssrc.c_str(), ex.getErrCode(), ex.what());
} }
}); });
NoticeCenter::Instance().addListener(&s_tag, Broadcast::kBroadcastRtcSctpConnecting,[](BroadcastRtcSctpConnectArgs){
if (s_events.on_mk_rtc_sctp_connecting) {
s_events.on_mk_rtc_sctp_connecting((mk_rtc_transport)&sender);
}
});
NoticeCenter::Instance().addListener(&s_tag, Broadcast::kBroadcastRtcSctpConnected,[](BroadcastRtcSctpConnectArgs){
if (s_events.on_mk_rtc_sctp_connected) {
s_events.on_mk_rtc_sctp_connected((mk_rtc_transport)&sender);
}
});
NoticeCenter::Instance().addListener(&s_tag, Broadcast::kBroadcastRtcSctpFailed,[](BroadcastRtcSctpConnectArgs){
if (s_events.on_mk_rtc_sctp_failed) {
s_events.on_mk_rtc_sctp_failed((mk_rtc_transport)&sender);
}
});
NoticeCenter::Instance().addListener(&s_tag, Broadcast::kBroadcastRtcSctpClosed,[](BroadcastRtcSctpConnectArgs){
if (s_events.on_mk_rtc_sctp_closed) {
s_events.on_mk_rtc_sctp_closed((mk_rtc_transport)&sender);
}
});
NoticeCenter::Instance().addListener(&s_tag, Broadcast::kBroadcastRtcSctpSend,[](BroadcastRtcSctpSendArgs){
if (s_events.on_mk_rtc_sctp_send) {
s_events.on_mk_rtc_sctp_send((mk_rtc_transport)&sender, data, len);
}
});
NoticeCenter::Instance().addListener(&s_tag, Broadcast::kBroadcastRtcSctpReceived,[](BroadcastRtcSctpReceivedArgs){
if (s_events.on_mk_rtc_sctp_received) {
s_events.on_mk_rtc_sctp_received((mk_rtc_transport)&sender, streamId, ppid, msg, len);
}
});
}); });
} }

19
api/source/mk_events_objects.cpp
View File

@ -17,6 +17,7 @@
#include "Http/HttpClient.h" #include "Http/HttpClient.h"
#include "Rtsp/RtspSession.h" #include "Rtsp/RtspSession.h"
#include "webrtc/WebRtcTransport.h"
using namespace toolkit; using namespace toolkit;
using namespace mediakit; using namespace mediakit;
@ -498,3 +499,21 @@ API_EXPORT void API_CALL mk_auth_invoker_clone_release(const mk_auth_invoker ctx
Broadcast::AuthInvoker *invoker = (Broadcast::AuthInvoker *)ctx; Broadcast::AuthInvoker *invoker = (Broadcast::AuthInvoker *)ctx;
delete invoker; delete invoker;
} }
///////////////////////////////////////////WebRtcTransport/////////////////////////////////////////////
API_EXPORT void API_CALL mk_rtc_sendDatachannel(const mk_rtc_transport ctx, uint16_t streamId, uint32_t ppid, const char *msg, size_t len) {
#ifdef ENABLE_WEBRTC
assert(ctx && msg);
WebRtcTransport *transport = (WebRtcTransport *)ctx;
std::string msg_str(msg, len);
std::weak_ptr<WebRtcTransport> weak_trans = transport->shared_from_this();
transport->getPoller()->async([streamId, ppid, msg_str, weak_trans]() {
// 切换线程后再操作
if (auto trans = weak_trans.lock()) {
trans->sendDatachannel(streamId, ppid, msg_str.c_str(), msg_str.size());
}
});
#else
WarnL << "未启用webrtc功能, 编译时请开启ENABLE_WEBRTC";
#endif
}

9
api/source/mk_proxyplayer.cpp
View File

@ -10,6 +10,7 @@
#include "mk_proxyplayer.h" #include "mk_proxyplayer.h"
#include "Player/PlayerProxy.h" #include "Player/PlayerProxy.h"
#include "mk_util.h"
using namespace toolkit; using namespace toolkit;
using namespace mediakit; using namespace mediakit;
@ -23,6 +24,14 @@ API_EXPORT mk_proxy_player API_CALL mk_proxy_player_create(const char *vhost, co
return (mk_proxy_player) obj; return (mk_proxy_player) obj;
} }
API_EXPORT mk_proxy_player API_CALL mk_proxy_player_create2(const char *vhost, const char *app, const char *stream, mk_ini ini) {
assert(vhost && app && stream);
ProtocolOption option(*((mINI *)ini));
PlayerProxy::Ptr *obj(new PlayerProxy::Ptr(new PlayerProxy(vhost, app, stream, option)));
return (mk_proxy_player)obj;
}
API_EXPORT void API_CALL mk_proxy_player_release(mk_proxy_player ctx) { API_EXPORT void API_CALL mk_proxy_player_release(mk_proxy_player ctx) {
assert(ctx); assert(ctx);
PlayerProxy::Ptr *obj = (PlayerProxy::Ptr *) ctx; PlayerProxy::Ptr *obj = (PlayerProxy::Ptr *) ctx;

9
conf/config.ini
View File

@ -24,7 +24,7 @@ bin=/usr/bin/ffmpeg
#FFmpeg拉流再推流的命令模板，通过该模板可以设置再编码的一些参数 #FFmpeg拉流再推流的命令模板，通过该模板可以设置再编码的一些参数
cmd=%s -re -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s cmd=%s -re -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s
#FFmpeg生成截图的命令，可以通过修改该配置改变截图分辨率或质量 #FFmpeg生成截图的命令，可以通过修改该配置改变截图分辨率或质量
snap=%s -i %s -y -f mjpeg -frames:v 1 %s snap=%s -i %s -y -f mjpeg -frames:v 1 -an %s
#FFmpeg日志的路径，如果置空则不生成FFmpeg日志 #FFmpeg日志的路径，如果置空则不生成FFmpeg日志
#可以为相对(相对于本可执行程序目录)或绝对路径 #可以为相对(相对于本可执行程序目录)或绝对路径
log=./ffmpeg/ffmpeg.log log=./ffmpeg/ffmpeg.log
@ -329,6 +329,13 @@ opus_pt=100
#如果不调用startSendRtp相关接口，可以置0节省内存 #如果不调用startSendRtp相关接口，可以置0节省内存
gop_cache=1 gop_cache=1
#国标发送g711 rtp 打包时，每个包的语音时长是多少，默认是100 ms，范围为20~180ms (gb28181-2016，c.2.4规定)，
#最好为20 的倍数，程序自动向20的倍数取整
rtp_g711_dur_ms = 100
#udp接收数据socket buffer大小配置
#4*1024*1024=4196304
udp_recv_socket_buffer=4194304
[rtc] [rtc]
#rtc播放推流、播放超时时间 #rtc播放推流、播放超时时间
timeoutSec=15 timeoutSec=15

28
ext-codec/G711Rtp.cpp
View File

@ -8,13 +8,27 @@ G711RtpEncoder::G711RtpEncoder(CodecId codec, uint32_t channels){
_channels = channels; _channels = channels;
} }
void G711RtpEncoder::setOpt(int opt, const toolkit::Any &param) {
if (opt == RTP_ENCODER_PKT_DUR_MS) {
if (param.is<uint32_t>()) {
auto dur = param.get<uint32_t>();
if (dur < 20 || dur > 180) {
WarnL << "set g711 rtp encoder duration ms failed for " << dur;
return;
}
// 向上 20ms 取整
_pkt_dur_ms = (dur + 19) / 20 * 20;
}
}
}
bool G711RtpEncoder::inputFrame(const Frame::Ptr &frame) { bool G711RtpEncoder::inputFrame(const Frame::Ptr &frame) {
auto dur = (_cache_frame->size() - _cache_frame->prefixSize()) / (8 * _channels); auto dur = (_cache_frame->size() - _cache_frame->prefixSize()) / (8 * _channels);
auto next_pts = _cache_frame->pts() + dur; auto next_pts = _cache_frame->pts() + dur;
if (next_pts == 0) { if (next_pts == 0) {
_cache_frame->_pts = frame->pts(); _cache_frame->_pts = frame->pts();
} else { } else {
if ((next_pts + 20) < frame->pts()) { // 有丢包超过20ms if ((next_pts + _pkt_dur_ms) < frame->pts()) { // 有丢包超过20ms
_cache_frame->_pts = frame->pts() - dur; _cache_frame->_pts = frame->pts() - dur;
} }
} }
@ -24,9 +38,9 @@ bool G711RtpEncoder::inputFrame(const Frame::Ptr &frame) {
auto ptr = _cache_frame->data() + _cache_frame->prefixSize(); auto ptr = _cache_frame->data() + _cache_frame->prefixSize();
auto len = _cache_frame->size() - _cache_frame->prefixSize(); auto len = _cache_frame->size() - _cache_frame->prefixSize();
auto remain_size = len; auto remain_size = len;
auto max_size = 160 * _channels; // 20 ms per rtp auto max_size = 160 * _channels * _pkt_dur_ms / 20; // 20 ms per 160 byte
int n = 0; uint32_t n = 0;
bool mark = false; bool mark = true;
while (remain_size >= max_size) { while (remain_size >= max_size) {
size_t rtp_size; size_t rtp_size;
if (remain_size >= max_size) { if (remain_size >= max_size) {
@ -35,13 +49,13 @@ bool G711RtpEncoder::inputFrame(const Frame::Ptr &frame) {
break; break;
} }
n++; n++;
stamp += 20; stamp += _pkt_dur_ms;
RtpCodec::inputRtp(getRtpInfo().makeRtp(TrackAudio, ptr, rtp_size, mark, stamp), false); RtpCodec::inputRtp(getRtpInfo().makeRtp(TrackAudio, ptr, rtp_size, mark, stamp), true);
ptr += rtp_size; ptr += rtp_size;
remain_size -= rtp_size; remain_size -= rtp_size;
} }
_cache_frame->_buffer.erase(0, n * max_size); _cache_frame->_buffer.erase(0, n * max_size);
_cache_frame->_pts += 20 * n; _cache_frame->_pts += (uint64_t)_pkt_dur_ms * n;
return len > 0; return len > 0;
} }

3
ext-codec/G711Rtp.h
View File

@ -36,8 +36,11 @@ public:
*/ */
bool inputFrame(const Frame::Ptr &frame) override; bool inputFrame(const Frame::Ptr &frame) override;
void setOpt(int opt, const toolkit::Any &param) override;
private: private:
uint32_t _channels = 1; uint32_t _channels = 1;
uint32_t _pkt_dur_ms = 20;
FrameImp::Ptr _cache_frame; FrameImp::Ptr _cache_frame;
}; };

12
ext-codec/H264Rtmp.cpp
View File

@ -14,16 +14,24 @@
using namespace std; using namespace std;
using namespace toolkit; using namespace toolkit;
#define CHECK_RET(...) \
try { \
CHECK(__VA_ARGS__); \
} catch (AssertFailedException & ex) { \
WarnL << ex.what(); \
return; \
}
namespace mediakit { namespace mediakit {
void H264RtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt) { void H264RtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt) {
if (pkt->isConfigFrame()) { if (pkt->isConfigFrame()) {
CHECK(pkt->size() > 5); CHECK_RET(pkt->size() > 5);
getTrack()->setExtraData((uint8_t *)pkt->data() + 5, pkt->size() - 5); getTrack()->setExtraData((uint8_t *)pkt->data() + 5, pkt->size() - 5);
return; return;
} }
CHECK(pkt->size() > 9); CHECK_RET(pkt->size() > 9);
uint8_t *cts_ptr = (uint8_t *)(pkt->buffer.data() + 2); uint8_t *cts_ptr = (uint8_t *)(pkt->buffer.data() + 2);
int32_t cts = (((cts_ptr[0] << 16) | (cts_ptr[1] << 8) | (cts_ptr[2])) + 0xff800000) ^ 0xff800000; int32_t cts = (((cts_ptr[0] << 16) | (cts_ptr[1] << 8) | (cts_ptr[2])) + 0xff800000) ^ 0xff800000;
auto pts = pkt->time_stamp + cts; auto pts = pkt->time_stamp + cts;

16
ext-codec/H265Rtmp.cpp
View File

@ -18,6 +18,14 @@
using namespace std; using namespace std;
using namespace toolkit; using namespace toolkit;
#define CHECK_RET(...) \
try { \
CHECK(__VA_ARGS__); \
} catch (AssertFailedException & ex) { \
WarnL << ex.what(); \
return; \
}
namespace mediakit { namespace mediakit {
void H265RtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt) { void H265RtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt) {
@ -44,7 +52,7 @@ void H265RtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt) {
auto data = (uint8_t *)pkt->data() + RtmpPacketInfo::kEnhancedRtmpHeaderSize; auto data = (uint8_t *)pkt->data() + RtmpPacketInfo::kEnhancedRtmpHeaderSize;
auto size = pkt->size() - RtmpPacketInfo::kEnhancedRtmpHeaderSize; auto size = pkt->size() - RtmpPacketInfo::kEnhancedRtmpHeaderSize;
auto pts = pkt->time_stamp; auto pts = pkt->time_stamp;
CHECK(size > 3); CHECK_RET(size > 3);
if (RtmpPacketType::PacketTypeCodedFrames == _info.video.pkt_type) { if (RtmpPacketType::PacketTypeCodedFrames == _info.video.pkt_type) {
// SI24 = [CompositionTime Offset] // SI24 = [CompositionTime Offset]
int32_t cts = (((data[0] << 16) | (data[1] << 8) | (data[2])) + 0xff800000) ^ 0xff800000; int32_t cts = (((data[0] << 16) | (data[1] << 8) | (data[2])) + 0xff800000) ^ 0xff800000;
@ -52,7 +60,7 @@ void H265RtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt) {
data += 3; data += 3;
size -= 3; size -= 3;
} }
CHECK(size > 4); CHECK_RET(size > 4);
splitFrame(data, size, pkt->time_stamp, pts); splitFrame(data, size, pkt->time_stamp, pts);
break; break;
} }
@ -63,12 +71,12 @@ void H265RtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt) {
// 国内扩展(12) H265 rtmp // 国内扩展(12) H265 rtmp
if (pkt->isConfigFrame()) { if (pkt->isConfigFrame()) {
CHECK(pkt->size() > 5); CHECK_RET(pkt->size() > 5);
getTrack()->setExtraData((uint8_t *)pkt->data() + 5, pkt->size() - 5); getTrack()->setExtraData((uint8_t *)pkt->data() + 5, pkt->size() - 5);
return; return;
} }
CHECK(pkt->size() > 9); CHECK_RET(pkt->size() > 9);
uint8_t *cts_ptr = (uint8_t *)(pkt->buffer.data() + 2); uint8_t *cts_ptr = (uint8_t *)(pkt->buffer.data() + 2);
int32_t cts = (((cts_ptr[0] << 16) | (cts_ptr[1] << 8) | (cts_ptr[2])) + 0xff800000) ^ 0xff800000; int32_t cts = (((cts_ptr[0] << 16) | (cts_ptr[1] << 8) | (cts_ptr[2])) + 0xff800000) ^ 0xff800000;
auto pts = pkt->time_stamp + cts; auto pts = pkt->time_stamp + cts;

14
ext-codec/JPEGRtp.cpp
View File

@ -605,6 +605,7 @@ void JPEGRtpEncoder::rtpSendJpeg(const uint8_t *buf, int size, uint64_t pts, uin
int i; int i;
int default_huffman_tables = 0; int default_huffman_tables = 0;
uint8_t *out = nullptr; uint8_t *out = nullptr;
uint16_t restart_interval = 0;
/* preparse the header for getting some info */ /* preparse the header for getting some info */
for (i = 0; i < size; i++) { for (i = 0; i < size; i++) {
@ -714,6 +715,9 @@ void JPEGRtpEncoder::rtpSendJpeg(const uint8_t *buf, int size, uint64_t pts, uin
return; return;
} }
break; break;
} else if (buf[i + 1] == DRI) {
type |= 0x40;
restart_interval = AV_RB16(&buf[i + 4]);
} }
} }
if (default_huffman_tables && default_huffman_tables != 31) { if (default_huffman_tables && default_huffman_tables != 31) {
@ -744,6 +748,9 @@ void JPEGRtpEncoder::rtpSendJpeg(const uint8_t *buf, int size, uint64_t pts, uin
if (off == 0 && nb_qtables) if (off == 0 && nb_qtables)
hdr_size += 4 + 64 * nb_qtables; hdr_size += 4 + 64 * nb_qtables;
if (type & 0x40)
hdr_size += 4;
/* payload max in one packet */ /* payload max in one packet */
len = MIN(size, (int)getRtpInfo().getMaxSize() - hdr_size); len = MIN(size, (int)getRtpInfo().getMaxSize() - hdr_size);
@ -759,6 +766,13 @@ void JPEGRtpEncoder::rtpSendJpeg(const uint8_t *buf, int size, uint64_t pts, uin
bytestream_put_byte(&p, w); bytestream_put_byte(&p, w);
bytestream_put_byte(&p, h); bytestream_put_byte(&p, h);
/* set dri */
if (type & 0x40) {
bytestream_put_be16(&p, restart_interval);
bytestream_put_byte(&p, 0xff);
bytestream_put_byte(&p, 0xff);
}
if (off == 0 && nb_qtables) { if (off == 0 && nb_qtables) {
/* set quantization tables header */ /* set quantization tables header */
bytestream_put_byte(&p, 0); bytestream_put_byte(&p, 0);

4
player/CMakeLists.txt
View File

@ -31,7 +31,9 @@ if(PKG_CONFIG_FOUND)
list(APPEND LINK_LIBRARIES PkgConfig::SDL2) list(APPEND LINK_LIBRARIES PkgConfig::SDL2)
message(STATUS "found library: ${SDL2_LIBRARIES}") message(STATUS "found library: ${SDL2_LIBRARIES}")
endif() endif()
else() endif()
if(NOT SDL2_FOUND)
find_package(SDL2 QUIET) find_package(SDL2 QUIET)
if(SDL2_FOUND) if(SDL2_FOUND)
include_directories(SYSTEM ${SDL2_INCLUDE_DIR}) include_directories(SYSTEM ${SDL2_INCLUDE_DIR})

131
player/test_player.cpp
View File

@ -10,6 +10,7 @@
#include <signal.h> #include <signal.h>
#include "Util/logger.h" #include "Util/logger.h"
#include "Util/util.h"
#include <iostream> #include <iostream>
#include "Common/config.h" #include "Common/config.h"
#include "Rtsp/UDPServer.h" #include "Rtsp/UDPServer.h"
@ -48,80 +49,76 @@ int WINAPI WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstanc, LPSTR lpCmdLine,
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
#endif #endif
static char *url = argv[1]; static char *url = argv[1];
//设置退出信号处理函数 {
signal(SIGINT, [](int) { SDLDisplayerHelper::Instance().shutdown(); }); // 设置退出信号处理函数
//设置日志 signal(SIGINT, [](int) { SDLDisplayerHelper::Instance().shutdown(); });
Logger::Instance().add(std::make_shared<ConsoleChannel>()); // 设置日志
Logger::Instance().setWriter(std::make_shared<AsyncLogWriter>()); Logger::Instance().add(std::make_shared<ConsoleChannel>());
Logger::Instance().setWriter(std::make_shared<AsyncLogWriter>());
if (argc < 3) { if (argc < 3) {
ErrorL << "\r\n测试方法：./test_player rtxp_url rtp_type\r\n" ErrorL << "\r\n测试方法：./test_player rtxp_url rtp_type\r\n"
<< "例如：./test_player rtsp://admin:123456@127.0.0.1/live/0 0\r\n"; << "例如：./test_player rtsp://admin:123456@127.0.0.1/live/0 0\r\n";
return 0; return 0;
}
auto player = std::make_shared<MediaPlayer>();
//sdl要求在main线程初始化
auto displayer = std::make_shared<YuvDisplayer>(nullptr, url);
weak_ptr<MediaPlayer> weakPlayer = player;
player->setOnPlayResult([weakPlayer, displayer](const SockException &ex) {
InfoL << "OnPlayResult:" << ex.what();
auto strongPlayer = weakPlayer.lock();
if (ex || !strongPlayer) {
return;
} }
auto videoTrack = dynamic_pointer_cast<VideoTrack>(strongPlayer->getTrack(TrackVideo, false)); auto player = std::make_shared<MediaPlayer>();
auto audioTrack = dynamic_pointer_cast<AudioTrack>(strongPlayer->getTrack(TrackAudio,false)); // sdl要求在main线程初始化
auto displayer = std::make_shared<YuvDisplayer>(nullptr, url);
weak_ptr<MediaPlayer> weakPlayer = player;
player->setOnPlayResult([weakPlayer, displayer](const SockException &ex) {
InfoL << "OnPlayResult:" << ex.what();
auto strongPlayer = weakPlayer.lock();
if (ex || !strongPlayer) {
return;
}
if (videoTrack) { auto videoTrack = dynamic_pointer_cast<VideoTrack>(strongPlayer->getTrack(TrackVideo, false));
auto decoder = std::make_shared<FFmpegDecoder>(videoTrack); auto audioTrack = dynamic_pointer_cast<AudioTrack>(strongPlayer->getTrack(TrackAudio, false));
decoder->setOnDecode([displayer](const FFmpegFrame::Ptr &yuv) {
SDLDisplayerHelper::Instance().doTask([yuv, displayer]() { if (videoTrack) {
//sdl要求在main线程渲染 auto decoder = std::make_shared<FFmpegDecoder>(videoTrack);
displayer->displayYUV(yuv->get()); decoder->setOnDecode([displayer](const FFmpegFrame::Ptr &yuv) {
return true; SDLDisplayerHelper::Instance().doTask([yuv, displayer]() {
// sdl要求在main线程渲染
displayer->displayYUV(yuv->get());
return true;
});
}); });
}); videoTrack->addDelegate([decoder](const Frame::Ptr &frame) { return decoder->inputFrame(frame, false, true); });
videoTrack->addDelegate([decoder](const Frame::Ptr &frame) { }
return decoder->inputFrame(frame, false, true);
}); if (audioTrack) {
auto decoder = std::make_shared<FFmpegDecoder>(audioTrack);
auto audio_player = std::make_shared<AudioPlayer>();
// FFmpeg解码时已经统一转换为16位整型pcm
audio_player->setup(audioTrack->getAudioSampleRate(), audioTrack->getAudioChannel(), AUDIO_S16);
FFmpegSwr::Ptr swr;
decoder->setOnDecode([audio_player, swr](const FFmpegFrame::Ptr &frame) mutable {
if (!swr) {
swr = std::make_shared<FFmpegSwr>(AV_SAMPLE_FMT_S16, frame->get()->channels, frame->get()->channel_layout, frame->get()->sample_rate);
}
auto pcm = swr->inputFrame(frame);
auto len = pcm->get()->nb_samples * pcm->get()->channels * av_get_bytes_per_sample((enum AVSampleFormat)pcm->get()->format);
audio_player->playPCM((const char *)(pcm->get()->data[0]), MIN(len, frame->get()->linesize[0]));
});
audioTrack->addDelegate([decoder](const Frame::Ptr &frame) { return decoder->inputFrame(frame, false, true); });
}
});
player->setOnShutdown([](const SockException &ex) { WarnL << "play shutdown: " << ex.what(); });
(*player)[Client::kRtpType] = atoi(argv[2]);
// 不等待track ready再回调播放成功事件，这样可以加快秒开速度
(*player)[Client::kWaitTrackReady] = false;
if (argc > 3) {
(*player)[Client::kPlayTrack] = atoi(argv[3]);
} }
player->play(argv[1]);
if (audioTrack) { SDLDisplayerHelper::Instance().runLoop();
auto decoder = std::make_shared<FFmpegDecoder>(audioTrack);
auto audio_player = std::make_shared<AudioPlayer>();
//FFmpeg解码时已经统一转换为16位整型pcm
audio_player->setup(audioTrack->getAudioSampleRate(), audioTrack->getAudioChannel(), AUDIO_S16);
FFmpegSwr::Ptr swr;
decoder->setOnDecode([audio_player, swr](const FFmpegFrame::Ptr &frame) mutable{
if (!swr) {
swr = std::make_shared<FFmpegSwr>(AV_SAMPLE_FMT_S16, frame->get()->channels,
frame->get()->channel_layout, frame->get()->sample_rate);
}
auto pcm = swr->inputFrame(frame);
auto len = pcm->get()->nb_samples * pcm->get()->channels * av_get_bytes_per_sample((enum AVSampleFormat)pcm->get()->format);
audio_player->playPCM((const char *) (pcm->get()->data[0]), MIN(len, frame->get()->linesize[0]));
});
audioTrack->addDelegate([decoder](const Frame::Ptr &frame) {
return decoder->inputFrame(frame, false, true);
});
}
});
player->setOnShutdown([](const SockException &ex){
WarnL << "play shutdown: " << ex.what();
});
(*player)[Client::kRtpType] = atoi(argv[2]);
//不等待track ready再回调播放成功事件，这样可以加快秒开速度
(*player)[Client::kWaitTrackReady] = false;
if (argc > 3) {
(*player)[Client::kPlayTrack] = atoi(argv[3]);
} }
player->play(argv[1]); sleep(1);
SDLDisplayerHelper::Instance().runLoop();
return 0; return 0;
} }

31
postman/ZLMediaKit.postman_collection.json
View File

@ -1,10 +1,11 @@
{ {
"info": { "info": {
"_postman_id": "509e5f6b-728c-4d5f-b3e8-521d76b2cc7a", "_postman_id": "08c77fc3-7670-428c-bde4-80c8cc9f389f",
"name": "ZLMediaKit", "name": "ZLMediaKit",
"description": "媒体服务器", "description": "媒体服务器",
"schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json", "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json",
"_exporter_id": "29185956" "_exporter_id": "29185956",
"_collection_link": "https://lively-station-598157.postman.co/workspace/%E6%B5%81%E5%AA%92%E4%BD%93%E6%9C%8D%E5%8A%A1~1e119172-45b0-4ed6-b1fc-8a15d0e2d5f8/collection/29185956-08c77fc3-7670-428c-bde4-80c8cc9f389f?action=share&source=collection_link&creator=29185956"
}, },
"item": [ "item": [
{ {
@ -1216,7 +1217,7 @@
"method": "GET", "method": "GET",
"header": [], "header": [],
"url": { "url": {
"raw": "{{ZLMediaKit_URL}}/index/api/seekRecordStamp?secret={{ZLMediaKit_secret}}&vhost={{defaultVhost}}&app=live&stream=obs&stamp", "raw": "{{ZLMediaKit_URL}}/index/api/seekRecordStamp?secret={{ZLMediaKit_secret}}&vhost={{defaultVhost}}&app=live&stream=obs&stamp=1000",
"host": [ "host": [
"{{ZLMediaKit_URL}}" "{{ZLMediaKit_URL}}"
], ],
@ -1469,9 +1470,15 @@
"disabled": true "disabled": true
}, },
{ {
"key": "only_audio", "key": "only_track",
"value": "1", "value": "1",
"description": "是否为单音频track，用于语音对讲", "description": "是否为单音频/单视频track，0：不设置，1：单音频，2：单视频",
"disabled": true
},
{
"key": "local_ip",
"value": "::",
"description": "指定创建RTP的本地ip，ipv4可填”0.0.0.0“，ipv6可填”::“，一般保持默认",
"disabled": true "disabled": true
} }
] ]
@ -1485,14 +1492,14 @@
"method": "GET", "method": "GET",
"header": [], "header": [],
"url": { "url": {
"raw": "{{ZLMediaKit_URL}}/index/api/openRtpServer?secret={{ZLMediaKit_secret}}&port=0&tcp_mode=1&stream_id=test", "raw": "{{ZLMediaKit_URL}}/index/api/openRtpServerMultiplex?secret={{ZLMediaKit_secret}}&port=0&tcp_mode=1&stream_id=test",
"host": [ "host": [
"{{ZLMediaKit_URL}}" "{{ZLMediaKit_URL}}"
], ],
"path": [ "path": [
"index", "index",
"api", "api",
"openRtpServer" "openRtpServerMultiplex"
], ],
"query": [ "query": [
{ {
@ -1516,9 +1523,15 @@
"description": "该端口绑定的流id\n" "description": "该端口绑定的流id\n"
}, },
{ {
"key": "only_audio", "key": "only_track",
"value": "0", "value": "0",
"description": "是否为单音频track，用于语音对讲", "description": "是否为单音频/单视频track，0：不设置，1：单音频，2：单视频",
"disabled": true
},
{
"key": "local_ip",
"value": "::",
"description": "指定创建RTP的本地ip，ipv4可填”0.0.0.0“，ipv6可填”::“，一般保持默认",
"disabled": true "disabled": true
} }
] ]

2
server/FFmpegSource.cpp
View File

@ -40,7 +40,7 @@ onceToken token([]() {
//ffmpeg日志保存路径 //ffmpeg日志保存路径
mINI::Instance()[kLog] = "./ffmpeg/ffmpeg.log"; mINI::Instance()[kLog] = "./ffmpeg/ffmpeg.log";
mINI::Instance()[kCmd] = "%s -re -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s"; mINI::Instance()[kCmd] = "%s -re -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s";
mINI::Instance()[kSnap] = "%s -i %s -y -f mjpeg -frames:v 1 %s"; mINI::Instance()[kSnap] = "%s -i %s -y -f mjpeg -frames:v 1 -an %s";
mINI::Instance()[kRestartSec] = 0; mINI::Instance()[kRestartSec] = 0;
}); });
} }

65
server/WebApi.cpp
View File

@ -20,6 +20,7 @@
#include <functional> #include <functional>
#include <unordered_map> #include <unordered_map>
#include <regex>
#include "Util/MD5.h" #include "Util/MD5.h"
#include "Util/util.h" #include "Util/util.h"
#include "Util/File.h" #include "Util/File.h"
@ -203,7 +204,7 @@ static ApiArgsType getAllArgs(const Parser &parser) {
if (parser["Content-Type"].find("application/x-www-form-urlencoded") == 0) { if (parser["Content-Type"].find("application/x-www-form-urlencoded") == 0) {
auto contentArgs = parser.parseArgs(parser.content()); auto contentArgs = parser.parseArgs(parser.content());
for (auto &pr : contentArgs) { for (auto &pr : contentArgs) {
allArgs[pr.first] = HttpSession::urlDecode(pr.second); allArgs[pr.first] = HttpSession::urlDecodeComponent(pr.second);
} }
} else if (parser["Content-Type"].find("application/json") == 0) { } else if (parser["Content-Type"].find("application/json") == 0) {
try { try {
@ -414,7 +415,7 @@ Value makeMediaSourceJson(MediaSource &media){
} }
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
uint16_t openRtpServer(uint16_t local_port, const string &stream_id, int tcp_mode, const string &local_ip, bool re_use_port, uint32_t ssrc, bool only_audio, bool multiplex) { uint16_t openRtpServer(uint16_t local_port, const string &stream_id, int tcp_mode, const string &local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) {
lock_guard<recursive_mutex> lck(s_rtpServerMapMtx); lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
if (s_rtpServerMap.find(stream_id) != s_rtpServerMap.end()) { if (s_rtpServerMap.find(stream_id) != s_rtpServerMap.end()) {
//为了防止RtpProcess所有权限混乱的问题，不允许重复添加相同的stream_id //为了防止RtpProcess所有权限混乱的问题，不允许重复添加相同的stream_id
@ -422,7 +423,7 @@ uint16_t openRtpServer(uint16_t local_port, const string &stream_id, int tcp_mod
} }
RtpServer::Ptr server = std::make_shared<RtpServer>(); RtpServer::Ptr server = std::make_shared<RtpServer>();
server->start(local_port, stream_id, (RtpServer::TcpMode)tcp_mode, local_ip.c_str(), re_use_port, ssrc, only_audio, multiplex); server->start(local_port, stream_id, (RtpServer::TcpMode)tcp_mode, local_ip.c_str(), re_use_port, ssrc, only_track, multiplex);
server->setOnDetach([stream_id]() { server->setOnDetach([stream_id]() {
//设置rtp超时移除事件 //设置rtp超时移除事件
lock_guard<recursive_mutex> lck(s_rtpServerMapMtx); lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
@ -1197,8 +1198,17 @@ void installWebApi() {
//兼容老版本请求，新版本去除enable_tcp参数并新增tcp_mode参数 //兼容老版本请求，新版本去除enable_tcp参数并新增tcp_mode参数
tcp_mode = 1; tcp_mode = 1;
} }
auto port = openRtpServer(allArgs["port"], stream_id, tcp_mode, "::", allArgs["re_use_port"].as<bool>(), auto only_track = allArgs["only_track"].as<int>();
allArgs["ssrc"].as<uint32_t>(), allArgs["only_audio"].as<bool>()); if (allArgs["only_audio"].as<bool>()) {
// 兼容老版本请求，新版本去除only_audio参数并新增only_track参数
only_track = 1;
}
std::string local_ip = "::";
if (!allArgs["local_ip"].empty()) {
local_ip = allArgs["local_ip"];
}
auto port = openRtpServer(allArgs["port"], stream_id, tcp_mode, local_ip, allArgs["re_use_port"].as<bool>(),
allArgs["ssrc"].as<uint32_t>(), only_track);
if (port == 0) { if (port == 0) {
throw InvalidArgsException("该stream_id已存在"); throw InvalidArgsException("该stream_id已存在");
} }
@ -1215,9 +1225,16 @@ void installWebApi() {
// 兼容老版本请求，新版本去除enable_tcp参数并新增tcp_mode参数 // 兼容老版本请求，新版本去除enable_tcp参数并新增tcp_mode参数
tcp_mode = 1; tcp_mode = 1;
} }
auto only_track = allArgs["only_track"].as<int>();
auto port = openRtpServer( if (allArgs["only_audio"].as<bool>()) {
allArgs["port"], stream_id, tcp_mode, "::", true, 0, allArgs["only_audio"].as<bool>(),true); // 兼容老版本请求，新版本去除only_audio参数并新增only_track参数
only_track = 1;
}
std::string local_ip = "::";
if (!allArgs["local_ip"].empty()) {
local_ip = allArgs["local_ip"];
}
auto port = openRtpServer(allArgs["port"], stream_id, tcp_mode, local_ip, true, 0, only_track,true);
if (port == 0) { if (port == 0) {
throw InvalidArgsException("该stream_id已存在"); throw InvalidArgsException("该stream_id已存在");
} }
@ -1282,7 +1299,10 @@ void installWebApi() {
if (!src) { if (!src) {
throw ApiRetException("can not find the source stream", API::NotFound); throw ApiRetException("can not find the source stream", API::NotFound);
} }
if (!allArgs["use_ps"].empty()) {
// 兼容之前的use_ps参数
allArgs["type"] = allArgs["use_ps"].as<int>();
}
MediaSourceEvent::SendRtpArgs args; MediaSourceEvent::SendRtpArgs args;
args.passive = false; args.passive = false;
args.dst_url = allArgs["dst_url"]; args.dst_url = allArgs["dst_url"];
@ -1292,11 +1312,11 @@ void installWebApi() {
args.is_udp = allArgs["is_udp"]; 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.use_ps = allArgs["use_ps"].empty() ? true : allArgs["use_ps"].as<bool>(); args.type = (MediaSourceEvent::SendRtpArgs::Type)(allArgs["type"].as<int>());
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) << " ps " << args.use_ps << " audio " << args.only_audio; TraceL << "startSendRtp, pt " << int(args.pt) << " rtp type " << args.type << " audio " << args.only_audio;
src->getOwnerPoller()->async([=]() mutable { src->getOwnerPoller()->async([=]() mutable {
src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable { src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable {
@ -1319,18 +1339,23 @@ void installWebApi() {
throw ApiRetException("can not find the source stream", API::NotFound); throw ApiRetException("can not find the source stream", API::NotFound);
} }
if (!allArgs["use_ps"].empty()) {
// 兼容之前的use_ps参数
allArgs["type"] = allArgs["use_ps"].as<int>();
}
MediaSourceEvent::SendRtpArgs args; MediaSourceEvent::SendRtpArgs args;
args.passive = true; args.passive = true;
args.ssrc = allArgs["ssrc"]; args.ssrc = allArgs["ssrc"];
args.is_udp = false; args.is_udp = false;
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.use_ps = allArgs["use_ps"].empty() ? true : allArgs["use_ps"].as<bool>(); args.type = (MediaSourceEvent::SendRtpArgs::Type)(allArgs["type"].as<int>());
args.only_audio = allArgs["only_audio"].as<bool>(); args.only_audio = allArgs["only_audio"].as<bool>();
args.recv_stream_id = allArgs["recv_stream_id"]; args.recv_stream_id = allArgs["recv_stream_id"];
//tcp被动服务器等待链接超时时间 //tcp被动服务器等待链接超时时间
args.tcp_passive_close_delay_ms = allArgs["close_delay_ms"]; args.tcp_passive_close_delay_ms = allArgs["close_delay_ms"];
TraceL << "startSendRtpPassive, pt " << int(args.pt) << " ps " << args.use_ps << " audio " << args.only_audio; TraceL << "startSendRtpPassive, pt " << int(args.pt) << " rtp type " << args.type << " audio " << args.only_audio;
src->getOwnerPoller()->async([=]() mutable { src->getOwnerPoller()->async([=]() mutable {
src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable { src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable {
@ -1740,16 +1765,26 @@ void installWebApi() {
auto type = allArgs["type"]; auto type = allArgs["type"];
auto offer = allArgs.getArgs(); auto offer = allArgs.getArgs();
CHECK(!offer.empty(), "http body(webrtc offer sdp) is empty"); CHECK(!offer.empty(), "http body(webrtc offer sdp) is empty");
std::string host = allArgs.getParser()["Host"];
std::string localIp = host.substr(0, host.find(':'));
auto isVaildIP = [](std::string ip)-> bool {
int a,b,c,d;
return sscanf(ip.c_str(),"%d.%d.%d.%d", &a, &b, &c, &d) == 4;
};
if (!isVaildIP(localIp) || localIp=="127.0.0.1") {
localIp = "";
}
auto args = std::make_shared<WebRtcArgsImp>(allArgs, sender.getIdentifier()); auto args = std::make_shared<WebRtcArgsImp>(allArgs, sender.getIdentifier());
WebRtcPluginManager::Instance().getAnswerSdp(static_cast<Session&>(sender), type, *args, WebRtcPluginManager::Instance().getAnswerSdp(static_cast<Session&>(sender), type, *args, [invoker, val, offer, headerOut, localIp](const WebRtcInterface &exchanger) mutable {
[invoker, val, offer, headerOut](const WebRtcInterface &exchanger) mutable {
//设置返回类型 //设置返回类型
headerOut["Content-Type"] = HttpFileManager::getContentType(".json"); headerOut["Content-Type"] = HttpFileManager::getContentType(".json");
//设置跨域 //设置跨域
headerOut["Access-Control-Allow-Origin"] = "*"; headerOut["Access-Control-Allow-Origin"] = "*";
try { try {
setLocalIp(exchanger,localIp);
val["sdp"] = exchangeSdp(exchanger, offer); val["sdp"] = exchangeSdp(exchanger, offer);
val["id"] = exchanger.getIdentifier(); val["id"] = exchanger.getIdentifier();
val["type"] = "answer"; val["type"] = "answer";

2
server/WebApi.h
View File

@ -233,7 +233,7 @@ void installWebApi();
void unInstallWebApi(); void unInstallWebApi();
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
uint16_t openRtpServer(uint16_t local_port, const std::string &stream_id, int tcp_mode, const std::string &local_ip, bool re_use_port, uint32_t ssrc, bool only_audio, bool multiplex=false); uint16_t openRtpServer(uint16_t local_port, const std::string &stream_id, int tcp_mode, const std::string &local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex=false);
void connectRtpServer(const std::string &stream_id, const std::string &dst_url, uint16_t dst_port, const std::function<void(const toolkit::SockException &ex)> &cb); void connectRtpServer(const std::string &stream_id, const std::string &dst_url, uint16_t dst_port, const std::function<void(const toolkit::SockException &ex)> &cb);
bool closeRtpServer(const std::string &stream_id); bool closeRtpServer(const std::string &stream_id);
#endif #endif

3
src/Common/MediaSink.cpp
View File

@ -37,6 +37,7 @@ bool MediaSink::addTrack(const Track::Ptr &track_in) {
} }
// 克隆Track，只拷贝其数据，不拷贝其数据转发关系 // 克隆Track，只拷贝其数据，不拷贝其数据转发关系
auto track = track_in->clone(); auto track = track_in->clone();
CHECK(track, "Clone track failed: ", track_in->getCodecName());
auto index = track->getIndex(); auto index = track->getIndex();
if (!_track_map.emplace(index, std::make_pair(track, false)).second) { if (!_track_map.emplace(index, std::make_pair(track, false)).second) {
WarnL << "Already add a same track: " << track->getIndex() << ", codec: " << track->getCodecName(); WarnL << "Already add a same track: " << track->getIndex() << ", codec: " << track->getCodecName();
@ -132,7 +133,7 @@ void MediaSink::checkTrackIfReady() {
} }
GET_CONFIG(uint32_t, kMaxAddTrackMS, General::kWaitAddTrackMS); GET_CONFIG(uint32_t, kMaxAddTrackMS, General::kWaitAddTrackMS);
if (_track_map.size() == 1 && _ticker.elapsedTime() > kMaxAddTrackMS) { if (_track_map.size() == 1 && (_ticker.elapsedTime() > kMaxAddTrackMS || !_enable_audio)) {
// 如果只有一个Track，那么在该Track添加后，我们最多还等待若干时间(可能后面还会添加Track) // 如果只有一个Track，那么在该Track添加后，我们最多还等待若干时间(可能后面还会添加Track)
emitAllTrackReady(); emitAllTrackReady();
return; return;

5
src/Common/MediaSource.h
View File

@ -92,10 +92,11 @@ public:
class SendRtpArgs { class SendRtpArgs {
public: public:
enum Type { kRtpRAW = 0, kRtpPS = 1, kRtpTS = 2 };
// 是否采用udp方式发送rtp // 是否采用udp方式发送rtp
bool is_udp = true; bool is_udp = true;
// rtp采用ps还是es方式 // rtp类型
bool use_ps = true; Type type = kRtpPS;
//发送es流时指定是否只发送纯音频流 //发送es流时指定是否只发送纯音频流
bool only_audio = false; bool only_audio = false;
//tcp被动方式 //tcp被动方式

4
src/Common/Parser.cpp
View File

@ -294,8 +294,8 @@ void RtspUrl::setup(bool is_ssl, const string &url, const string &user, const st
splitUrl(ip, ip, port); splitUrl(ip, ip, port);
_url = std::move(url); _url = std::move(url);
_user = strCoding::UrlDecode(std::move(user)); _user = strCoding::UrlDecodeComponent(std::move(user));
_passwd = strCoding::UrlDecode(std::move(passwd)); _passwd = strCoding::UrlDecodeComponent(std::move(passwd));
_host = std::move(ip); _host = std::move(ip);
_port = port; _port = port;
_is_ssl = is_ssl; _is_ssl = is_ssl;

10
src/Common/config.cpp
View File

@ -58,6 +58,12 @@ const string kBroadcastStreamNoneReader = "kBroadcastStreamNoneReader";
const string kBroadcastHttpBeforeAccess = "kBroadcastHttpBeforeAccess"; const string kBroadcastHttpBeforeAccess = "kBroadcastHttpBeforeAccess";
const string kBroadcastSendRtpStopped = "kBroadcastSendRtpStopped"; const string kBroadcastSendRtpStopped = "kBroadcastSendRtpStopped";
const string kBroadcastRtpServerTimeout = "kBroadcastRtpServerTimeout"; const string kBroadcastRtpServerTimeout = "kBroadcastRtpServerTimeout";
const string kBroadcastRtcSctpConnecting = "kBroadcastRtcSctpConnecting";
const string kBroadcastRtcSctpConnected = "kBroadcastRtcSctpConnected";
const string kBroadcastRtcSctpFailed = "kBroadcastRtcSctpFailed";
const string kBroadcastRtcSctpClosed = "kBroadcastRtcSctpClosed";
const string kBroadcastRtcSctpSend = "kBroadcastRtcSctpSend";
const string kBroadcastRtcSctpReceived = "kBroadcastRtcSctpReceived";
} // namespace Broadcast } // namespace Broadcast
@ -338,6 +344,8 @@ const string kH265PT = RTP_PROXY_FIELD "h265_pt";
const string kPSPT = RTP_PROXY_FIELD "ps_pt"; const string kPSPT = RTP_PROXY_FIELD "ps_pt";
const string kOpusPT = RTP_PROXY_FIELD "opus_pt"; const string kOpusPT = RTP_PROXY_FIELD "opus_pt";
const string kGopCache = RTP_PROXY_FIELD "gop_cache"; const string kGopCache = RTP_PROXY_FIELD "gop_cache";
const string kRtpG711DurMs = RTP_PROXY_FIELD "rtp_g711_dur_ms";
const string kUdpRecvSocketBuffer = RTP_PROXY_FIELD "udp_recv_socket_buffer";
static onceToken token([]() { static onceToken token([]() {
mINI::Instance()[kDumpDir] = ""; mINI::Instance()[kDumpDir] = "";
@ -348,6 +356,8 @@ static onceToken token([]() {
mINI::Instance()[kPSPT] = 96; mINI::Instance()[kPSPT] = 96;
mINI::Instance()[kOpusPT] = 100; mINI::Instance()[kOpusPT] = 100;
mINI::Instance()[kGopCache] = 1; mINI::Instance()[kGopCache] = 1;
mINI::Instance()[kRtpG711DurMs] = 100;
mINI::Instance()[kUdpRecvSocketBuffer] = 4 * 1024 * 1024;
}); });
} // namespace RtpProxy } // namespace RtpProxy

20
src/Common/config.h
View File

@ -109,6 +109,21 @@ extern const std::string kBroadcastReloadConfig;
extern const std::string kBroadcastRtpServerTimeout; extern const std::string kBroadcastRtpServerTimeout;
#define BroadcastRtpServerTimeoutArgs uint16_t &local_port, const string &stream_id,int &tcp_mode, bool &re_use_port, uint32_t &ssrc #define BroadcastRtpServerTimeoutArgs uint16_t &local_port, const string &stream_id,int &tcp_mode, bool &re_use_port, uint32_t &ssrc
// rtc transport sctp 连接状态
extern const std::string kBroadcastRtcSctpConnecting;
extern const std::string kBroadcastRtcSctpConnected;
extern const std::string kBroadcastRtcSctpFailed;
extern const std::string kBroadcastRtcSctpClosed;
#define BroadcastRtcSctpConnectArgs WebRtcTransport& sender
// rtc transport sctp 发送数据
extern const std::string kBroadcastRtcSctpSend;
#define BroadcastRtcSctpSendArgs WebRtcTransport& sender, const uint8_t *&data, size_t& len
// rtc transport sctp 接收数据
extern const std::string kBroadcastRtcSctpReceived;
#define BroadcastRtcSctpReceivedArgs WebRtcTransport& sender, uint16_t &streamId, uint32_t &ppid, const uint8_t *&msg, size_t &len
#define ReloadConfigTag ((void *)(0xFF)) #define ReloadConfigTag ((void *)(0xFF))
#define RELOAD_KEY(arg, key) \ #define RELOAD_KEY(arg, key) \
do { \ do { \
@ -382,6 +397,11 @@ extern const std::string kPSPT;
extern const std::string kOpusPT; extern const std::string kOpusPT;
// RtpSender相关功能是否提前开启gop缓存优化级联秒开体验，默认开启 // RtpSender相关功能是否提前开启gop缓存优化级联秒开体验，默认开启
extern const std::string kGopCache; extern const std::string kGopCache;
//国标发送g711 rtp 打包时，每个包的语音时长是多少，默认是100 ms，范围为20~180ms (gb28181-2016，c.2.4规定)，
//最好为20 的倍数，程序自动向20的倍数取整
extern const std::string kRtpG711DurMs;
// udp recv socket buffer size
extern const std::string kUdpRecvSocketBuffer;
} // namespace RtpProxy } // namespace RtpProxy
/** /**

90
src/Common/strCoding.cpp
View File

@ -69,6 +69,40 @@ string strCoding::UrlEncode(const string &str) {
return out; return out;
} }
string strCoding::UrlEncodePath(const string &str) {
const char *dont_escape = "!#&'*+:=?@/._-$,;~()";
string out;
size_t len = str.size();
for (size_t i = 0; i < len; ++i) {
char ch = str[i];
if (isalnum((uint8_t) ch) || strchr(dont_escape, (uint8_t) ch) != NULL) {
out.push_back(ch);
} else {
char buf[4];
snprintf(buf, 4, "%%%X%X", (uint8_t) ch >> 4, (uint8_t) ch & 0x0F);
out.append(buf);
}
}
return out;
}
string strCoding::UrlEncodeComponent(const string &str) {
const char *dont_escape = "!'()*-._~";
string out;
size_t len = str.size();
for (size_t i = 0; i < len; ++i) {
char ch = str[i];
if (isalnum((uint8_t) ch) || strchr(dont_escape, (uint8_t) ch) != NULL) {
out.push_back(ch);
} else {
char buf[4];
snprintf(buf, 4, "%%%X%X", (uint8_t) ch >> 4, (uint8_t) ch & 0x0F);
out.append(buf);
}
}
return out;
}
string strCoding::UrlDecode(const string &str) { string strCoding::UrlDecode(const string &str) {
string output; string output;
size_t i = 0, len = str.length(); size_t i = 0, len = str.length();
@ -95,6 +129,62 @@ string strCoding::UrlDecode(const string &str) {
return output; return output;
} }
string strCoding::UrlDecodePath(const string &str) {
const char *dont_unescape = "#$&+,/:;=?@";
string output;
size_t i = 0, len = str.length();
while (i < len) {
if (str[i] == '%') {
if (i + 3 > len) {
// %后面必须还有两个字节才会反转义
output.append(str, i, len - i);
break;
}
char ch = HexStrToBin(&(str[i + 1]));
if (ch == -1 || strchr(dont_unescape, (unsigned char)ch) != NULL) {
// %后面两个字节不是16进制字符串，转义失败；或者转义出来可能会造成url包含非path部分，比如#?，说明提交的是非法拼接的url；直接拼接3个原始字符
output.append(str, i, 3);
} else {
output += ch;
}
i += 3;
} else {
output += str[i];
++i;
}
}
return output;
}
std::string strCoding::UrlDecodeComponent(const std::string &str) {
string output;
size_t i = 0, len = str.length();
while (i < len) {
if (str[i] == '%') {
if (i + 3 > len) {
// %后面必须还有两个字节才会反转义
output.append(str, i, len - i);
break;
}
char ch = HexStrToBin(&(str[i + 1]));
if (ch == -1) {
// %后面两个字节不是16进制字符串，转义失败；直接拼接3个原始字符
output.append(str, i, 3);
} else {
output += ch;
}
i += 3;
} else if (str[i] == '+') {
output += ' ';
++i;
} else {
output += str[i];
++i;
}
}
return output;
}
#if 0 #if 0
#include "Util/onceToken.h" #include "Util/onceToken.h"
static toolkit::onceToken token([]() { static toolkit::onceToken token([]() {

8
src/Common/strCoding.h
View File

@ -18,8 +18,12 @@ namespace mediakit {
class strCoding { class strCoding {
public: public:
static std::string UrlEncode(const std::string &str); //urlutf8 编码 [[deprecated]] static std::string UrlEncode(const std::string &str); //url utf8编码, deprecated
static std::string UrlDecode(const std::string &str); //urlutf8解码 static std::string UrlEncodePath(const std::string &str); //url路径 utf8编码
static std::string UrlEncodeComponent(const std::string &str); // url参数 utf8编码
[[deprecated]] static std::string UrlDecode(const std::string &str); //url utf8解码, deprecated
static std::string UrlDecodePath(const std::string &str); //url路径 utf8解码
static std::string UrlDecodeComponent(const std::string &str); // url参数 utf8解码
#if defined(_WIN32) #if defined(_WIN32)
static std::string UTF8ToGB2312(const std::string &str);//utf_8转为gb2312 static std::string UTF8ToGB2312(const std::string &str);//utf_8转为gb2312
static std::string GB2312ToUTF8(const std::string &str); //gb2312 转utf_8 static std::string GB2312ToUTF8(const std::string &str); //gb2312 转utf_8

72
src/Http/HttpBody.cpp
View File

@ -28,10 +28,6 @@
#include "HttpClient.h" #include "HttpClient.h"
#include "Common/macros.h" #include "Common/macros.h"
#ifndef _WIN32
#define ENABLE_MMAP
#endif
using namespace std; using namespace std;
using namespace toolkit; using namespace toolkit;
@ -57,12 +53,25 @@ Buffer::Ptr HttpStringBody::readData(size_t size) {
} }
////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////
#ifdef ENABLE_MMAP
static mutex s_mtx; static mutex s_mtx;
static unordered_map<string /*file_path*/, std::tuple<char */*ptr*/, int64_t /*size*/, weak_ptr<char> /*mmap*/ > > s_shared_mmap; static unordered_map<string /*file_path*/, std::tuple<char */*ptr*/, int64_t /*size*/, weak_ptr<char> /*mmap*/ > > s_shared_mmap;
#if defined(_WIN32)
static void mmap_close(HANDLE _hfile, HANDLE _hmapping, void *_addr) {
if (_addr) {
::UnmapViewOfFile(_addr);
}
if (_hmapping) {
::CloseHandle(_hmapping);
}
if (_hfile != INVALID_HANDLE_VALUE) {
::CloseHandle(_hfile);
}
}
#endif
//删除mmap记录 //删除mmap记录
static void delSharedMmap(const string &file_path, char *ptr) { static void delSharedMmap(const string &file_path, char *ptr) {
lock_guard<mutex> lck(s_mtx); lock_guard<mutex> lck(s_mtx);
@ -97,24 +106,67 @@ static std::shared_ptr<char> getSharedMmap(const string &file_path, int64_t &fil
file_size = -1; file_size = -1;
return nullptr; return nullptr;
} }
#if defined(_WIN32)
auto fd = _fileno(fp.get());
#else
//获取文件大小 //获取文件大小
file_size = File::fileSize(fp.get()); file_size = File::fileSize(fp.get());
auto fd = fileno(fp.get());
#endif
int fd = fileno(fp.get());
if (fd < 0) { if (fd < 0) {
WarnL << "fileno failed:" << get_uv_errmsg(false); WarnL << "fileno failed:" << get_uv_errmsg(false);
return nullptr; return nullptr;
} }
#ifndef _WIN32
auto ptr = (char *)mmap(NULL, file_size, PROT_READ, MAP_SHARED, fd, 0); auto ptr = (char *)mmap(NULL, file_size, PROT_READ, MAP_SHARED, fd, 0);
if (ptr == MAP_FAILED) { if (ptr == MAP_FAILED) {
WarnL << "mmap " << file_path << " failed:" << get_uv_errmsg(false); WarnL << "mmap " << file_path << " failed:" << get_uv_errmsg(false);
return nullptr; return nullptr;
} }
std::shared_ptr<char> ret(ptr, [file_size, fp, file_path](char *ptr) { std::shared_ptr<char> ret(ptr, [file_size, fp, file_path](char *ptr) {
munmap(ptr, file_size); munmap(ptr, file_size);
delSharedMmap(file_path, ptr); delSharedMmap(file_path, ptr);
}); });
#else
auto hfile = ::CreateFileA(file_path.data(), GENERIC_READ, FILE_SHARE_READ, NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL);
if (hfile == INVALID_HANDLE_VALUE) {
WarnL << "CreateFileA() " << file_path << " failed:";
return nullptr;
}
file_size = ::GetFileSize(hfile, NULL);
auto hmapping = ::CreateFileMapping(hfile, NULL, PAGE_READONLY, 0, 0, NULL);
if (hmapping == NULL) {
mmap_close(hfile, NULL, NULL);
WarnL << "CreateFileMapping() " << file_path << " failed:";
return nullptr;
}
auto addr_ = ::MapViewOfFile(hmapping, FILE_MAP_READ, 0, 0, 0);
if (addr_ == nullptr) {
mmap_close(hfile, hmapping, addr_);
WarnL << "MapViewOfFile() " << file_path << " failed:";
return nullptr;
}
std::shared_ptr<char> ret((char *)(addr_), [hfile, hmapping, file_path](char *addr_) {
mmap_close(hfile, hmapping, addr_);
delSharedMmap(file_path, addr_);
});
#endif
#if 0 #if 0
if (file_size < 10 * 1024 * 1024 && file_path.rfind(".ts") != string::npos) { if (file_size < 10 * 1024 * 1024 && file_path.rfind(".ts") != string::npos) {
//如果是小ts文件，那么尝试先加载到内存 //如果是小ts文件，那么尝试先加载到内存
@ -131,14 +183,12 @@ static std::shared_ptr<char> getSharedMmap(const string &file_path, int64_t &fil
} }
return ret; return ret;
} }
#endif
HttpFileBody::HttpFileBody(const string &file_path, bool use_mmap) { HttpFileBody::HttpFileBody(const string &file_path, bool use_mmap) {
#ifdef ENABLE_MMAP
if (use_mmap ) { if (use_mmap ) {
_map_addr = getSharedMmap(file_path, _read_to); _map_addr = getSharedMmap(file_path, _read_to);
} }
#endif
if (!_map_addr && _read_to != -1) { if (!_map_addr && _read_to != -1) {
//mmap失败(且不是由于文件不存在导致的)或未执行mmap时，才进入fread逻辑分支 //mmap失败(且不是由于文件不存在导致的)或未执行mmap时，才进入fread逻辑分支
_fp.reset(fopen(file_path.data(), "rb"), [](FILE *fp) { _fp.reset(fopen(file_path.data(), "rb"), [](FILE *fp) {

2
src/Http/HttpClient.h
View File

@ -34,7 +34,7 @@ public:
for (auto &pr : *this) { for (auto &pr : *this) {
ret.append(pr.first); ret.append(pr.first);
ret.append("="); ret.append("=");
ret.append(strCoding::UrlEncode(pr.second)); ret.append(strCoding::UrlEncodeComponent(pr.second));
ret.append("&"); ret.append("&");
} }
if (ret.size()) { if (ret.size()) {

2
src/Http/HttpFileManager.cpp
View File

@ -228,7 +228,7 @@ static bool makeFolderMenu(const string &httpPath, const string &strFullPath, st
multimap<string/*url name*/, std::pair<string/*note name*/, string/*file path*/> > file_map; multimap<string/*url name*/, std::pair<string/*note name*/, string/*file path*/> > file_map;
File::scanDir(strPathPrefix, [&](const std::string &path, bool isDir) { File::scanDir(strPathPrefix, [&](const std::string &path, bool isDir) {
auto name = fileName(strPathPrefix, path); auto name = fileName(strPathPrefix, path);
file_map.emplace(strCoding::UrlEncode(name), std::make_pair(name, path)); file_map.emplace(strCoding::UrlEncodePath(name), std::make_pair(name, path));
return true; return true;
}); });
//如果是root目录，添加虚拟目录 //如果是root目录，添加虚拟目录

33
src/Http/HttpSession.cpp
View File

@ -65,6 +65,7 @@ ssize_t HttpSession::onRecvHeader(const char *header, size_t len) {
_parser.parse(header, len); _parser.parse(header, len);
CHECK(_parser.url()[0] == '/'); CHECK(_parser.url()[0] == '/');
_origin = _parser["Origin"];
urlDecode(_parser); urlDecode(_parser);
auto &cmd = _parser.method(); auto &cmd = _parser.method();
@ -606,8 +607,8 @@ void HttpSession::sendResponse(int code,
headerOut.emplace("Connection", bClose ? "close" : "keep-alive"); headerOut.emplace("Connection", bClose ? "close" : "keep-alive");
GET_CONFIG(bool, allow_cross_domains, Http::kAllowCrossDomains); GET_CONFIG(bool, allow_cross_domains, Http::kAllowCrossDomains);
if (allow_cross_domains) { if (allow_cross_domains && !_origin.empty()) {
headerOut.emplace("Access-Control-Allow-Origin", "*"); headerOut.emplace("Access-Control-Allow-Origin", _origin);
headerOut.emplace("Access-Control-Allow-Credentials", "true"); headerOut.emplace("Access-Control-Allow-Credentials", "true");
} }
@ -694,10 +695,34 @@ string HttpSession::urlDecode(const string &str) {
return ret; return ret;
} }
string HttpSession::urlDecodePath(const string &str) {
auto ret = strCoding::UrlDecodePath(str);
#ifdef _WIN32
GET_CONFIG(string, charSet, Http::kCharSet);
bool isGb2312 = !strcasecmp(charSet.data(), "gb2312");
if (isGb2312) {
ret = strCoding::UTF8ToGB2312(ret);
}
#endif // _WIN32
return ret;
}
string HttpSession::urlDecodeComponent(const string &str) {
auto ret = strCoding::UrlDecodeComponent(str);
#ifdef _WIN32
GET_CONFIG(string, charSet, Http::kCharSet);
bool isGb2312 = !strcasecmp(charSet.data(), "gb2312");
if (isGb2312) {
ret = strCoding::UTF8ToGB2312(ret);
}
#endif // _WIN32
return ret;
}
void HttpSession::urlDecode(Parser &parser) { void HttpSession::urlDecode(Parser &parser) {
parser.setUrl(urlDecode(parser.url())); parser.setUrl(urlDecodePath(parser.url()));
for (auto &pr : _parser.getUrlArgs()) { for (auto &pr : _parser.getUrlArgs()) {
const_cast<string &>(pr.second) = urlDecode(pr.second); const_cast<string &>(pr.second) = urlDecodeComponent(pr.second);
} }
} }

6
src/Http/HttpSession.h
View File

@ -44,7 +44,9 @@ public:
void onRecv(const toolkit::Buffer::Ptr &) override; void onRecv(const toolkit::Buffer::Ptr &) override;
void onError(const toolkit::SockException &err) override; void onError(const toolkit::SockException &err) override;
void onManager() override; void onManager() override;
static std::string urlDecode(const std::string &str); [[deprecated]] static std::string urlDecode(const std::string &str);
static std::string urlDecodePath(const std::string &str);
static std::string urlDecodeComponent(const std::string &str);
void setTimeoutSec(size_t second); void setTimeoutSec(size_t second);
void setMaxReqSize(size_t max_req_size); void setMaxReqSize(size_t max_req_size);
@ -136,6 +138,8 @@ private:
size_t _max_req_size = 0; size_t _max_req_size = 0;
//消耗的总流量 //消耗的总流量
uint64_t _total_bytes_usage = 0; uint64_t _total_bytes_usage = 0;
// http请求中的 Origin字段
std::string _origin;
Parser _parser; Parser _parser;
toolkit::Ticker _ticker; toolkit::Ticker _ticker;
TSMediaSource::RingType::RingReader::Ptr _ts_reader; TSMediaSource::RingType::RingReader::Ptr _ts_reader;

29
src/Player/PlayerBase.cpp
View File

@ -23,11 +23,16 @@ namespace mediakit {
PlayerBase::Ptr PlayerBase::createPlayer(const EventPoller::Ptr &in_poller, const string &url_in) { PlayerBase::Ptr PlayerBase::createPlayer(const EventPoller::Ptr &in_poller, const string &url_in) {
auto poller = in_poller ? in_poller : EventPollerPool::Instance().getPoller(); auto poller = in_poller ? in_poller : EventPollerPool::Instance().getPoller();
static auto releasePlayer = [poller](PlayerBase *ptr) { std::weak_ptr<EventPoller> weak_poller = poller;
poller->async([ptr]() { static auto release_func = [weak_poller](PlayerBase *ptr) {
onceToken token(nullptr, [&]() { delete ptr; }); if (auto poller = weak_poller.lock()) {
ptr->teardown(); poller->async([ptr]() {
}); onceToken token(nullptr, [&]() { delete ptr; });
ptr->teardown();
});
} else {
delete ptr;
}
}; };
string url = url_in; string url = url_in;
string prefix = findSubString(url.data(), NULL, "://"); string prefix = findSubString(url.data(), NULL, "://");
@ -38,29 +43,29 @@ PlayerBase::Ptr PlayerBase::createPlayer(const EventPoller::Ptr &in_poller, cons
} }
if (strcasecmp("rtsps", prefix.data()) == 0) { if (strcasecmp("rtsps", prefix.data()) == 0) {
return PlayerBase::Ptr(new TcpClientWithSSL<RtspPlayerImp>(poller), releasePlayer); return PlayerBase::Ptr(new TcpClientWithSSL<RtspPlayerImp>(poller), release_func);
} }
if (strcasecmp("rtsp", prefix.data()) == 0) { if (strcasecmp("rtsp", prefix.data()) == 0) {
return PlayerBase::Ptr(new RtspPlayerImp(poller), releasePlayer); return PlayerBase::Ptr(new RtspPlayerImp(poller), release_func);
} }
if (strcasecmp("rtmps", prefix.data()) == 0) { if (strcasecmp("rtmps", prefix.data()) == 0) {
return PlayerBase::Ptr(new TcpClientWithSSL<RtmpPlayerImp>(poller), releasePlayer); return PlayerBase::Ptr(new TcpClientWithSSL<RtmpPlayerImp>(poller), release_func);
} }
if (strcasecmp("rtmp", prefix.data()) == 0) { if (strcasecmp("rtmp", prefix.data()) == 0) {
return PlayerBase::Ptr(new RtmpPlayerImp(poller), releasePlayer); return PlayerBase::Ptr(new RtmpPlayerImp(poller), release_func);
} }
if ((strcasecmp("http", prefix.data()) == 0 || strcasecmp("https", prefix.data()) == 0)) { if ((strcasecmp("http", prefix.data()) == 0 || strcasecmp("https", prefix.data()) == 0)) {
if (end_with(url, ".m3u8") || end_with(url_in, ".m3u8")) { if (end_with(url, ".m3u8") || end_with(url_in, ".m3u8")) {
return PlayerBase::Ptr(new HlsPlayerImp(poller), releasePlayer); return PlayerBase::Ptr(new HlsPlayerImp(poller), release_func);
} }
if (end_with(url, ".ts") || end_with(url_in, ".ts")) { if (end_with(url, ".ts") || end_with(url_in, ".ts")) {
return PlayerBase::Ptr(new TsPlayerImp(poller), releasePlayer); return PlayerBase::Ptr(new TsPlayerImp(poller), release_func);
} }
if (end_with(url, ".flv") || end_with(url_in, ".flv")) { if (end_with(url, ".flv") || end_with(url_in, ".flv")) {
return PlayerBase::Ptr(new FlvPlayerImp(poller), releasePlayer); return PlayerBase::Ptr(new FlvPlayerImp(poller), release_func);
} }
} }

26
src/Pusher/PusherBase.cpp
View File

@ -17,31 +17,37 @@ using namespace toolkit;
namespace mediakit { namespace mediakit {
PusherBase::Ptr PusherBase::createPusher(const EventPoller::Ptr &poller, PusherBase::Ptr PusherBase::createPusher(const EventPoller::Ptr &in_poller,
const MediaSource::Ptr &src, const MediaSource::Ptr &src,
const std::string & url) { const std::string & url) {
static auto releasePusher = [](PusherBase *ptr){ auto poller = in_poller ? in_poller : EventPollerPool::Instance().getPoller();
onceToken token(nullptr,[&](){ std::weak_ptr<EventPoller> weak_poller = poller;
delete ptr; static auto release_func = [weak_poller](PusherBase *ptr) {
}); if (auto poller = weak_poller.lock()) {
ptr->teardown(); poller->async([ptr]() {
onceToken token(nullptr, [&]() { delete ptr; });
ptr->teardown();
});
} else {
delete ptr;
}
}; };
std::string prefix = findSubString(url.data(), NULL, "://"); std::string prefix = findSubString(url.data(), NULL, "://");
if (strcasecmp("rtsps",prefix.data()) == 0) { if (strcasecmp("rtsps",prefix.data()) == 0) {
return PusherBase::Ptr(new TcpClientWithSSL<RtspPusherImp>(poller, std::dynamic_pointer_cast<RtspMediaSource>(src)), releasePusher); return PusherBase::Ptr(new TcpClientWithSSL<RtspPusherImp>(poller, std::dynamic_pointer_cast<RtspMediaSource>(src)), release_func);
} }
if (strcasecmp("rtsp",prefix.data()) == 0) { if (strcasecmp("rtsp",prefix.data()) == 0) {
return PusherBase::Ptr(new RtspPusherImp(poller, std::dynamic_pointer_cast<RtspMediaSource>(src)), releasePusher); return PusherBase::Ptr(new RtspPusherImp(poller, std::dynamic_pointer_cast<RtspMediaSource>(src)), release_func);
} }
if (strcasecmp("rtmps",prefix.data()) == 0) { if (strcasecmp("rtmps",prefix.data()) == 0) {
return PusherBase::Ptr(new TcpClientWithSSL<RtmpPusherImp>(poller, std::dynamic_pointer_cast<RtmpMediaSource>(src)), releasePusher); return PusherBase::Ptr(new TcpClientWithSSL<RtmpPusherImp>(poller, std::dynamic_pointer_cast<RtmpMediaSource>(src)), release_func);
} }
if (strcasecmp("rtmp",prefix.data()) == 0) { if (strcasecmp("rtmp",prefix.data()) == 0) {
return PusherBase::Ptr(new RtmpPusherImp(poller, std::dynamic_pointer_cast<RtmpMediaSource>(src)), releasePusher); return PusherBase::Ptr(new RtmpPusherImp(poller, std::dynamic_pointer_cast<RtmpMediaSource>(src)), release_func);
} }
throw std::invalid_argument("not supported push schema:" + url); throw std::invalid_argument("not supported push schema:" + url);

12
src/Rtp/PSEncoder.cpp
View File

@ -19,9 +19,17 @@ using namespace toolkit;
namespace mediakit{ namespace mediakit{
PSEncoderImp::PSEncoderImp(uint32_t ssrc, uint8_t payload_type) : MpegMuxer(true) { PSEncoderImp::PSEncoderImp(uint32_t ssrc, uint8_t payload_type, bool ps_or_ts) : MpegMuxer(ps_or_ts) {
GET_CONFIG(uint32_t,video_mtu,Rtp::kVideoMtuSize); GET_CONFIG(uint32_t, s_video_mtu, Rtp::kVideoMtuSize);
_rtp_encoder = std::make_shared<CommonRtpEncoder>(); _rtp_encoder = std::make_shared<CommonRtpEncoder>();
auto video_mtu = s_video_mtu;
if (!ps_or_ts) {
// 确保ts rtp负载部分长度是188的倍数
video_mtu = RtpPacket::kRtpHeaderSize + (s_video_mtu - (s_video_mtu % 188));
if (video_mtu > s_video_mtu) {
video_mtu -= 188;
}
}
_rtp_encoder->setRtpInfo(ssrc, video_mtu, 90000, payload_type); _rtp_encoder->setRtpInfo(ssrc, video_mtu, 90000, payload_type);
auto ring = std::make_shared<RtpRing::RingType>(); auto ring = std::make_shared<RtpRing::RingType>();
ring->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key) { onRTP(std::move(rtp), is_key); })); ring->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key) { onRTP(std::move(rtp), is_key); }));

14
src/Rtp/PSEncoder.h
View File

@ -16,11 +16,19 @@
#include "Record/MPEG.h" #include "Record/MPEG.h"
#include "Common/MediaSink.h" #include "Common/MediaSink.h"
namespace mediakit{ namespace mediakit {
class CommonRtpEncoder; class CommonRtpEncoder;
class PSEncoderImp : public MpegMuxer{
class PSEncoderImp : public MpegMuxer {
public: public:
PSEncoderImp(uint32_t ssrc, uint8_t payload_type = 96); /**
* 创建psh或ts rtp编码器
* @param ssrc rtp的ssrc
* @param payload_type rtp的pt
* @param ps_or_ts true: ps, false: ts
*/
PSEncoderImp(uint32_t ssrc, uint8_t payload_type = 96, bool ps_or_ts = true);
~PSEncoderImp() override; ~PSEncoderImp() override;
protected: protected:

6
src/Rtp/RawEncoder.cpp
View File

@ -34,6 +34,12 @@ bool RawEncoderImp::addTrack(const Track::Ptr &track) {
auto ring = std::make_shared<RtpRing::RingType>(); auto ring = std::make_shared<RtpRing::RingType>();
ring->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key) { onRTP(std::move(rtp), true); })); ring->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key) { onRTP(std::move(rtp), true); }));
_rtp_encoder->setRtpRing(std::move(ring)); _rtp_encoder->setRtpRing(std::move(ring));
if (track->getCodecId() == CodecG711A || track->getCodecId() == CodecG711U) {
GET_CONFIG(uint32_t, dur_ms, RtpProxy::kRtpG711DurMs);
Any param;
param.set<uint32_t>(dur_ms);
_rtp_encoder->setOpt(RtpCodec::RTP_ENCODER_PKT_DUR_MS, param);
}
return true; return true;
} }

7
src/Rtp/RtpCache.h
View File

@ -40,7 +40,9 @@ 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) : RtpCache(std::move(cb)), PSEncoderImp(ssrc, payload_type) {}; 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) {};
void flush() override; void flush() override;
protected: protected:
@ -56,6 +58,7 @@ protected:
void onRTP(toolkit::Buffer::Ptr rtp, bool is_key = false) override; void onRTP(toolkit::Buffer::Ptr rtp, bool is_key = false) override;
}; };
}//namespace mediakit } //namespace mediakit
#endif//ENABLE_RTPPROXY #endif//ENABLE_RTPPROXY
#endif //ZLMEDIAKIT_RTPCACHE_H #endif //ZLMEDIAKIT_RTPCACHE_H

14
src/Rtp/RtpProcess.cpp
View File

@ -11,6 +11,7 @@
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
#include "GB28181Process.h" #include "GB28181Process.h"
#include "RtpProcess.h" #include "RtpProcess.h"
#include "RtpSelector.h"
#include "Http/HttpTSPlayer.h" #include "Http/HttpTSPlayer.h"
#include "Util/File.h" #include "Util/File.h"
#include "Common/config.h" #include "Common/config.h"
@ -198,8 +199,8 @@ void RtpProcess::setStopCheckRtp(bool is_check){
} }
} }
void RtpProcess::setOnlyAudio(bool only_audio){ void RtpProcess::setOnlyTrack(OnlyTrack only_track) {
_only_audio = only_audio; _only_track = only_track;
} }
void RtpProcess::onDetach() { void RtpProcess::onDetach() {
@ -255,8 +256,13 @@ void RtpProcess::emitOnPublish() {
} }
if (err.empty()) { if (err.empty()) {
strong_self->_muxer = std::make_shared<MultiMediaSourceMuxer>(strong_self->_media_info, 0.0f, option); strong_self->_muxer = std::make_shared<MultiMediaSourceMuxer>(strong_self->_media_info, 0.0f, option);
if (strong_self->_only_audio) { if (!option.stream_replace.empty()) {
strong_self->_muxer->setOnlyAudio(); RtpSelector::Instance().addStreamReplace(strong_self->_media_info.stream, option.stream_replace);
}
switch (strong_self->_only_track) {
case kOnlyAudio: strong_self->_muxer->setOnlyAudio(); break;
case kOnlyVideo: strong_self->_muxer->enableAudio(false); break;
default: break;
} }
strong_self->_muxer->setMediaListener(strong_self); strong_self->_muxer->setMediaListener(strong_self);
strong_self->doCachedFunc(); strong_self->doCachedFunc();

7
src/Rtp/RtpProcess.h
View File

@ -24,6 +24,7 @@ public:
friend class RtpProcessHelper; friend class RtpProcessHelper;
RtpProcess(const std::string &stream_id); RtpProcess(const std::string &stream_id);
~RtpProcess(); ~RtpProcess();
enum OnlyTrack { kAll = 0, kOnlyAudio = 1, kOnlyVideo = 2 };
/** /**
* 输入rtp * 输入rtp
@ -58,10 +59,10 @@ public:
void setStopCheckRtp(bool is_check=false); void setStopCheckRtp(bool is_check=false);
/** /**
* 设置为单track，单音频时可以加快媒体注册速度 * 设置为单track，单音频/单视频时可以加快媒体注册速度
* 请在inputRtp前调用此方法，否则可能会是空操作 * 请在inputRtp前调用此方法，否则可能会是空操作
*/ */
void setOnlyAudio(bool only_audio); void setOnlyTrack(OnlyTrack only_track);
/** /**
* flush输出缓存 * flush输出缓存
@ -93,7 +94,7 @@ private:
void doCachedFunc(); void doCachedFunc();
private: private:
bool _only_audio = false; OnlyTrack _only_track = kAll;
std::string _auth_err; std::string _auth_err;
uint64_t _dts = 0; uint64_t _dts = 0;
uint64_t _total_bytes = 0; uint64_t _total_bytes = 0;

31
src/Rtp/RtpSelector.cpp
View File

@ -23,6 +23,7 @@ INSTANCE_IMP(RtpSelector);
void RtpSelector::clear(){ void RtpSelector::clear(){
lock_guard<decltype(_mtx_map)> lck(_mtx_map); lock_guard<decltype(_mtx_map)> lck(_mtx_map);
_map_rtp_process.clear(); _map_rtp_process.clear();
_map_stream_replace.clear();
} }
bool RtpSelector::getSSRC(const char *data, size_t data_len, uint32_t &ssrc){ bool RtpSelector::getSSRC(const char *data, size_t data_len, uint32_t &ssrc){
@ -36,17 +37,23 @@ bool RtpSelector::getSSRC(const char *data, size_t data_len, uint32_t &ssrc){
RtpProcess::Ptr RtpSelector::getProcess(const string &stream_id,bool makeNew) { RtpProcess::Ptr RtpSelector::getProcess(const string &stream_id,bool makeNew) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map); lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(stream_id); string stream_id_origin = stream_id;
auto it_replace = _map_stream_replace.find(stream_id);
if (it_replace != _map_stream_replace.end()) {
stream_id_origin = it_replace->second;
}
auto it = _map_rtp_process.find(stream_id_origin);
if (it == _map_rtp_process.end() && !makeNew) { if (it == _map_rtp_process.end() && !makeNew) {
return nullptr; return nullptr;
} }
if (it != _map_rtp_process.end() && makeNew) { if (it != _map_rtp_process.end() && makeNew) {
//已经被其他线程持有了，不得再被持有，否则会存在线程安全的问题 //已经被其他线程持有了，不得再被持有，否则会存在线程安全的问题
throw ProcessExisted(StrPrinter << "RtpProcess(" << stream_id << ") already existed"); throw ProcessExisted(StrPrinter << "RtpProcess(" << stream_id_origin << ") already existed");
} }
RtpProcessHelper::Ptr &ref = _map_rtp_process[stream_id]; RtpProcessHelper::Ptr &ref = _map_rtp_process[stream_id_origin];
if (!ref) { if (!ref) {
ref = std::make_shared<RtpProcessHelper>(stream_id, shared_from_this()); ref = std::make_shared<RtpProcessHelper>(stream_id_origin, shared_from_this());
ref->attachEvent(); ref->attachEvent();
createTimer(); createTimer();
} }
@ -81,10 +88,25 @@ void RtpSelector::delProcess(const string &stream_id,const RtpProcess *ptr) {
} }
process = it->second->getProcess(); process = it->second->getProcess();
_map_rtp_process.erase(it); _map_rtp_process.erase(it);
delStreamReplace(stream_id);
} }
process->onDetach(); process->onDetach();
} }
void RtpSelector::addStreamReplace(const string &stream_id, const std::string &stream_replace) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
_map_stream_replace[stream_replace] = stream_id;
}
void RtpSelector::delStreamReplace(const string &stream_id) {
for (auto it = _map_stream_replace.begin(); it != _map_stream_replace.end(); ++it) {
if (it->second == stream_id) {
_map_stream_replace.erase(it);
break;
}
}
}
void RtpSelector::onManager() { void RtpSelector::onManager() {
List<RtpProcess::Ptr> clear_list; List<RtpProcess::Ptr> clear_list;
{ {
@ -96,6 +118,7 @@ void RtpSelector::onManager() {
} }
WarnL << "RtpProcess timeout:" << it->first; WarnL << "RtpProcess timeout:" << it->first;
clear_list.emplace_back(it->second->getProcess()); clear_list.emplace_back(it->second->getProcess());
delStreamReplace(it->first);
it = _map_rtp_process.erase(it); it = _map_rtp_process.erase(it);
} }
} }

4
src/Rtp/RtpSelector.h
View File

@ -70,14 +70,18 @@ public:
*/ */
void delProcess(const std::string &stream_id, const RtpProcess *ptr); void delProcess(const std::string &stream_id, const RtpProcess *ptr);
void addStreamReplace(const std::string &stream_id, const std::string &stream_replace);
private: private:
void onManager(); void onManager();
void createTimer(); void createTimer();
void delStreamReplace(const std::string &stream_id);
private: private:
toolkit::Timer::Ptr _timer; toolkit::Timer::Ptr _timer;
std::recursive_mutex _mtx_map; std::recursive_mutex _mtx_map;
std::unordered_map<std::string,RtpProcessHelper::Ptr> _map_rtp_process; std::unordered_map<std::string,RtpProcessHelper::Ptr> _map_rtp_process;
std::unordered_map<std::string,std::string> _map_stream_replace;
}; };
}//namespace mediakit }//namespace mediakit

9
src/Rtp/RtpSender.cpp
View File

@ -40,10 +40,11 @@ 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)); };
if (args.use_ps) { switch (args.type) {
_interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt); case MediaSourceEvent::SendRtpArgs::kRtpPS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, true); break;
} else { case MediaSourceEvent::SendRtpArgs::kRtpTS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, false); break;
_interface = std::make_shared<RtpCacheRaw>(lam, atoi(args.ssrc.data()), args.pt, args.only_audio); case MediaSourceEvent::SendRtpArgs::kRtpRAW: _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;
} }
} }

20
src/Rtp/RtpServer.cpp
View File

@ -42,12 +42,12 @@ public:
} }
} }
void setRtpServerInfo(uint16_t local_port,RtpServer::TcpMode mode,bool re_use_port,uint32_t ssrc, bool only_audio) { void setRtpServerInfo(uint16_t local_port, RtpServer::TcpMode mode, bool re_use_port, uint32_t ssrc, int only_track) {
_local_port = local_port; _local_port = local_port;
_tcp_mode = mode; _tcp_mode = mode;
_re_use_port = re_use_port; _re_use_port = re_use_port;
_ssrc = ssrc; _ssrc = ssrc;
_only_audio = only_audio; _only_track = only_track;
} }
void setOnDetach(function<void()> cb) { void setOnDetach(function<void()> cb) {
@ -61,7 +61,7 @@ public:
void onRecvRtp(const Socket::Ptr &sock, const Buffer::Ptr &buf, struct sockaddr *addr) { void onRecvRtp(const Socket::Ptr &sock, const Buffer::Ptr &buf, struct sockaddr *addr) {
if (!_process) { if (!_process) {
_process = RtpSelector::Instance().getProcess(_stream_id, true); _process = RtpSelector::Instance().getProcess(_stream_id, true);
_process->setOnlyAudio(_only_audio); _process->setOnlyTrack((RtpProcess::OnlyTrack)_only_track);
_process->setOnDetach(std::move(_on_detach)); _process->setOnDetach(std::move(_on_detach));
cancelDelayTask(); cancelDelayTask();
} }
@ -142,7 +142,7 @@ private:
private: private:
bool _re_use_port = false; bool _re_use_port = false;
bool _only_audio = false; int _only_track = 0;
uint16_t _local_port = 0; uint16_t _local_port = 0;
uint32_t _ssrc = 0; uint32_t _ssrc = 0;
RtpServer::TcpMode _tcp_mode = RtpServer::NONE; RtpServer::TcpMode _tcp_mode = RtpServer::NONE;
@ -156,7 +156,7 @@ private:
EventPoller::DelayTask::Ptr _delay_task; EventPoller::DelayTask::Ptr _delay_task;
}; };
void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc, bool only_audio, bool multiplex) { void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) {
//创建udp服务器 //创建udp服务器
Socket::Ptr rtp_socket = Socket::createSocket(nullptr, true); Socket::Ptr rtp_socket = Socket::createSocket(nullptr, true);
Socket::Ptr rtcp_socket = Socket::createSocket(nullptr, true); Socket::Ptr rtcp_socket = Socket::createSocket(nullptr, true);
@ -174,7 +174,8 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
} }
//设置udp socket读缓存 //设置udp socket读缓存
SockUtil::setRecvBuf(rtp_socket->rawFD(), 4 * 1024 * 1024); GET_CONFIG(int, udpRecvSocketBuffer, RtpProxy::kUdpRecvSocketBuffer);
SockUtil::setRecvBuf(rtp_socket->rawFD(), udpRecvSocketBuffer);
TcpServer::Ptr tcp_server; TcpServer::Ptr tcp_server;
_tcp_mode = tcp_mode; _tcp_mode = tcp_mode;
@ -183,7 +184,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
tcp_server = std::make_shared<TcpServer>(rtp_socket->getPoller()); tcp_server = std::make_shared<TcpServer>(rtp_socket->getPoller());
(*tcp_server)[RtpSession::kStreamID] = stream_id; (*tcp_server)[RtpSession::kStreamID] = stream_id;
(*tcp_server)[RtpSession::kSSRC] = ssrc; (*tcp_server)[RtpSession::kSSRC] = ssrc;
(*tcp_server)[RtpSession::kOnlyAudio] = only_audio; (*tcp_server)[RtpSession::kOnlyTrack] = only_track;
if (tcp_mode == PASSIVE) { if (tcp_mode == PASSIVE) {
tcp_server->start<RtpSession>(local_port, local_ip); tcp_server->start<RtpSession>(local_port, local_ip);
} else if (stream_id.empty()) { } else if (stream_id.empty()) {
@ -200,7 +201,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
//指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流) //指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), stream_id); helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), stream_id);
helper->startRtcp(); helper->startRtcp();
helper->setRtpServerInfo(local_port, tcp_mode, re_use_port, ssrc, only_audio); helper->setRtpServerInfo(local_port, tcp_mode, re_use_port, ssrc, only_track);
bool bind_peer_addr = false; bool bind_peer_addr = false;
auto ssrc_ptr = std::make_shared<uint32_t>(ssrc); auto ssrc_ptr = std::make_shared<uint32_t>(ssrc);
_ssrc = ssrc_ptr; _ssrc = ssrc_ptr;
@ -222,7 +223,8 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
} else { } else {
//单端口多线程接收多个流，根据ssrc区分流 //单端口多线程接收多个流，根据ssrc区分流
udp_server = std::make_shared<UdpServer>(rtp_socket->getPoller()); udp_server = std::make_shared<UdpServer>(rtp_socket->getPoller());
(*udp_server)[RtpSession::kOnlyAudio] = only_audio; (*udp_server)[RtpSession::kOnlyTrack] = only_track;
(*udp_server)[RtpSession::kUdpRecvBuffer] = udpRecvSocketBuffer;
udp_server->start<RtpSession>(local_port, local_ip); udp_server->start<RtpSession>(local_port, local_ip);
rtp_socket = nullptr; rtp_socket = nullptr;
} }

4
src/Rtp/RtpServer.h
View File

@ -44,7 +44,7 @@ public:
* @param multiplex 多路复用 * @param multiplex 多路复用
*/ */
void start(uint16_t local_port, const std::string &stream_id = "", TcpMode tcp_mode = PASSIVE, void start(uint16_t local_port, const std::string &stream_id = "", TcpMode tcp_mode = PASSIVE,
const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0, bool only_audio = false, bool multiplex = false); const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0, int only_track = 0, bool multiplex = false);
/** /**
* 连接到tcp服务(tcp主动模式) * 连接到tcp服务(tcp主动模式)
@ -81,7 +81,7 @@ protected:
std::shared_ptr<RtcpHelper> _rtcp_helper; std::shared_ptr<RtcpHelper> _rtcp_helper;
std::function<void()> _on_cleanup; std::function<void()> _on_cleanup;
bool _only_audio = false; int _only_track = 0;
//用于tcp主动模式 //用于tcp主动模式
TcpMode _tcp_mode = NONE; TcpMode _tcp_mode = NONE;
}; };

18
src/Rtp/RtpSession.cpp
View File

@ -23,7 +23,8 @@ namespace mediakit{
const string RtpSession::kStreamID = "stream_id"; const string RtpSession::kStreamID = "stream_id";
const string RtpSession::kSSRC = "ssrc"; const string RtpSession::kSSRC = "ssrc";
const string RtpSession::kOnlyAudio = "only_audio"; const string RtpSession::kOnlyTrack = "only_track";
const string RtpSession::kUdpRecvBuffer = "udp_recv_socket_buffer";
void RtpSession::attachServer(const Server &server) { void RtpSession::attachServer(const Server &server) {
setParams(const_cast<Server &>(server)); setParams(const_cast<Server &>(server));
@ -32,7 +33,13 @@ void RtpSession::attachServer(const Server &server) {
void RtpSession::setParams(mINI &ini) { void RtpSession::setParams(mINI &ini) {
_stream_id = ini[kStreamID]; _stream_id = ini[kStreamID];
_ssrc = ini[kSSRC]; _ssrc = ini[kSSRC];
_only_audio = ini[kOnlyAudio]; _only_track = ini[kOnlyTrack];
int udp_socket_buffer = ini[kUdpRecvBuffer];
if (_is_udp) {
// 设置udp socket读缓存
SockUtil::setRecvBuf(getSock()->rawFD(),
(udp_socket_buffer > 0) ? udp_socket_buffer : (4 * 1024 * 1024));
}
} }
RtpSession::RtpSession(const Socket::Ptr &sock) RtpSession::RtpSession(const Socket::Ptr &sock)
@ -40,10 +47,6 @@ RtpSession::RtpSession(const Socket::Ptr &sock)
socklen_t addr_len = sizeof(_addr); socklen_t addr_len = sizeof(_addr);
getpeername(sock->rawFD(), (struct sockaddr *)&_addr, &addr_len); getpeername(sock->rawFD(), (struct sockaddr *)&_addr, &addr_len);
_is_udp = sock->sockType() == SockNum::Sock_UDP; _is_udp = sock->sockType() == SockNum::Sock_UDP;
if (_is_udp) {
// 设置udp socket读缓存
SockUtil::setRecvBuf(getSock()->rawFD(), 4 * 1024 * 1024);
}
} }
RtpSession::~RtpSession() = default; RtpSession::~RtpSession() = default;
@ -60,6 +63,7 @@ void RtpSession::onError(const SockException &err) {
WarnP(this) << _stream_id << " " << err; WarnP(this) << _stream_id << " " << err;
if (_process) { if (_process) {
RtpSelector::Instance().delProcess(_stream_id, _process.get()); RtpSelector::Instance().delProcess(_stream_id, _process.get());
_process = nullptr;
} }
} }
@ -121,7 +125,7 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
_delay_close = true; _delay_close = true;
return; return;
} }
_process->setOnlyAudio(_only_audio); _process->setOnlyTrack((RtpProcess::OnlyTrack)_only_track);
_process->setDelegate(static_pointer_cast<RtpSession>(shared_from_this())); _process->setDelegate(static_pointer_cast<RtpSession>(shared_from_this()));
} }
try { try {

5
src/Rtp/RtpSession.h
View File

@ -24,7 +24,8 @@ class RtpSession : public toolkit::Session, public RtpSplitter, public MediaSour
public: public:
static const std::string kStreamID; static const std::string kStreamID;
static const std::string kSSRC; static const std::string kSSRC;
static const std::string kOnlyAudio; static const std::string kOnlyTrack;
static const std::string kUdpRecvBuffer;
RtpSession(const toolkit::Socket::Ptr &sock); RtpSession(const toolkit::Socket::Ptr &sock);
~RtpSession() override; ~RtpSession() override;
@ -51,7 +52,7 @@ private:
bool _is_udp = false; bool _is_udp = false;
bool _search_rtp = false; bool _search_rtp = false;
bool _search_rtp_finished = false; bool _search_rtp_finished = false;
bool _only_audio = false; int _only_track = 0;
uint32_t _ssrc = 0; uint32_t _ssrc = 0;
toolkit::Ticker _ticker; toolkit::Ticker _ticker;
std::string _stream_id; std::string _stream_id;

11
src/Rtsp/RtpCodec.h
View File

@ -93,6 +93,17 @@ public:
RtpInfo &getRtpInfo() { return *_rtp_info; } RtpInfo &getRtpInfo() { return *_rtp_info; }
enum {
RTP_ENCODER_PKT_DUR_MS = 1 // 主要应用于g711 rtp 打包器每个包的时间长度，option_value 为int*, option_len 为4
};
/**
* @brief 设置rtp打包器与解包器的相关参数，主要应用与g711 rtp 打包器，使用方法类似setsockopt
*
* @param opt 设置的选项
* @param param 设置的参数
*/
virtual void setOpt(int opt, const toolkit::Any &param) {};
private: private:
std::unique_ptr<RtpInfo> _rtp_info; std::unique_ptr<RtpInfo> _rtp_info;
}; };

78
srt/SrtSession.cpp
View File

@ -16,41 +16,17 @@ SrtSession::SrtSession(const Socket::Ptr &sock)
// TraceL<<"after addr len "<<addr_len<<" family "<<_peer_addr.ss_family; // TraceL<<"after addr len "<<addr_len<<" family "<<_peer_addr.ss_family;
} }
EventPoller::Ptr SrtSession::queryPoller(const Buffer::Ptr &buffer) {
uint8_t *data = (uint8_t *)buffer->data();
size_t size = buffer->size();
if (DataPacket::isDataPacket(data, size)) {
uint32_t socket_id = DataPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
return trans ? trans->getPoller() : nullptr;
}
if (HandshakePacket::isHandshakePacket(data, size)) {
auto type = HandshakePacket::getHandshakeType(data, size);
if (type == HandshakePacket::HS_TYPE_INDUCTION) {
// 握手第一阶段
return nullptr;
} else if (type == HandshakePacket::HS_TYPE_CONCLUSION) {
// 握手第二阶段
uint32_t sync_cookie = HandshakePacket::getSynCookie(data, size);
auto trans = SrtTransportManager::Instance().getHandshakeItem(std::to_string(sync_cookie));
return trans ? trans->getPoller() : nullptr;
} else {
WarnL << " not reach there";
}
} else {
uint32_t socket_id = ControlPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
return trans ? trans->getPoller() : nullptr;
}
return nullptr;
}
void SrtSession::attachServer(const toolkit::Server &server) { void SrtSession::attachServer(const toolkit::Server &server) {
SockUtil::setRecvBuf(getSock()->rawFD(), 1024 * 1024); SockUtil::setRecvBuf(getSock()->rawFD(), 1024 * 1024);
} }
extern SrtTransport::Ptr querySrtTransport(uint8_t *data, size_t size, const EventPoller::Ptr& poller);
EventPoller::Ptr SrtSession::queryPoller(const Buffer::Ptr &buffer) {
auto transport = querySrtTransport((uint8_t *)buffer->data(), buffer->size(), nullptr);
return transport ? transport->getPoller() : nullptr;
}
void SrtSession::onRecv(const Buffer::Ptr &buffer) { void SrtSession::onRecv(const Buffer::Ptr &buffer) {
uint8_t *data = (uint8_t *)buffer->data(); uint8_t *data = (uint8_t *)buffer->data();
size_t size = buffer->size(); size_t size = buffer->size();
@ -58,45 +34,7 @@ void SrtSession::onRecv(const Buffer::Ptr &buffer) {
if (_find_transport) { if (_find_transport) {
//只允许寻找一次transport //只允许寻找一次transport
_find_transport = false; _find_transport = false;
_transport = querySrtTransport(data, size, getPoller());
if (DataPacket::isDataPacket(data, size)) {
uint32_t socket_id = DataPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
if (trans) {
_transport = std::move(trans);
} else {
WarnL << " data packet not find transport ";
}
}
if (HandshakePacket::isHandshakePacket(data, size)) {
auto type = HandshakePacket::getHandshakeType(data, size);
if (type == HandshakePacket::HS_TYPE_INDUCTION) {
// 握手第一阶段
_transport = std::make_shared<SrtTransportImp>(getPoller());
} else if (type == HandshakePacket::HS_TYPE_CONCLUSION) {
// 握手第二阶段
uint32_t sync_cookie = HandshakePacket::getSynCookie(data, size);
auto trans = SrtTransportManager::Instance().getHandshakeItem(std::to_string(sync_cookie));
if (trans) {
_transport = std::move(trans);
} else {
WarnL << " hanshake packet not find transport ";
}
} else {
WarnL << " not reach there";
}
} else {
uint32_t socket_id = ControlPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
if (trans) {
_transport = std::move(trans);
} else {
WarnL << " not find transport";
}
}
if (_transport) { if (_transport) {
_transport->setSession(static_pointer_cast<Session>(shared_from_this())); _transport->setSession(static_pointer_cast<Session>(shared_from_this()));
} }

34
srt/SrtTransport.cpp
View File

@ -61,7 +61,7 @@ void SrtTransport::switchToOtherTransport(uint8_t *buf, int len, uint32_t socket
BufferRaw::Ptr tmp = BufferRaw::create(); BufferRaw::Ptr tmp = BufferRaw::create();
struct sockaddr_storage tmp_addr = *addr; struct sockaddr_storage tmp_addr = *addr;
tmp->assign((char *)buf, len); tmp->assign((char *)buf, len);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socketid)); auto trans = SrtTransportManager::Instance().getItem(socketid);
if (trans) { if (trans) {
trans->getPoller()->async([tmp, tmp_addr, trans] { trans->getPoller()->async([tmp, tmp_addr, trans] {
trans->inputSockData((uint8_t *)tmp->data(), tmp->size(), (struct sockaddr_storage *)&tmp_addr); trans->inputSockData((uint8_t *)tmp->data(), tmp->size(), (struct sockaddr_storage *)&tmp_addr);
@ -700,30 +700,30 @@ void SrtTransport::sendPacket(Buffer::Ptr pkt, bool flush) {
} }
} }
std::string SrtTransport::getIdentifier() { std::string SrtTransport::getIdentifier() const {
return _selected_session ? _selected_session->getIdentifier() : ""; return _selected_session ? _selected_session->getIdentifier() : "";
} }
void SrtTransport::registerSelfHandshake() { void SrtTransport::registerSelfHandshake() {
SrtTransportManager::Instance().addHandshakeItem(std::to_string(_sync_cookie), shared_from_this()); SrtTransportManager::Instance().addHandshakeItem(_sync_cookie, shared_from_this());
} }
void SrtTransport::unregisterSelfHandshake() { void SrtTransport::unregisterSelfHandshake() {
if (_sync_cookie == 0) { if (_sync_cookie == 0) {
return; return;
} }
SrtTransportManager::Instance().removeHandshakeItem(std::to_string(_sync_cookie)); SrtTransportManager::Instance().removeHandshakeItem(_sync_cookie);
} }
void SrtTransport::registerSelf() { void SrtTransport::registerSelf() {
if (_socket_id == 0) { if (_socket_id == 0) {
return; return;
} }
SrtTransportManager::Instance().addItem(std::to_string(_socket_id), shared_from_this()); SrtTransportManager::Instance().addItem(_socket_id, shared_from_this());
} }
void SrtTransport::unregisterSelf() { void SrtTransport::unregisterSelf() {
SrtTransportManager::Instance().removeItem(std::to_string(_socket_id)); SrtTransportManager::Instance().removeItem(_socket_id);
} }
void SrtTransport::onShutdown(const SockException &ex) { void SrtTransport::onShutdown(const SockException &ex) {
@ -739,7 +739,7 @@ void SrtTransport::onShutdown(const SockException &ex) {
} }
} }
size_t SrtTransport::getPayloadSize() { size_t SrtTransport::getPayloadSize() const {
size_t ret = (_mtu - 28 - 16) / 188 * 188; size_t ret = (_mtu - 28 - 16) / 188 * 188;
return ret; return ret;
} }
@ -792,15 +792,13 @@ SrtTransportManager &SrtTransportManager::Instance() {
return s_instance; return s_instance;
} }
void SrtTransportManager::addItem(const std::string &key, const SrtTransport::Ptr &ptr) { void SrtTransportManager::addItem(const uint32_t key, const SrtTransport::Ptr &ptr) {
std::lock_guard<std::mutex> lck(_mtx); std::lock_guard<std::mutex> lck(_mtx);
_map[key] = ptr; _map[key] = ptr;
} }
SrtTransport::Ptr SrtTransportManager::getItem(const std::string &key) { SrtTransport::Ptr SrtTransportManager::getItem(const uint32_t key) {
if (key.empty()) { assert(key > 0);
return nullptr;
}
std::lock_guard<std::mutex> lck(_mtx); std::lock_guard<std::mutex> lck(_mtx);
auto it = _map.find(key); auto it = _map.find(key);
if (it == _map.end()) { if (it == _map.end()) {
@ -809,25 +807,23 @@ SrtTransport::Ptr SrtTransportManager::getItem(const std::string &key) {
return it->second.lock(); return it->second.lock();
} }
void SrtTransportManager::removeItem(const std::string &key) { void SrtTransportManager::removeItem(const uint32_t key) {
std::lock_guard<std::mutex> lck(_mtx); std::lock_guard<std::mutex> lck(_mtx);
_map.erase(key); _map.erase(key);
} }
void SrtTransportManager::addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr) { void SrtTransportManager::addHandshakeItem(const uint32_t key, const SrtTransport::Ptr &ptr) {
std::lock_guard<std::mutex> lck(_handshake_mtx); std::lock_guard<std::mutex> lck(_handshake_mtx);
_handshake_map[key] = ptr; _handshake_map[key] = ptr;
} }
void SrtTransportManager::removeHandshakeItem(const std::string &key) { void SrtTransportManager::removeHandshakeItem(const uint32_t key) {
std::lock_guard<std::mutex> lck(_handshake_mtx); std::lock_guard<std::mutex> lck(_handshake_mtx);
_handshake_map.erase(key); _handshake_map.erase(key);
} }
SrtTransport::Ptr SrtTransportManager::getHandshakeItem(const std::string &key) { SrtTransport::Ptr SrtTransportManager::getHandshakeItem(const uint32_t key) {
if (key.empty()) { assert(key > 0);
return nullptr;
}
std::lock_guard<std::mutex> lck(_handshake_mtx); std::lock_guard<std::mutex> lck(_handshake_mtx);
auto it = _handshake_map.find(key); auto it = _handshake_map.find(key);
if (it == _handshake_map.end()) { if (it == _handshake_map.end()) {

20
srt/SrtTransport.hpp
View File

@ -45,7 +45,7 @@ public:
virtual void inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr); virtual void inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr);
virtual void onSendTSData(const Buffer::Ptr &buffer, bool flush); virtual void onSendTSData(const Buffer::Ptr &buffer, bool flush);
std::string getIdentifier(); std::string getIdentifier() const;
void unregisterSelf(); void unregisterSelf();
void unregisterSelfHandshake(); void unregisterSelfHandshake();
@ -89,7 +89,7 @@ private:
void sendShutDown(); void sendShutDown();
void sendMsgDropReq(uint32_t first, uint32_t last); void sendMsgDropReq(uint32_t first, uint32_t last);
size_t getPayloadSize(); size_t getPayloadSize() const;
void createTimerForCheckAlive(); void createTimerForCheckAlive();
@ -164,23 +164,23 @@ private:
class SrtTransportManager { class SrtTransportManager {
public: public:
static SrtTransportManager &Instance(); static SrtTransportManager &Instance();
SrtTransport::Ptr getItem(const std::string &key); SrtTransport::Ptr getItem(const uint32_t key);
void addItem(const std::string &key, const SrtTransport::Ptr &ptr); void addItem(const uint32_t key, const SrtTransport::Ptr &ptr);
void removeItem(const std::string &key); void removeItem(const uint32_t key);
void addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr); void addHandshakeItem(const uint32_t key, const SrtTransport::Ptr &ptr);
void removeHandshakeItem(const std::string &key); void removeHandshakeItem(const uint32_t key);
SrtTransport::Ptr getHandshakeItem(const std::string &key); SrtTransport::Ptr getHandshakeItem(const uint32_t key);
private: private:
SrtTransportManager() = default; SrtTransportManager() = default;
private: private:
std::mutex _mtx; std::mutex _mtx;
std::unordered_map<std::string, std::weak_ptr<SrtTransport>> _map; std::unordered_map<uint32_t , std::weak_ptr<SrtTransport>> _map;
std::mutex _handshake_mtx; std::mutex _handshake_mtx;
std::unordered_map<std::string, std::weak_ptr<SrtTransport>> _handshake_map; std::unordered_map<uint32_t, std::weak_ptr<SrtTransport>> _handshake_map;
}; };
} // namespace SRT } // namespace SRT

26
srt/SrtTransportImp.cpp
View File

@ -24,6 +24,32 @@ SrtTransportImp::~SrtTransportImp() {
} }
} }
SrtTransport::Ptr querySrtTransport(uint8_t *data, size_t size, const EventPoller::Ptr& poller) {
if (DataPacket::isDataPacket(data, size)) {
uint32_t socket_id = DataPacket::getSocketID(data, size);
return SrtTransportManager::Instance().getItem(socket_id);
}
if (HandshakePacket::isHandshakePacket(data, size)) {
auto type = HandshakePacket::getHandshakeType(data, size);
if (type == HandshakePacket::HS_TYPE_INDUCTION) {
// 握手第一阶段
return poller ? std::make_shared<SrtTransportImp>(poller) : nullptr;
}
if (type == HandshakePacket::HS_TYPE_CONCLUSION) {
// 握手第二阶段
uint32_t sync_cookie = HandshakePacket::getSynCookie(data, size);
return SrtTransportManager::Instance().getHandshakeItem(sync_cookie);
}
}
uint32_t socket_id = ControlPacket::getSocketID(data, size);
return SrtTransportManager::Instance().getItem(socket_id);
}
void SrtTransportImp::onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) { void SrtTransportImp::onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) {
SrtTransport::onHandShakeFinished(streamid,addr); SrtTransport::onHandShakeFinished(streamid,addr);
// TODO parse stream id like this zlmediakit.com/live/test?token=1213444&type=push // TODO parse stream id like this zlmediakit.com/live/test?token=1213444&type=push

75
webrtc/WebRtcTransport.cpp
View File

@ -55,6 +55,9 @@ const string kStartBitrate = RTC_FIELD "start_bitrate";
const string kMaxBitrate = RTC_FIELD "max_bitrate"; const string kMaxBitrate = RTC_FIELD "max_bitrate";
const string kMinBitrate = RTC_FIELD "min_bitrate"; const string kMinBitrate = RTC_FIELD "min_bitrate";
// 数据通道设置
const string kDataChannelEcho = RTC_FIELD "datachannel_echo";
static onceToken token([]() { static onceToken token([]() {
mINI::Instance()[kTimeOutSec] = 15; mINI::Instance()[kTimeOutSec] = 15;
mINI::Instance()[kExternIP] = ""; mINI::Instance()[kExternIP] = "";
@ -65,6 +68,8 @@ static onceToken token([]() {
mINI::Instance()[kStartBitrate] = 0; mINI::Instance()[kStartBitrate] = 0;
mINI::Instance()[kMaxBitrate] = 0; mINI::Instance()[kMaxBitrate] = 0;
mINI::Instance()[kMinBitrate] = 0; mINI::Instance()[kMinBitrate] = 0;
mINI::Instance()[kDataChannelEcho] = true;
}); });
} // namespace RTC } // namespace RTC
@ -201,9 +206,26 @@ void WebRtcTransport::OnDtlsTransportConnected(
onStartWebRTC(); onStartWebRTC();
} }
#pragma pack(push, 1)
struct DtlsHeader {
uint8_t content_type;
uint16_t dtls_version;
uint16_t epoch;
uint8_t seq[6];
uint16_t length;
uint8_t payload[1];
};
#pragma pack(pop)
void WebRtcTransport::OnDtlsTransportSendData( void WebRtcTransport::OnDtlsTransportSendData(
const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) { const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) {
sendSockData((char *)data, len, nullptr); size_t offset = 0;
while(offset < len) {
auto *header = reinterpret_cast<const DtlsHeader *>(data + offset);
auto length = ntohs(header->length) + offsetof(DtlsHeader, payload);
sendSockData((char *)data + offset, length, nullptr);
offset += length;
}
} }
void WebRtcTransport::OnDtlsTransportConnecting(const RTC::DtlsTransport *dtlsTransport) { void WebRtcTransport::OnDtlsTransportConnecting(const RTC::DtlsTransport *dtlsTransport) {
@ -233,22 +255,47 @@ void WebRtcTransport::OnDtlsTransportApplicationDataReceived(
#ifdef ENABLE_SCTP #ifdef ENABLE_SCTP
void WebRtcTransport::OnSctpAssociationConnecting(RTC::SctpAssociation *sctpAssociation) { void WebRtcTransport::OnSctpAssociationConnecting(RTC::SctpAssociation *sctpAssociation) {
TraceL << getIdentifier(); TraceL << getIdentifier();
try {
NOTICE_EMIT(BroadcastRtcSctpConnectArgs, Broadcast::kBroadcastRtcSctpConnecting, *this);
} catch (std::exception &ex) {
WarnL << "Exception occurred: " << ex.what();
}
} }
void WebRtcTransport::OnSctpAssociationConnected(RTC::SctpAssociation *sctpAssociation) { void WebRtcTransport::OnSctpAssociationConnected(RTC::SctpAssociation *sctpAssociation) {
InfoL << getIdentifier(); InfoL << getIdentifier();
try {
NOTICE_EMIT(BroadcastRtcSctpConnectArgs, Broadcast::kBroadcastRtcSctpConnected, *this);
} catch (std::exception &ex) {
WarnL << "Exception occurred: " << ex.what();
}
} }
void WebRtcTransport::OnSctpAssociationFailed(RTC::SctpAssociation *sctpAssociation) { void WebRtcTransport::OnSctpAssociationFailed(RTC::SctpAssociation *sctpAssociation) {
WarnL << getIdentifier(); WarnL << getIdentifier();
try {
NOTICE_EMIT(BroadcastRtcSctpConnectArgs, Broadcast::kBroadcastRtcSctpFailed, *this);
} catch (std::exception &ex) {
WarnL << "Exception occurred: " << ex.what();
}
} }
void WebRtcTransport::OnSctpAssociationClosed(RTC::SctpAssociation *sctpAssociation) { void WebRtcTransport::OnSctpAssociationClosed(RTC::SctpAssociation *sctpAssociation) {
InfoL << getIdentifier(); InfoL << getIdentifier();
try {
NOTICE_EMIT(BroadcastRtcSctpConnectArgs, Broadcast::kBroadcastRtcSctpClosed, *this);
} catch (std::exception &ex) {
WarnL << "Exception occurred: " << ex.what();
}
} }
void WebRtcTransport::OnSctpAssociationSendData( void WebRtcTransport::OnSctpAssociationSendData(
RTC::SctpAssociation *sctpAssociation, const uint8_t *data, size_t len) { RTC::SctpAssociation *sctpAssociation, const uint8_t *data, size_t len) {
try {
NOTICE_EMIT(BroadcastRtcSctpSendArgs, Broadcast::kBroadcastRtcSctpSend, *this, data, len);
} catch (std::exception &ex) {
WarnL << "Exception occurred: " << ex.what();
}
_dtls_transport->SendApplicationData(data, len); _dtls_transport->SendApplicationData(data, len);
} }
@ -257,8 +304,18 @@ void WebRtcTransport::OnSctpAssociationMessageReceived(
InfoL << getIdentifier() << " " << streamId << " " << ppid << " " << len << " " << string((char *)msg, len); InfoL << getIdentifier() << " " << streamId << " " << ppid << " " << len << " " << string((char *)msg, len);
RTC::SctpStreamParameters params; RTC::SctpStreamParameters params;
params.streamId = streamId; params.streamId = streamId;
// 回显数据
_sctp->SendSctpMessage(params, ppid, msg, len); GET_CONFIG(bool, datachannel_echo, Rtc::kDataChannelEcho);
if (datachannel_echo) {
// 回显数据
_sctp->SendSctpMessage(params, ppid, msg, len);
}
try {
NOTICE_EMIT(BroadcastRtcSctpReceivedArgs, Broadcast::kBroadcastRtcSctpReceived, *this, streamId, ppid, msg, len);
} catch (std::exception &ex) {
WarnL << "Exception occurred: " << ex.what();
}
} }
#endif #endif
@ -618,7 +675,7 @@ void WebRtcTransportImp::onCheckAnswer(RtcSession &sdp) {
}); });
for (auto &m : sdp.media) { for (auto &m : sdp.media) {
m.addr.reset(); m.addr.reset();
m.addr.address = extern_ips.empty() ? SockUtil::get_local_ip() : extern_ips[0]; m.addr.address = extern_ips.empty() ? _localIp.empty() ? SockUtil::get_local_ip() : _localIp : extern_ips[0];
m.rtcp_addr.reset(); m.rtcp_addr.reset();
m.rtcp_addr.address = m.addr.address; m.rtcp_addr.address = m.addr.address;
@ -713,7 +770,7 @@ void WebRtcTransportImp::onRtcConfigure(RtcConfigure &configure) const {
return ret; return ret;
}); });
if (extern_ips.empty()) { if (extern_ips.empty()) {
std::string local_ip = SockUtil::get_local_ip(); std::string local_ip = _localIp.empty() ? SockUtil::get_local_ip() : _localIp;
if (local_udp_port) { configure.addCandidate(*makeIceCandidate(local_ip, local_udp_port, 120, "udp")); } if (local_udp_port) { configure.addCandidate(*makeIceCandidate(local_ip, local_udp_port, 120, "udp")); }
if (local_tcp_port) { configure.addCandidate(*makeIceCandidate(local_ip, local_tcp_port, _preferred_tcp ? 125 : 115, "tcp")); } if (local_tcp_port) { configure.addCandidate(*makeIceCandidate(local_ip, local_tcp_port, _preferred_tcp ? 125 : 115, "tcp")); }
} else { } else {
@ -731,6 +788,10 @@ void WebRtcTransportImp::setIceCandidate(vector<SdpAttrCandidate> cands) {
_cands = std::move(cands); _cands = std::move(cands);
} }
void WebRtcTransportImp::setLocalIp(const std::string &localIp) {
_localIp = localIp;
}
/////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////
class RtpChannel : public RtpTrackImp, public std::enable_shared_from_this<RtpChannel> { class RtpChannel : public RtpTrackImp, public std::enable_shared_from_this<RtpChannel> {
@ -1222,6 +1283,10 @@ std::string exchangeSdp(const WebRtcInterface &exchanger, const std::string& off
return const_cast<WebRtcInterface &>(exchanger).getAnswerSdp(offer); return const_cast<WebRtcInterface &>(exchanger).getAnswerSdp(offer);
} }
void setLocalIp(const WebRtcInterface& exchanger, const std::string& localIp) {
return const_cast<WebRtcInterface &>(exchanger).setLocalIp(localIp);
}
void WebRtcPluginManager::setListener(Listener cb) { void WebRtcPluginManager::setListener(Listener cb) {
lock_guard<mutex> lck(_mtx_creator); lock_guard<mutex> lck(_mtx_creator);
_listener = std::move(cb); _listener = std::move(cb);

6
webrtc/WebRtcTransport.h
View File

@ -42,10 +42,13 @@ public:
virtual const std::string& getIdentifier() const = 0; virtual const std::string& getIdentifier() const = 0;
virtual const std::string& deleteRandStr() const { static std::string s_null; return s_null; } virtual const std::string& deleteRandStr() const { static std::string s_null; return s_null; }
virtual void setIceCandidate(std::vector<SdpAttrCandidate> cands) {} virtual void setIceCandidate(std::vector<SdpAttrCandidate> cands) {}
virtual void setLocalIp(const std::string &localIp) {}
}; };
std::string exchangeSdp(const WebRtcInterface &exchanger, const std::string& offer); std::string exchangeSdp(const WebRtcInterface &exchanger, const std::string& offer);
void setLocalIp(const WebRtcInterface &exchanger, const std::string &localIp);
class WebRtcException : public WebRtcInterface { class WebRtcException : public WebRtcInterface {
public: public:
WebRtcException(const SockException &ex) : _ex(ex) {}; WebRtcException(const SockException &ex) : _ex(ex) {};
@ -253,6 +256,7 @@ public:
void removeTuple(RTC::TransportTuple* tuple); void removeTuple(RTC::TransportTuple* tuple);
void safeShutdown(const SockException &ex); void safeShutdown(const SockException &ex);
void setLocalIp(const std::string &localIp) override;
protected: protected:
void OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) override; void OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) override;
WebRtcTransportImp(const EventPoller::Ptr &poller,bool preferred_tcp = false); WebRtcTransportImp(const EventPoller::Ptr &poller,bool preferred_tcp = false);
@ -306,6 +310,8 @@ private:
//根据接收rtp的pt获取相关信息 //根据接收rtp的pt获取相关信息
std::unordered_map<uint8_t/*pt*/, std::unique_ptr<WrappedMediaTrack>> _pt_to_track; std::unordered_map<uint8_t/*pt*/, std::unique_ptr<WrappedMediaTrack>> _pt_to_track;
std::vector<SdpAttrCandidate> _cands; std::vector<SdpAttrCandidate> _cands;
//源访问的hostip
std::string _localIp;
}; };
class WebRtcTransportManager { class WebRtcTransportManager {