Mirror/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2026-06-29 06:12:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

AI automatically translates all comments in the code into English (#3917)

Browse Source
This commit is contained in:
alex
2024-09-19 14:53:50 +08:00
committed by GitHub
parent 046de691cb
commit 4152dcd409
279 changed files with 10602 additions and 3038 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/Rtp/Decoder.cpp
View File

@@ -88,12 +88,14 @@ void DecoderImp::onStream(int stream, int codecid, const void *extra, size_t byt
if (_finished) {
return;
}
// G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了
// G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了 [AUTO-TRANSLATED:851813f7]
// G711 traditionally only supports the 8000/1/16 specification. FFmpeg seems to have extended it, but we'll ignore that here.
auto track = Factory::getTrackByCodecId(getCodecByMpegId(codecid), 8000, 1, 16);
if (track) {
onTrack(stream, std::move(track));
}
// 防止未获取视频track提前complete导致忽略后续视频的问题，用于兼容一些不太规范的ps流
// 防止未获取视频track提前complete导致忽略后续视频的问题，用于兼容一些不太规范的ps流 [AUTO-TRANSLATED:d6b349b5]
// Prevent the problem of ignoring subsequent video due to premature completion of the video track before it is obtained. This is used to be compatible with some non-standard PS streams.
if (finish && _have_video) {
_finished = true;
_sink->addTrackCompleted();

42
src/Rtp/GB28181Process.cpp
View File

@@ -23,7 +23,8 @@ using namespace toolkit;
namespace mediakit {
// 判断是否为ts负载
// 判断是否为ts负载 [AUTO-TRANSLATED:77d1aa3c]
// Determine if it is a ts payload
static inline bool checkTS(const uint8_t *packet, size_t bytes) {
return bytes % TS_PACKET_SIZE == 0 && packet[0] == TS_SYNC_BYTE;
}
@@ -36,7 +37,8 @@ public:
_sample_rate = sample_rate;
setOnSorted(std::move(cb));
setBeforeSorted(std::move(cb_before));
// GB28181推流不支持ntp时间戳
// GB28181推流不支持ntp时间戳 [AUTO-TRANSLATED:f661f052]
// GB28181 streaming does not support ntp timestamps
setNtpStamp(0, 0);
}
@@ -77,7 +79,8 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
auto &ref = _rtp_receiver[pt];
if (!ref) {
if (_rtp_receiver.size() > 2) {
// 防止pt类型太多导致内存溢出
// 防止pt类型太多导致内存溢出 [AUTO-TRANSLATED:7695e49b]
// Prevent too many pt types from causing memory overflow
WarnL << "Rtp payload type more than 2 types: " << _rtp_receiver.size();
}
switch (pt) {
@@ -104,7 +107,8 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
}
default: {
if (pt == opus_pt) {
// opus负载
// opus负载 [AUTO-TRANSLATED:defa6a8d]
// opus payload
ref = std::make_shared<RtpReceiverImp>(48000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
auto track = Factory::getTrackByCodecId(CodecOpus);
CHECK(track);
@@ -112,7 +116,8 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
_interface->addTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByCodecId(track->getCodecId());
} else if (pt == h265_pt) {
// H265负载
// H265负载 [AUTO-TRANSLATED:61fbcf7f]
// H265 payload
ref = std::make_shared<RtpReceiverImp>(90000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
auto track = Factory::getTrackByCodecId(CodecH265);
CHECK(track);
@@ -120,7 +125,8 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
_interface->addTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByCodecId(track->getCodecId());
} else if (pt == h264_pt) {
// H264负载
// H264负载 [AUTO-TRANSLATED:6f3fbb0d]
// H264 payload
ref = std::make_shared<RtpReceiverImp>(90000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
auto track = Factory::getTrackByCodecId(CodecH264);
CHECK(track);
@@ -132,9 +138,11 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
WarnL << "Unknown rtp payload type(" << (int)pt << "), decode it as mpeg-ps or mpeg-ts";
}
ref = std::make_shared<RtpReceiverImp>(90000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
// ts或ps负载
// ts或ps负载 [AUTO-TRANSLATED:3ca31480]
// ts or ps payload
_rtp_decoder[pt] = std::make_shared<CommonRtpDecoder>(CodecInvalid, 32 * 1024);
// 设置dump目录
// 设置dump目录 [AUTO-TRANSLATED:23c88ace]
// Set dump directory
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);
if (!dump_dir.empty()) {
auto save_path = File::absolutePath(_media_info.stream + ".mpeg", dump_dir);
@@ -148,7 +156,8 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
break;
}
}
// 设置frame回调
// 设置frame回调 [AUTO-TRANSLATED:dec7590f]
// Set frame callback
_rtp_decoder[pt]->addDelegate([this, pt](const Frame::Ptr &frame) {
frame->setIndex(pt);
onRtpDecode(frame);
@@ -161,24 +170,29 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
void GB28181Process::onRtpDecode(const Frame::Ptr &frame) {
if (frame->getCodecId() != CodecInvalid) {
// 这里不是ps或ts
// 这里不是ps或ts [AUTO-TRANSLATED:6f79ac69]
// This is not ps or ts
_interface->inputFrame(frame);
return;
}
// 这是TS或PS
// 这是TS或PS [AUTO-TRANSLATED:55782860]
// This is TS or PS
if (_save_file_ps) {
fwrite(frame->data(), frame->size(), 1, _save_file_ps.get());
}
if (!_decoder) {
// 创建解码器
// 创建解码器 [AUTO-TRANSLATED:0cc03d90]
// Create decoder
if (checkTS((uint8_t *)frame->data(), frame->size())) {
// 猜测是ts负载
// 猜测是ts负载 [AUTO-TRANSLATED:c2be3a47]
// Guess it is a ts payload
InfoL << _media_info.stream << " judged to be TS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, _interface);
} else {
// 猜测是ps负载
// 猜测是ps负载 [AUTO-TRANSLATED:b7c0ff45]
// Guess it is a ps payload
InfoL << _media_info.stream << " judged to be PS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ps, _interface);
}

10
src/Rtp/GB28181Process.h
View File

@@ -33,11 +33,21 @@ public:
* @param data rtp数据指针
* @param data_len rtp数据长度
* @return 是否解析成功
* Input rtp
* @param data rtp data pointer
* @param data_len rtp data length
* @return Whether the parsing is successful
* [AUTO-TRANSLATED:d7b14ffe]
*/
bool inputRtp(bool, const char *data, size_t data_len) override;
/**
* 刷新输出所有缓存
* Refresh and output all caches
* [AUTO-TRANSLATED:4509b01f]
*/
void flush() override;

9
src/Rtp/PSDecoder.cpp
View File

@@ -57,17 +57,20 @@ const char *PSDecoder::onSearchPacketTail(const char *data, size_t len) {
try {
auto ret = ps_demuxer_input(static_cast<struct ps_demuxer_t *>(_ps_demuxer), reinterpret_cast<const uint8_t *>(data), len);
if (ret >= 0) {
//解析成功全部或部分
// 解析成功全部或部分 [AUTO-TRANSLATED:a8085d34]
// Parse successful, all or part
return data + ret;
}
//解析失败，丢弃所有数据
// 解析失败，丢弃所有数据 [AUTO-TRANSLATED:e6f644d9]
// Parse failed, discard all data
return data + len;
} catch (toolkit::AssertFailedException &ex) {
InfoL << "解析 ps 异常: bytes=" << len
<< ", exception=" << ex.what()
<< ", hex=" << hexdump(data, MIN(len, 32));
//触发断言，解析失败，丢弃所有数据
// 触发断言，解析失败，丢弃所有数据 [AUTO-TRANSLATED:b60c6db0]
// Trigger assertion, parse failed, discard all data
return data + len;
}
}

3
src/Rtp/PSDecoder.h
View File

@@ -18,7 +18,8 @@
namespace mediakit{
//ps解析器
// ps解析器 [AUTO-TRANSLATED:f156a1f1]
// ps parser
class PSDecoder : public Decoder, private HttpRequestSplitter {
public:
PSDecoder();

3
src/Rtp/PSEncoder.cpp
View File

@@ -24,7 +24,8 @@ PSEncoderImp::PSEncoderImp(uint32_t ssrc, uint8_t payload_type, bool ps_or_ts) :
_rtp_encoder = std::make_shared<CommonRtpEncoder>();
auto video_mtu = s_video_mtu;
if (!ps_or_ts) {
// 确保ts rtp负载部分长度是188的倍数
// 确保ts rtp负载部分长度是188的倍数 [AUTO-TRANSLATED:ad7aa6c0]
// Ensure the ts rtp payload length is a multiple of 188
video_mtu = RtpPacket::kRtpHeaderSize + (s_video_mtu - (s_video_mtu % 188));
if (video_mtu > s_video_mtu) {
video_mtu -= 188;

9
src/Rtp/PSEncoder.h
View File

@@ -27,12 +27,19 @@ public:
* @param ssrc rtp的ssrc
* @param payload_type rtp的pt
* @param ps_or_ts true: ps, false: ts
* Create a psh or ts rtp encoder
* @param ssrc rtp's ssrc
* @param payload_type rtp's pt
* @param ps_or_ts true: ps, false: ts
* [AUTO-TRANSLATED:b79d8b65]
*/
PSEncoderImp(uint32_t ssrc, uint8_t payload_type = 96, bool ps_or_ts = true);
~PSEncoderImp() override;
protected:
//rtp打包后回调
// rtp打包后回调 [AUTO-TRANSLATED:8f88aef9]
// Callback after rtp packaging
virtual void onRTP(toolkit::Buffer::Ptr rtp, bool is_key = false) = 0;
protected:

11
src/Rtp/ProcessInterface.h
View File

@@ -28,11 +28,22 @@ public:
* @param data rtp数据指针
* @param data_len rtp数据长度
* @return 是否解析成功
* Input rtp
* @param is_udp Whether it is udp mode
* @param data rtp data pointer
* @param data_len rtp data length
* @return Whether the parsing is successful
* [AUTO-TRANSLATED:7d5b06f0]
*/
virtual bool inputRtp(bool is_udp, const char *data, size_t data_len) = 0;
/**
* 刷新输出所有缓存
* Refresh and output all caches
* [AUTO-TRANSLATED:4509b01f]
*/
virtual void flush() {}
};

12
src/Rtp/RawEncoder.h
View File

@@ -25,21 +25,31 @@ public:
/**
* 添加音视频轨道
* Add audio and video tracks
* [AUTO-TRANSLATED:7b0c1d64]
*/
bool addTrack(const Track::Ptr &track) override;
/**
* 重置音视频轨道
* Reset audio and video tracks
* [AUTO-TRANSLATED:6eb1b742]
*/
void resetTracks() override;
/**
* 输入帧数据
* Input frame data
* [AUTO-TRANSLATED:d13bc7f2]
*/
bool inputFrame(const Frame::Ptr &frame) override;
protected:
// rtp打包后回调
// rtp打包后回调 [AUTO-TRANSLATED:61f5159b]
// Callback after RTP packaging
virtual void onRTP(toolkit::Buffer::Ptr rtp, bool is_key = false) = 0;
private:

5
src/Rtp/RtpCache.h
View File

@@ -28,6 +28,11 @@ protected:
/**
* 输入rtp(目的是为了合并写)
* @param buffer rtp数据
* Input rtp (for merging)
* @param buffer rtp data
* [AUTO-TRANSLATED:de9469b5]
*/
void input(uint64_t stamp, toolkit::Buffer::Ptr buffer,bool is_key = false);

27
src/Rtp/RtpProcess.cpp
View File

@@ -17,8 +17,10 @@
using namespace std;
using namespace toolkit;
//在创建_muxer对象前(也就是推流鉴权成功前)，需要先缓存frame，这样可以防止丢包，提高体验
//但是同时需要控制缓冲长度，防止内存溢出。最多缓存10秒数据，应该足矣等待鉴权hook返回
// 在创建_muxer对象前(也就是推流鉴权成功前)，需要先缓存frame，这样可以防止丢包，提高体验 [AUTO-TRANSLATED:fb12a6c2]
// Before creating the _muxer object (before the streaming authentication is successful), you need to cache the frame first, which can prevent packet loss and improve the experience.
// 但是同时需要控制缓冲长度，防止内存溢出。最多缓存10秒数据，应该足矣等待鉴权hook返回 [AUTO-TRANSLATED:23ff0a4a]
// But at the same time, you need to control the buffer length to prevent memory overflow. Caching 10 seconds of data should be enough to wait for the authentication hook to return.
static constexpr size_t kMaxCachedFrameMS = 10 * 1000;
namespace mediakit {
@@ -65,7 +67,8 @@ RtpProcess::~RtpProcess() {
<< _media_info.shortUrl()
<< ")断开,耗时(s):" << duration;
//流量统计事件广播
// 流量统计事件广播 [AUTO-TRANSLATED:6b0b1234]
// Traffic statistics event broadcast
GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
if (_total_bytes >= iFlowThreshold * 1024) {
try {
@@ -83,7 +86,8 @@ void RtpProcess::onManager() {
}
void RtpProcess::createTimer() {
//创建超时管理定时器
// 创建超时管理定时器 [AUTO-TRANSLATED:865cf865]
// Create a timeout management timer
weak_ptr<RtpProcess> weakSelf = shared_from_this();
_timer = std::make_shared<Timer>(3.0f, [weakSelf] {
auto strongSelf = weakSelf.lock();
@@ -104,7 +108,8 @@ bool RtpProcess::inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data
throw toolkit::SockException(toolkit::Err_other, _auth_err);
}
if (_sock != sock) {
// 第一次运行本函数
// 第一次运行本函数 [AUTO-TRANSLATED:a1d7ac17]
// First time running this function
bool first = !_sock;
_sock = sock;
_addr.reset(new sockaddr_storage(*((sockaddr_storage *)addr)));
@@ -130,7 +135,8 @@ bool RtpProcess::inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);
if (_muxer && !_muxer->isEnabled() && !dts_out && dump_dir.empty()) {
//无人访问、且不取时间戳、不导出调试文件时，我们可以直接丢弃数据
// 无人访问、且不取时间戳、不导出调试文件时，我们可以直接丢弃数据 [AUTO-TRANSLATED:2fc75705]
// When there is no access, and no timestamp is taken, and no debug file is exported, we can directly discard the data.
_last_frame_time.resetTime();
return false;
}
@@ -198,7 +204,8 @@ void RtpProcess::doCachedFunc() {
bool RtpProcess::alive() {
if (_stop_rtp_check.load()) {
if(_last_check_alive.elapsedTime() > 5 * 60 * 1000){
//最多暂停5分钟的rtp超时检测，因为NAT映射有效期一般不会太长
// 最多暂停5分钟的rtp超时检测，因为NAT映射有效期一般不会太长 [AUTO-TRANSLATED:2df59aad]
// Pause the RTP timeout detection for a maximum of 5 minutes, because the NAT mapping validity period is generally not very long.
_stop_rtp_check = false;
} else {
return true;
@@ -293,10 +300,12 @@ void RtpProcess::emitOnPublish() {
});
};
//触发推流鉴权事件
// 触发推流鉴权事件 [AUTO-TRANSLATED:cd889b29]
// Trigger the streaming authentication event
auto flag = NOTICE_EMIT(BroadcastMediaPublishArgs, Broadcast::kBroadcastMediaPublish, MediaOriginType::rtp_push, _media_info, invoker, *this);
if (!flag) {
// 该事件无人监听,默认不鉴权
// 该事件无人监听,默认不鉴权 [AUTO-TRANSLATED:e1fbc6ae]
// No one is listening to this event, and authentication is not performed by default.
invoker("", ProtocolOption());
}
}

27
src/Rtp/RtpProcess.h
View File

@@ -38,33 +38,60 @@ public:
* @param addr 数据源地址
* @param dts_out 解析出最新的dts
* @return 是否解析成功
* Input rtp
* @param is_udp Whether it is udp mode
* @param sock Local listening socket
* @param data Rtp data pointer
* @param len Rtp data length
* @param addr Data source address
* @param dts_out Parse out the latest dts
* @return Whether the parsing is successful
* [AUTO-TRANSLATED:a10c5edf]
*/
bool inputRtp(bool is_udp, const toolkit::Socket::Ptr &sock, const char *data, size_t len, const struct sockaddr *addr , uint64_t *dts_out = nullptr);
/**
* 超时时被RtpSelector移除时触发
* Triggered when removed by RtpSelector when timeout
* [AUTO-TRANSLATED:dc4c6609]
*/
void onDetach(const toolkit::SockException &ex);
/**
* 设置onDetach事件回调
* Set onDetach event callback
* [AUTO-TRANSLATED:b30f67c3]
*/
void setOnDetach(onDetachCB cb);
/**
* 设置onDetach事件回调,false检查RTP超时，true停止
* Set onDetach event callback, false checks RTP timeout, true stops
* [AUTO-TRANSLATED:2780397f]
*/
void setStopCheckRtp(bool is_check=false);
/**
* 设置为单track，单音频/单视频时可以加快媒体注册速度
* 请在inputRtp前调用此方法，否则可能会是空操作
* Set to single track, single audio/single video can speed up media registration
* Please call this method before inputRtp, otherwise it may be a null operation
* [AUTO-TRANSLATED:55095289]
*/
void setOnlyTrack(OnlyTrack only_track);
/**
* flush输出缓存
* Flush output cache
* [AUTO-TRANSLATED:40618a29]
*/
void flush() override;

111
src/Rtp/RtpSender.cpp
View File

@@ -38,7 +38,8 @@ RtpSender::~RtpSender() {
void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const function<void(uint16_t local_port, const SockException &ex)> &cb){
_args = args;
if (!_interface) {
//重连时不重新创建对象
// 重连时不重新创建对象 [AUTO-TRANSLATED:b788cd5d]
// Do not recreate the object when reconnecting
auto lam = [this](std::shared_ptr<List<Buffer::Ptr>> list) { onFlushRtpList(std::move(list)); };
switch (args.data_type) {
case MediaSourceEvent::SendRtpArgs::kRtpPS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, true); break;
@@ -53,18 +54,22 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
if (args.con_type == MediaSourceEvent::SendRtpArgs::kTcpPassive) {
auto tcp_listener = Socket::createSocket(_poller, false);
if (args.src_port) {
// 指定端口
// 指定端口 [AUTO-TRANSLATED:ed4ca3dd]
// Specify the port
if (!tcp_listener->listen(args.src_port)) {
throw std::invalid_argument(StrPrinter << "open tcp passive server failed on port: " << args.src_port << ", err: " << get_uv_errmsg(true));
}
} else {
auto pr = std::make_pair(tcp_listener, Socket::createSocket(_poller, false));
// 从端口池获取随机端口
// 从端口池获取随机端口 [AUTO-TRANSLATED:139ceb4f]
// Get a random port from the port pool
makeSockPair(pr, "::", true, false);
}
// 定时器持有tcp_listener，保证超时时间内保持监听
// 定时器持有tcp_listener，保证超时时间内保持监听 [AUTO-TRANSLATED:39df3f48]
// The timer holds the tcp_listener to ensure listening within the timeout period
auto delay_task = _poller->doDelayTask(delay_ms, [weak_self, tcp_listener]() mutable {
// 防止循环引用
// 防止循环引用 [AUTO-TRANSLATED:e2e9f9e7]
// Prevent circular references
tcp_listener = nullptr;
if (auto strong_self = weak_self.lock()) {
strong_self->onClose(SockException(Err_timeout, "wait tcp connection timeout"));
@@ -86,18 +91,21 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
} else if (args.con_type == MediaSourceEvent::SendRtpArgs::kUdpPassive) {
if (args.src_port) {
// 指定端口
// 指定端口 [AUTO-TRANSLATED:ed4ca3dd]
// Specify the port
if (!_socket_rtp->bindUdpSock(args.src_port, "::", true)) {
throw std::invalid_argument(StrPrinter << "open udp passive server failed on port: " << args.src_port << ", err: " << get_uv_errmsg(true));
}
} else {
auto pr = std::make_pair(_socket_rtp, Socket::createSocket(_poller, false));
// 从端口池获取随机端口
// 从端口池获取随机端口 [AUTO-TRANSLATED:139ceb4f]
// Get a random port from the port pool
makeSockPair(pr, "::", true, true);
}
auto delay_task = _poller->doDelayTask(delay_ms, [weak_self]() mutable {
if (auto strong_self = weak_self.lock()) {
// 关闭端口
// 关闭端口 [AUTO-TRANSLATED:3b3dff64]
// Close the port
strong_self->_socket_rtp->closeSock();
strong_self->onClose(SockException(Err_timeout, "wait udp connection timeout"));
}
@@ -110,7 +118,8 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
}
delay_task->cancel();
strong_self->_socket_rtp->bindPeerAddr(addr, addr_len, true);
// 异步执行onConnect，防止在OnRead回调中调用setOnRead
// 异步执行onConnect，防止在OnRead回调中调用setOnRead [AUTO-TRANSLATED:83881d7f]
// Execute onConnect asynchronously to prevent calling setOnRead in the OnRead callback
strong_self->_poller->async([strong_self]() { strong_self->onConnect(); }, false);
InfoL << "accept udp connection from: " << strong_self->_socket_rtp->get_peer_ip() << ":" << strong_self->_socket_rtp->get_peer_port();
});
@@ -121,18 +130,22 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
auto poller = _poller;
WorkThreadPool::Instance().getPoller()->async([cb, args, weak_self, poller]() {
struct sockaddr_storage addr;
// 切换线程目的是为了dns解析放在后台线程执行
// 切换线程目的是为了dns解析放在后台线程执行 [AUTO-TRANSLATED:1a09ba8a]
// The purpose of switching threads is to perform DNS resolution in the background thread
if (!SockUtil::getDomainIP(args.dst_url.data(), args.dst_port, addr, AF_INET, SOCK_DGRAM, IPPROTO_UDP)) {
poller->async([args, cb]() {
// 切回自己的线程
// 切回自己的线程 [AUTO-TRANSLATED:b95746d6]
// Switch back to your own thread
cb(0, SockException(Err_dns, StrPrinter << "dns resolution failed: " << args.dst_url));
});
return;
}
// dns解析成功
// dns解析成功 [AUTO-TRANSLATED:e1b35821]
// DNS resolution successful
poller->async([args, addr, weak_self, cb]() {
// 切回自己的线程
// 切回自己的线程 [AUTO-TRANSLATED:b95746d6]
// Switch back to your own thread
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
@@ -140,13 +153,15 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
string ifr_ip = addr.ss_family == AF_INET ? "0.0.0.0" : "::";
try {
if (args.src_port) {
// 指定端口
// 指定端口 [AUTO-TRANSLATED:ed4ca3dd]
// Specify the port
if (!strong_self->_socket_rtp->bindUdpSock(args.src_port, ifr_ip, true)) {
throw std::invalid_argument(StrPrinter << "open udp active client failed on port: " << args.src_port << ", err: " << get_uv_errmsg(true));
}
} else {
auto pr = std::make_pair(strong_self->_socket_rtp, Socket::createSocket(strong_self->_poller, false));
// 从端口池获取随机端口
// 从端口池获取随机端口 [AUTO-TRANSLATED:139ceb4f]
// Get a random port from the port pool
makeSockPair(pr, ifr_ip, true, true);
}
} catch (std::exception &ex) {
@@ -165,7 +180,8 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
auto strong_self = weak_self.lock();
if (strong_self) {
if (!err) {
// tcp连接成功
// tcp连接成功 [AUTO-TRANSLATED:1a33669a]
// TCP connection successful
strong_self->onConnect();
}
cb(strong_self->_socket_rtp->get_local_port(), err);
@@ -184,14 +200,16 @@ void RtpSender::createRtcpSocket() {
return;
}
_socket_rtcp = Socket::createSocket(_socket_rtp->getPoller(), false);
//rtcp端口使用户rtp端口+1
// rtcp端口使用户rtp端口+1 [AUTO-TRANSLATED:8a0a6b2c]
// The RTCP port is the RTP port + 1
if(!_socket_rtcp->bindUdpSock(_socket_rtp->get_local_port() + 1, _socket_rtp->get_local_ip(), true)){
WarnL << "bind rtcp udp socket failed: " << get_uv_errmsg(true);
_socket_rtcp = nullptr;
return;
}
// 绑定目标rtcp端口(目标rtp端口 + 1)
// 绑定目标rtcp端口(目标rtp端口 + 1) [AUTO-TRANSLATED:c402de5b]
// Bind the target RTCP port (target RTP port + 1)
auto addr = SockUtil::make_sockaddr(_socket_rtp->get_peer_ip().data(), _socket_rtp->get_peer_port() + 1);
_socket_rtcp->bindPeerAddr((struct sockaddr *)&addr, 0, true);
@@ -199,13 +217,15 @@ void RtpSender::createRtcpSocket() {
weak_ptr<RtpSender> weak_self = shared_from_this();
bool bind_addr = false;
_socket_rtcp->setOnRead([weak_self, bind_addr](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) mutable {
//接收receive report rtcp
// 接收receive report rtcp [AUTO-TRANSLATED:38d3c1ba]
// Receive receive report RTCP
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
if (!bind_addr) {
// 收到对方rtcp打洞包后，再回复rtcp
// 收到对方rtcp打洞包后，再回复rtcp [AUTO-TRANSLATED:393868ca]
// Reply RTCP after receiving the other party's RTCP hole punching packet
bind_addr = true;
strong_self->_socket_rtcp->bindPeerAddr(addr, addr_len, true);
}
@@ -222,24 +242,29 @@ void RtpSender::onRecvRtcp(RtcpHeader *rtcp) {
_rtcp_recv_ticker.resetTime();
}
// 连接建立成功事件
// 连接建立成功事件 [AUTO-TRANSLATED:ac279c86]
// Connection established successfully event
void RtpSender::onConnect() {
_is_connect = true;
// 加大发送缓存,防止udp丢包之类的问题
// 加大发送缓存,防止udp丢包之类的问题 [AUTO-TRANSLATED:6e1cb40a]
// Increase the send buffer to prevent problems such as UDP packet loss
SockUtil::setSendBuf(_socket_rtp->rawFD(), 4 * 1024 * 1024);
if (_args.con_type == MediaSourceEvent::SendRtpArgs::kTcpActive || _args.con_type == MediaSourceEvent::SendRtpArgs::kTcpPassive) {
// 关闭tcp no_delay并开启MSG_MORE, 提高发送性能
// 关闭tcp no_delay并开启MSG_MORE, 提高发送性能 [AUTO-TRANSLATED:c0f4e378]
// Close TCP no_delay and enable MSG_MORE to improve sending performance
SockUtil::setNoDelay(_socket_rtp->rawFD(), false);
_socket_rtp->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
} else if (_args.udp_rtcp_timeout) {
createRtcpSocket();
}
// 连接建立成功事件
// 连接建立成功事件 [AUTO-TRANSLATED:ac279c86]
// Connection established successfully event
weak_ptr<RtpSender> weak_self = shared_from_this();
if (!_args.recv_stream_id.empty()) {
mINI ini;
ini[RtpSession::kStreamID] = _args.recv_stream_id;
// 强制同步接收流和发送流的app和vhost
// 强制同步接收流和发送流的app和vhost [AUTO-TRANSLATED:134c9663]
// Force synchronization of the app and vhost of the receive stream and send stream
ini[RtpSession::kApp] = _args.recv_stream_app;
ini[RtpSession::kVhost] = _args.recv_stream_vhost;
_rtp_session = std::make_shared<RtpSession>(_socket_rtp);
@@ -271,7 +296,8 @@ void RtpSender::onConnect() {
bool RtpSender::addTrack(const Track::Ptr &track) {
if (_args.only_audio && track->getTrackType() == TrackVideo) {
// 如果只发送音频则忽略视频
// 如果只发送音频则忽略视频 [AUTO-TRANSLATED:6843e322]
// Ignore video if only audio is sent
return false;
}
return _interface->addTrack(track);
@@ -293,10 +319,12 @@ void RtpSender::flush() {
bool RtpSender::inputFrame(const Frame::Ptr &frame) {
if (_args.only_audio && frame->getTrackType() == TrackVideo) {
// 如果只发送音频则忽略视频
// 如果只发送音频则忽略视频 [AUTO-TRANSLATED:6843e322]
// Ignore video if only audio is sent
return false;
}
// 连接成功后才做实质操作(节省cpu资源)
// 连接成功后才做实质操作(节省cpu资源) [AUTO-TRANSLATED:666253b3]
// Perform the actual operation after the connection is successful (save CPU resources)
return _is_connect ? _interface->inputFrame(frame) : false;
}
@@ -308,20 +336,24 @@ void RtpSender::onSendRtpUdp(const toolkit::Buffer::Ptr &buf, bool check) {
_rtcp_context->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->ntp_stamp, 90000 /*not used*/, rtp->size());
if (!check) {
// 减少判断次数
// 减少判断次数 [AUTO-TRANSLATED:0cfaddd8]
// Reduce the number of judgments
return;
}
// 每5秒发送一次rtcp
// 每5秒发送一次rtcp [AUTO-TRANSLATED:3c9bcb7b]
// Send RTCP every 5 seconds
if (_rtcp_send_ticker.elapsedTime() > _args.rtcp_send_interval_ms) {
_rtcp_send_ticker.resetTime();
// rtcp ssrc为rtp ssrc + 1
// rtcp ssrc为rtp ssrc + 1 [AUTO-TRANSLATED:318fada3]
// rtcp ssrc is rtp ssrc + 1
auto sr = _rtcp_context->createRtcpSR(atoi(_args.ssrc.data()) + 1);
// send sender report rtcp
_socket_rtcp->send(sr);
}
if (_rtcp_recv_ticker.elapsedTime() > _args.rtcp_timeout_ms) {
// 接收rr rtcp超时
// 接收rr rtcp超时 [AUTO-TRANSLATED:a6ccd262]
// Receive rr rtcp timeout
WarnL << "recv rr rtcp timeout";
_rtcp_recv_ticker.resetTime();
onClose(SockException(Err_timeout, "recv rr rtcp timeout"));
@@ -331,15 +363,18 @@ void RtpSender::onSendRtpUdp(const toolkit::Buffer::Ptr &buf, bool check) {
void RtpSender::onClose(const SockException &ex) {
auto cb = _on_close;
if (cb) {
// 在下次循环时触发onClose，原因是防止遍历map时删除元素
// 在下次循环时触发onClose，原因是防止遍历map时删除元素 [AUTO-TRANSLATED:2841df7f]
// Trigger onClose in the next loop to prevent deleting elements while iterating through the map
_poller->async([cb, ex]() { cb(ex); }, false);
}
}
// 此函数在其他线程执行
// 此函数在其他线程执行 [AUTO-TRANSLATED:3569a681]
// This function is executed in another thread
void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr>> rtp_list) {
if (!_is_connect) {
// 连接成功后才能发送数据
// 连接成功后才能发送数据 [AUTO-TRANSLATED:14d00ad5]
// Data can only be sent after the connection is successful
return;
}
@@ -350,13 +385,15 @@ void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr>> rtp_list) {
case MediaSourceEvent::SendRtpArgs::kUdpActive:
case MediaSourceEvent::SendRtpArgs::kUdpPassive: {
onSendRtpUdp(packet, i == 0);
// udp模式，rtp over tcp前4个字节可以忽略
// udp模式，rtp over tcp前4个字节可以忽略 [AUTO-TRANSLATED:5d648f4b]
// UDP mode, the first 4 bytes of rtp over tcp can be ignored
_socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), RtpPacket::kRtpTcpHeaderSize), nullptr, 0, ++i == size);
break;
}
case MediaSourceEvent::SendRtpArgs::kTcpActive:
case MediaSourceEvent::SendRtpArgs::kTcpPassive: {
// tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节
// tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节 [AUTO-TRANSLATED:a3bc338a]
// TCP mode, the first 2 bytes of rtp over tcp can be ignored, only the subsequent 2 bytes of rtp length are retained
_socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
break;
}

37
src/Rtp/RtpSender.h
View File

@@ -21,7 +21,8 @@ namespace mediakit{
class RtpSession;
//rtp发送客户端，支持发送GB28181协议
// rtp发送客户端，支持发送GB28181协议 [AUTO-TRANSLATED:668038b6]
// RTP sending client, supporting sending GB28181 protocol
class RtpSender final : public MediaSinkInterface, public std::enable_shared_from_this<RtpSender>{
public:
using Ptr = std::shared_ptr<RtpSender>;
@@ -33,16 +34,27 @@ public:
* 开始发送ps-rtp包
* @param args 发送参数
* @param cb 连接目标端口是否成功的回调
* Start sending ps-rtp packets
* @param args Sending parameters
* @param cb Callback for whether the connection to the target port is successful
* [AUTO-TRANSLATED:c31bd9b3]
*/
void startSend(const MediaSourceEvent::SendRtpArgs &args, const std::function<void(uint16_t local_port, const toolkit::SockException &ex)> &cb);
/**
* 输入帧数据
* Input frame data
* [AUTO-TRANSLATED:d13bc7f2]
*/
bool inputFrame(const Frame::Ptr &frame) override;
/**
* 刷新输出frame缓存
* Refresh the output frame cache
* [AUTO-TRANSLATED:547f851c]
*/
void flush() override;
@@ -50,30 +62,47 @@ public:
* 添加track，内部会调用Track的clone方法
* 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系
* @param track
* Add track, internally calls the clone method of Track
* Only clones sps pps information, not Delegate relationships
* @param track
* [AUTO-TRANSLATED:ba6faf58]
*/
virtual bool addTrack(const Track::Ptr & track) override;
/**
* 添加所有Track完毕
* All Tracks added
* [AUTO-TRANSLATED:751c45ca]
*/
virtual void addTrackCompleted() override;
/**
* 重置track
* Reset track
* [AUTO-TRANSLATED:95dc0b4f]
*/
virtual void resetTracks() override;
/**
* 设置发送rtp停止回调
* Set RTP sending stop callback
* [AUTO-TRANSLATED:7e0a6714]
*/
void setOnClose(std::function<void(const toolkit::SockException &ex)> on_close);
private:
//合并写输出
// 合并写输出 [AUTO-TRANSLATED:23544836]
// Merge write output
void onFlushRtpList(std::shared_ptr<toolkit::List<toolkit::Buffer::Ptr> > rtp_list);
//udp/tcp连接成功回调
// udp/tcp连接成功回调 [AUTO-TRANSLATED:ca35017d]
// UDP/TCP connection success callback
void onConnect();
//异常断开socket事件
// 异常断开socket事件 [AUTO-TRANSLATED:a59cd9de]
// Abnormal socket disconnect event
void onErr(const toolkit::SockException &ex);
void createRtcpSocket();
void onRecvRtcp(RtcpHeader *rtcp);

60
src/Rtp/RtpServer.cpp
View File

@@ -63,7 +63,8 @@ public:
void onRecvRtp(const Socket::Ptr &sock, const Buffer::Ptr &buf, struct sockaddr *addr) {
_process->inputRtp(true, sock, buf->data(), buf->size(), addr);
// 统计rtp接受情况，用于发送rr包
// 统计rtp接受情况，用于发送rr包 [AUTO-TRANSLATED:bd2fbe7e]
// Count RTP reception status, used to send RR packets
auto header = (RtpHeader *)buf->data();
sendRtcp(ntohl(header->ssrc), addr);
}
@@ -71,13 +72,15 @@ public:
void startRtcp() {
weak_ptr<RtcpHelper> weak_self = shared_from_this();
_rtcp_sock->setOnRead([weak_self](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
// 用于接受rtcp打洞包
// 用于接受rtcp打洞包 [AUTO-TRANSLATED:d75d9d87]
// Used to receive RTCP hole punching packets
auto strong_self = weak_self.lock();
if (!strong_self || !strong_self->_process) {
return;
}
if (!strong_self->_rtcp_addr) {
// 只设置一次rtcp对端端口
// 只设置一次rtcp对端端口 [AUTO-TRANSLATED:fdc4eb4e]
// Set the RTCP peer port only once
strong_self->_rtcp_addr = std::make_shared<struct sockaddr_storage>();
memcpy(strong_self->_rtcp_addr.get(), addr, addr_len);
}
@@ -85,14 +88,16 @@ public:
for (auto &rtcp : rtcps) {
strong_self->_process->onRtcp(rtcp);
}
// 收到sr rtcp后驱动返回rr rtcp
// 收到sr rtcp后驱动返回rr rtcp [AUTO-TRANSLATED:d7373077]
// After receiving SR RTCP, the driver returns RR RTCP
strong_self->sendRtcp(strong_self->_ssrc, (struct sockaddr *)(strong_self->_rtcp_addr.get()));
});
}
private:
void sendRtcp(uint32_t rtp_ssrc, struct sockaddr *addr) {
// 每5秒发送一次rtcp
// 每5秒发送一次rtcp [AUTO-TRANSLATED:3c9bcb7b]
// Send RTCP every 5 seconds
if (_ticker.elapsedTime() < 5000) {
return;
}
@@ -100,12 +105,14 @@ private:
auto rtcp_addr = (struct sockaddr *)_rtcp_addr.get();
if (!rtcp_addr) {
// 默认的，rtcp端口为rtp端口+1
// 默认的，rtcp端口为rtp端口+1 [AUTO-TRANSLATED:273d164d]
// By default, the RTCP port is the RTP port + 1
switch (addr->sa_family) {
case AF_INET: ((sockaddr_in *)addr)->sin_port = htons(ntohs(((sockaddr_in *)addr)->sin_port) + 1); break;
case AF_INET6: ((sockaddr_in6 *)addr)->sin6_port = htons(ntohs(((sockaddr_in6 *)addr)->sin6_port) + 1); break;
}
// 未收到rtcp打洞包时，采用默认的rtcp端口
// 未收到rtcp打洞包时，采用默认的rtcp端口 [AUTO-TRANSLATED:d99b12b8]
// When no RTCP hole punching packet is received, the default RTCP port is used
rtcp_addr = addr;
}
_rtcp_sock->send(_process->createRtcpRR(rtp_ssrc + 1, rtp_ssrc), rtcp_addr);
@@ -123,33 +130,41 @@ private:
};
void RtpServer::start(uint16_t local_port, const char *local_ip, const MediaTuple &tuple, TcpMode tcp_mode, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) {
//创建udp服务器
// 创建udp服务器 [AUTO-TRANSLATED:99619428]
// Create UDP server
auto poller = EventPollerPool::Instance().getPoller();
Socket::Ptr rtp_socket = Socket::createSocket(poller, true);
Socket::Ptr rtcp_socket = Socket::createSocket(poller, true);
if (local_port == 0) {
//随机端口，rtp端口采用偶数
// 随机端口，rtp端口采用偶数 [AUTO-TRANSLATED:3664eaf5]
// Random port, RTP port uses even numbers
auto pair = std::make_pair(rtp_socket, rtcp_socket);
makeSockPair(pair, local_ip, re_use_port);
local_port = rtp_socket->get_local_port();
} else if (!rtp_socket->bindUdpSock(local_port, local_ip, re_use_port)) {
//用户指定端口
// 用户指定端口 [AUTO-TRANSLATED:1328393b]
// User-specified port
throw std::runtime_error(StrPrinter << "创建rtp端口 " << local_ip << ":" << local_port << " 失败:" << get_uv_errmsg(true));
} else if (!rtcp_socket->bindUdpSock(local_port + 1, local_ip, re_use_port)) {
// rtcp端口
// rtcp端口 [AUTO-TRANSLATED:00cf932e]
// RTCP port
throw std::runtime_error(StrPrinter << "创建rtcp端口 " << local_ip << ":" << local_port + 1 << " 失败:" << get_uv_errmsg(true));
}
//设置udp socket读缓存
// 设置udp socket读缓存 [AUTO-TRANSLATED:3bf101d7]
// Set UDP socket read cache
GET_CONFIG(int, udpRecvSocketBuffer, RtpProxy::kUdpRecvSocketBuffer);
SockUtil::setRecvBuf(rtp_socket->rawFD(), udpRecvSocketBuffer);
//创建udp服务器
// 创建udp服务器 [AUTO-TRANSLATED:99619428]
// Create UDP server
UdpServer::Ptr udp_server;
RtcpHelper::Ptr helper;
//增加了多路复用判断，如果多路复用为true，就走else逻辑，同时保留了原来stream_id为空走else逻辑
// 增加了多路复用判断，如果多路复用为true，就走else逻辑，同时保留了原来stream_id为空走else逻辑 [AUTO-TRANSLATED:114690b1]
// Added multiplexing judgment. If multiplexing is true, then go to the else logic, while retaining the original stream_id is empty to go to the else logic
if (!tuple.stream.empty() && !multiplex) {
//指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
// 指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流) [AUTO-TRANSLATED:3aeb611e]
// If a stream ID is specified, then one port is one stream (regardless of whether it contains multiple streams with multiple SSRCs, after binding the RTP source, streams that do not match the IP port will be filtered out)
helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), tuple);
helper->startRtcp();
helper->setRtpServerInfo(local_port, tcp_mode, re_use_port, ssrc, only_track);
@@ -164,7 +179,8 @@ void RtpServer::start(uint16_t local_port, const char *local_ip, const MediaTupl
WarnL << "ssrc mismatched, rtp dropped: " << rtp_ssrc << " != " << ssrc;
} else {
if (!bind_peer_addr) {
//绑定对方ip+端口，防止多个设备或一个设备多次推流从而日志报ssrc不匹配问题
// 绑定对方ip+端口，防止多个设备或一个设备多次推流从而日志报ssrc不匹配问题 [AUTO-TRANSLATED:f27dd373]
// Bind the peer IP + port to prevent multiple devices or one device from pushing multiple streams, resulting in log reports of mismatched SSRCs
bind_peer_addr = true;
rtp_socket->bindPeerAddr(addr, addr_len);
}
@@ -172,7 +188,8 @@ void RtpServer::start(uint16_t local_port, const char *local_ip, const MediaTupl
}
});
} else {
//单端口多线程接收多个流，根据ssrc区分流
// 单端口多线程接收多个流，根据ssrc区分流 [AUTO-TRANSLATED:e11c3ca8]
// Single-port multi-threaded reception of multiple streams, distinguishing streams based on SSRC
udp_server = std::make_shared<UdpServer>();
(*udp_server)[RtpSession::kOnlyTrack] = only_track;
(*udp_server)[RtpSession::kUdpRecvBuffer] = udpRecvSocketBuffer;
@@ -185,7 +202,8 @@ void RtpServer::start(uint16_t local_port, const char *local_ip, const MediaTupl
TcpServer::Ptr tcp_server;
if (tcp_mode == PASSIVE || tcp_mode == ACTIVE) {
auto processor = helper ? helper->getProcess() : nullptr;
// 如果共享同一个processor对象，那么tcp server深圳为单线程模式确保线程安全
// 如果共享同一个processor对象，那么tcp server深圳为单线程模式确保线程安全 [AUTO-TRANSLATED:68bdd877]
// If the same processor object is shared, then the TCP server Shenzhen is in single-threaded mode to ensure thread safety
tcp_server = std::make_shared<TcpServer>(processor ? poller : nullptr);
(*tcp_server)[RtpSession::kVhost] = tuple.vhost;
(*tcp_server)[RtpSession::kApp] = tuple.app;
@@ -198,14 +216,16 @@ void RtpServer::start(uint16_t local_port, const char *local_ip, const MediaTupl
session->setRtpProcess(processor);
});
} else if (tuple.stream.empty()) {
// tcp主动模式时只能一个端口一个流，必须指定流id; 创建TcpServer对象也仅用于传参
// tcp主动模式时只能一个端口一个流，必须指定流id; 创建TcpServer对象也仅用于传参 [AUTO-TRANSLATED:61d2a642]
// In TCP active mode, only one port can have one stream, and the stream ID must be specified; the TcpServer object is created only for parameter passing
throw std::runtime_error(StrPrinter << "tcp主动模式时必需指定流id");
}
}
_on_cleanup = [rtp_socket]() {
if (rtp_socket) {
//去除循环引用
// 去除循环引用 [AUTO-TRANSLATED:9afaed31]
// Remove circular references
rtp_socket->setOnRead(nullptr);
}
};

34
src/Rtp/RtpServer.h
View File

@@ -24,6 +24,9 @@ class RtcpHelper;
/**
* RTP服务器，支持UDP/TCP
* RTP server, supports UDP/TCP
* [AUTO-TRANSLATED:03e7daba]
*/
class RtpServer : public std::enable_shared_from_this<RtpServer> {
public:
@@ -42,6 +45,16 @@ public:
* @param re_use_port 是否设置socket为re_use属性
* @param ssrc 指定的ssrc
* @param multiplex 多路复用
* Start the server, may throw an exception
* @param local_port Local port, 0 for random port
* @param local_ip Local network interface ip to bind
* @param stream_id Stream id, use ssrc if empty
* @param tcp_mode TCP service mode
* @param re_use_port Whether to set the socket to re_use property
* @param ssrc Specified ssrc
* @param multiplex Multiplexing
* [AUTO-TRANSLATED:cb9e0717]
*/
void start(uint16_t local_port, const char *local_ip = "::", const MediaTuple &tuple = MediaTuple{DEFAULT_VHOST, kRtpAppName, "", ""}, TcpMode tcp_mode = PASSIVE,
bool re_use_port = true, uint32_t ssrc = 0, int only_track = 0, bool multiplex = false);
@@ -51,26 +64,42 @@ public:
* @param url 服务器地址
* @param port 服务器端口
* @param cb 连接服务器是否成功的回调
* Connect to the tcp service (tcp active mode)
* @param url Server address
* @param port Server port
* @param cb Callback whether the connection to the server is successful
* [AUTO-TRANSLATED:3de57bc0]
*/
void connectToServer(const std::string &url, uint16_t port, const std::function<void(const toolkit::SockException &ex)> &cb);
/**
* 获取绑定的本地端口
* Get the bound local port
* [AUTO-TRANSLATED:beed15d6]
*/
uint16_t getPort();
/**
* 设置RtpProcess onDetach事件回调
* Set RtpProcess onDetach event callback
* [AUTO-TRANSLATED:a316d07e]
*/
void setOnDetach(RtpProcess::onDetachCB cb);
/**
* 更新ssrc
* Update ssrc
* [AUTO-TRANSLATED:dae5e276]
*/
void updateSSRC(uint32_t ssrc);
private:
// tcp主动模式连接服务器成功回调
// tcp主动模式连接服务器成功回调 [AUTO-TRANSLATED:0775844e]
// tcp active mode connection server success callback
void onConnect();
protected:
@@ -82,7 +111,8 @@ protected:
std::function<void()> _on_cleanup;
int _only_track = 0;
//用于tcp主动模式
// 用于tcp主动模式 [AUTO-TRANSLATED:faedf05c]
// Used for tcp active mode
TcpMode _tcp_mode = NONE;
};

75
src/Rtp/RtpSession.cpp
View File

@@ -40,7 +40,8 @@ void RtpSession::setParams(mINI &ini) {
_only_track = ini[kOnlyTrack];
int udp_socket_buffer = ini[kUdpRecvBuffer];
if (_is_udp) {
// 设置udp socket读缓存
// 设置udp socket读缓存 [AUTO-TRANSLATED:80cfb6e3]
// Set udp socket read buffer
SockUtil::setRecvBuf(getSock()->rawFD(),
(udp_socket_buffer > 0) ? udp_socket_buffer : (4 * 1024 * 1024));
}
@@ -83,15 +84,18 @@ void RtpSession::setRtpProcess(RtpProcess::Ptr process) {
void RtpSession::onRtpPacket(const char *data, size_t len) {
if (!isRtp(data, len)) {
// 忽略非rtp数据
// 忽略非rtp数据 [AUTO-TRANSLATED:771b77d8]
// Ignore non-rtp data
WarnP(this) << "Not rtp packet";
return;
}
if (!_is_udp) {
if (_search_rtp) {
//搜索上下文期间，数据丢弃
// 搜索上下文期间，数据丢弃 [AUTO-TRANSLATED:e0a3b407]
// Data discarded during context search
if (_search_rtp_finished) {
//下个包开始就是正确的rtp包了
// 下个包开始就是正确的rtp包了 [AUTO-TRANSLATED:a73a3a61]
// The next packet is the correct rtp packet
_search_rtp_finished = false;
_search_rtp = false;
}
@@ -105,12 +109,14 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
}
}
// 未设置ssrc时，尝试获取ssrc
// 未设置ssrc时，尝试获取ssrc [AUTO-TRANSLATED:30f31a81]
// Try to get ssrc when ssrc is not set
if (!_ssrc && !getSSRC(data, len, _ssrc)) {
return;
}
// 未指定流id就使用ssrc为流id
// 未指定流id就使用ssrc为流id [AUTO-TRANSLATED:9eb98394]
// Use ssrc as stream id if stream id is not specified
if (_tuple.stream.empty()) {
_tuple.stream = printSSRC(_ssrc);
}
@@ -146,7 +152,8 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
}
static const char *findSSRC(const char *data, ssize_t len, uint32_t ssrc) {
// rtp前面必须预留两个字节的长度字段
// rtp前面必须预留两个字节的长度字段 [AUTO-TRANSLATED:2af4e647]
// Two bytes of length field must be reserved before rtp
for (ssize_t i = 2; i <= len - 4; ++i) {
auto ptr = (const uint8_t *)data + i;
if (ptr[0] == (ssrc >> 24) && ptr[1] == ((ssrc >> 16) & 0xFF) && ptr[2] == ((ssrc >> 8) & 0xFF)
@@ -160,7 +167,8 @@ static const char *findSSRC(const char *data, ssize_t len, uint32_t ssrc) {
static const char *findPsHeaderFlag(const char *data, ssize_t len) {
for (ssize_t i = 2; i <= len - 4; ++i) {
auto ptr = (const uint8_t *)data + i;
// PsHeader 0x000001ba、PsSystemHeader0x000001bb（关键帧标识）
// PsHeader 0x000001ba、PsSystemHeader0x000001bb（关键帧标识） [AUTO-TRANSLATED:f8146534]
// PsHeader 0x000001ba, PsSystemHeader 0x000001bb (keyframe identifier)
if (ptr[0] == (0x00) && ptr[1] == (0x00) && ptr[2] == (0x01) && ptr[3] == (0xbb)) {
return (const char *)ptr;
}
@@ -169,15 +177,19 @@ static const char *findPsHeaderFlag(const char *data, ssize_t len) {
return nullptr;
}
// rtp长度到ssrc间的长度固定为10
// rtp长度到ssrc间的长度固定为10 [AUTO-TRANSLATED:7428bd59]
// The length between rtp length and ssrc is fixed to 10
static size_t constexpr kSSRCOffset = 2 + 4 + 4;
// rtp长度到ps header间的长度固定为14 （暂时不采用找ps header,采用找system header代替）
// rtp长度到ps system header间的长度固定为20 (关键帧标识)
// rtp长度到ps header间的长度固定为14 （暂时不采用找ps header,采用找system header代替） [AUTO-TRANSLATED:cf6b289c]
// The length between rtp length and ps header is fixed to 14 (temporarily not using ps header, using system header instead)
// rtp长度到ps system header间的长度固定为20 (关键帧标识) [AUTO-TRANSLATED:abe8bb8e]
// The length between rtp length and ps system header is fixed to 20 (keyframe identifier)
static size_t constexpr kPSHeaderOffset = 2 + 4 + 4 + 4 + 20;
const char *RtpSession::onSearchPacketTail(const char *data, size_t len) {
if (!_search_rtp) {
// tcp上下文正常，不用搜索ssrc
// tcp上下文正常，不用搜索ssrc [AUTO-TRANSLATED:cab86669]
// Tcp context is normal, no need to search ssrc
return RtpSplitter::onSearchPacketTail(data, len);
}
if (!_process) {
@@ -189,51 +201,62 @@ const char *RtpSession::onSearchPacketTail(const char *data, size_t len) {
if (rtp_ptr0) {
return rtp_ptr0;
}
// ssrc搜索失败继续尝试搜索ps header flag
// ssrc搜索失败继续尝试搜索ps header flag [AUTO-TRANSLATED:e8f65bd2]
// Continue to search for ps header flag if ssrc search fails
auto rtp_ptr2 = searchByPsHeaderFlag(data, len);
return rtp_ptr2;
}
const char *RtpSession::searchBySSRC(const char *data, size_t len) {
InfoL << "尝试rtp搜索ssrc..._ssrc=" << _ssrc;
// 搜索第一个rtp的ssrc
// 搜索第一个rtp的ssrc [AUTO-TRANSLATED:6b010df0]
// Search for the first rtp's ssrc
auto ssrc_ptr0 = findSSRC(data, len, _ssrc);
if (!ssrc_ptr0) {
// 未搜索到任意rtp，返回数据不够
// 未搜索到任意rtp，返回数据不够 [AUTO-TRANSLATED:50db17ed]
// Return insufficient data if no rtp is found
InfoL << "rtp搜索ssrc失败（第一个数据不够），丢弃rtp数据为：" << len;
return nullptr;
}
// 这两个字节是第一个rtp的长度字段
// 这两个字节是第一个rtp的长度字段 [AUTO-TRANSLATED:75816ba4]
// These two bytes are the length field of the first rtp
auto rtp_len_ptr = (ssrc_ptr0 - kSSRCOffset);
auto rtp_len = ((uint8_t *)rtp_len_ptr)[0] << 8 | ((uint8_t *)rtp_len_ptr)[1];
// 搜索第二个rtp的ssrc
// 搜索第二个rtp的ssrc [AUTO-TRANSLATED:238eaa43]
// Search for the second rtp's ssrc
auto ssrc_ptr1 = findSSRC(ssrc_ptr0 + rtp_len, data + (ssize_t)len - ssrc_ptr0 - rtp_len, _ssrc);
if (!ssrc_ptr1) {
// 未搜索到第二个rtp，返回数据不够
// 未搜索到第二个rtp，返回数据不够 [AUTO-TRANSLATED:3a78a586]
// Return insufficient data if the second rtp is not found
InfoL << "rtp搜索ssrc失败(第二个数据不够)，丢弃rtp数据为：" << len;
return nullptr;
}
// 两个ssrc的间隔正好等于rtp的长度(外加rtp长度字段)，那么说明找到rtp
// 两个ssrc的间隔正好等于rtp的长度(外加rtp长度字段)，那么说明找到rtp [AUTO-TRANSLATED:b1517bfd]
// The interval between the two ssrcs is exactly equal to the length of the rtp (plus the rtp length field), which means that the rtp is found
auto ssrc_offset = ssrc_ptr1 - ssrc_ptr0;
if (ssrc_offset == rtp_len + 2 || ssrc_offset == rtp_len + 4) {
InfoL << "rtp搜索ssrc成功，tcp上下文恢复成功，丢弃的rtp残余数据为：" << rtp_len_ptr - data;
_search_rtp_finished = true;
if (rtp_len_ptr == data) {
// 停止搜索rtp，否则会进入死循环
// 停止搜索rtp，否则会进入死循环 [AUTO-TRANSLATED:319eefa7]
// Stop searching for rtp, otherwise it will enter an infinite loop
_search_rtp = false;
}
// 前面的数据都需要丢弃，这个是rtp的起始
// 前面的数据都需要丢弃，这个是rtp的起始 [AUTO-TRANSLATED:129082d2]
// All previous data needs to be discarded, this is the start of rtp
return rtp_len_ptr;
}
// 第一个rtp长度不匹配，说明第一个找到的ssrc不是rtp，丢弃之，我们从第二个ssrc所在rtp开始搜索
// 第一个rtp长度不匹配，说明第一个找到的ssrc不是rtp，丢弃之，我们从第二个ssrc所在rtp开始搜索 [AUTO-TRANSLATED:ec35b2ba]
// The length of the first rtp does not match, which means that the first ssrc found is not rtp, discard it, we start searching from the second ssrc rtp
return ssrc_ptr1 - kSSRCOffset;
}
const char *RtpSession::searchByPsHeaderFlag(const char *data, size_t len) {
InfoL << "尝试rtp搜索PsSystemHeaderFlag..._ssrc=" << _ssrc;
// 搜索rtp中的第一个PsHeaderFlag
// 搜索rtp中的第一个PsHeaderFlag [AUTO-TRANSLATED:77a18970]
// Search for the first PsHeaderFlag in rtp
auto ps_header_flag_ptr = findPsHeaderFlag(data, len);
if (!ps_header_flag_ptr) {
InfoL << "rtp搜索flag失败，丢弃rtp数据为：" << len;
@@ -243,12 +266,14 @@ const char *RtpSession::searchByPsHeaderFlag(const char *data, size_t len) {
auto rtp_ptr = ps_header_flag_ptr - kPSHeaderOffset;
_search_rtp_finished = true;
if (rtp_ptr == data) {
// 停止搜索rtp，否则会进入死循环
// 停止搜索rtp，否则会进入死循环 [AUTO-TRANSLATED:319eefa7]
// Stop searching for rtp, otherwise it will enter an infinite loop
_search_rtp = false;
}
InfoL << "rtp搜索flag成功，tcp上下文恢复成功，丢弃的rtp残余数据为：" << rtp_ptr - data;
// TODO or Not ? 更新设置ssrc
// TODO or Not ? 更新设置ssrc [AUTO-TRANSLATED:9c21db0a]
// TODO or Not ? Update setting ssrc
uint32_t rtp_ssrc = 0;
getSSRC(rtp_ptr + 2, len, rtp_ssrc);
_ssrc = rtp_ssrc;

9
src/Rtp/RtpSession.h
View File

@@ -39,13 +39,16 @@ public:
void setRtpProcess(RtpProcess::Ptr process);
protected:
// 收到rtp回调
// 收到rtp回调 [AUTO-TRANSLATED:446b2cda]
// Received RTP callback
void onRtpPacket(const char *data, size_t len) override;
// RtpSplitter override
const char *onSearchPacketTail(const char *data, size_t len) override;
// 搜寻SSRC
// 搜寻SSRC [AUTO-TRANSLATED:2cfec2e1]
// Search for SSRC
const char *searchBySSRC(const char *data, size_t len);
// 搜寻PS包里的关键帧标头
// 搜寻PS包里的关键帧标头 [AUTO-TRANSLATED:d8e88339]
// Search for keyframe header in PS packet
const char *searchByPsHeaderFlag(const char *data, size_t len);
private:

36
src/Rtp/RtpSplitter.cpp
View File

@@ -16,12 +16,14 @@ namespace mediakit{
static const int kEHOME_OFFSET = 256;
ssize_t RtpSplitter::onRecvHeader(const char *data,size_t len){
//忽略偏移量
// 忽略偏移量 [AUTO-TRANSLATED:7fbc3d5d]
// Ignore offset
data += _offset;
len -= _offset;
if (_is_ehome && len > 12 && data[12] == '\r') {
//这是ehome,移除第12个字节
// 这是ehome,移除第12个字节 [AUTO-TRANSLATED:643b3f39]
// This is ehome, remove the 12th byte
memmove((char *) data + 1, data, 12);
data += 1;
len -= 1;
@@ -42,21 +44,26 @@ static bool isEhome(const char *data, size_t len){
const char *RtpSplitter::onSearchPacketTail(const char *data, size_t len) {
if (len < 4) {
//数据不够
// 数据不够 [AUTO-TRANSLATED:72802244]
// Not enough data
return nullptr;
}
if (_check_ehome_count) {
if (isEhome(data, len)) {
//是ehome协议
// 是ehome协议 [AUTO-TRANSLATED:daa874ca]
// It is the ehome protocol
if (len < kEHOME_OFFSET + 4) {
//数据不够
// 数据不够 [AUTO-TRANSLATED:72802244]
// Not enough data
return nullptr;
}
//忽略ehome私有头后是rtsp样式的rtp，多4个字节，
// 忽略ehome私有头后是rtsp样式的rtp，多4个字节， [AUTO-TRANSLATED:d9bc652c]
// Ignore the ehome private header, then it is an rtsp-style rtp, with 4 extra bytes,
_offset = kEHOME_OFFSET + 4;
_is_ehome = true;
//忽略ehome私有头
// 忽略ehome私有头 [AUTO-TRANSLATED:4fc98661]
// Ignore the ehome private header
return onSearchPacketTail_l(data + kEHOME_OFFSET + 2, len - kEHOME_OFFSET - 2);
}
_check_ehome_count--;
@@ -64,26 +71,31 @@ const char *RtpSplitter::onSearchPacketTail(const char *data, size_t len) {
if ( _is_rtsp_interleaved ) {
if (data[0] == '$') {
//可能是4个字节的rtp头
// 可能是4个字节的rtp头 [AUTO-TRANSLATED:c287c3cb]
// It may be a 4-byte rtp header
_offset = 4;
return onSearchPacketTail_l(data + 2, len - 2);
}
_is_rtsp_interleaved = false;
}
//两个字节的rtp头
// 两个字节的rtp头 [AUTO-TRANSLATED:85fed462]
// A 2-byte rtp header
_offset = 2;
return onSearchPacketTail_l(data, len);
}
const char *RtpSplitter::onSearchPacketTail_l(const char *data, size_t len) {
//这是rtp包
// 这是rtp包 [AUTO-TRANSLATED:627d4881]
// This is an rtp packet
uint16_t length = (((uint8_t *) data)[0] << 8) | ((uint8_t *) data)[1];
if (len < (size_t)(length + 2)) {
//数据不够
// 数据不够 [AUTO-TRANSLATED:72802244]
// Not enough data
return nullptr;
}
//返回rtp包末尾
// 返回rtp包末尾 [AUTO-TRANSLATED:2546d5ce]
// Return the end of the rtp packet
return data + 2 + length;
}

6
src/Rtp/RtpSplitter.h
View File

@@ -22,6 +22,12 @@ protected:
* 收到rtp包回调
* @param data RTP包数据指针
* @param len RTP包数据长度
* RTP packet received callback
* @param data RTP packet data pointer
* @param len RTP packet data length
* [AUTO-TRANSLATED:18a85278]
*/
virtual void onRtpPacket(const char *data, size_t len) = 0;

12
src/Rtp/TSDecoder.cpp
View File

@@ -35,7 +35,8 @@ const char *TSSegment::onSearchPacketTail(const char *data, size_t len) {
}
return nullptr;
}
//下一个包头
// 下一个包头 [AUTO-TRANSLATED:c653c49d]
// Next packet header
if (((uint8_t *) data)[_size] == TS_SYNC_BYTE) {
return data + _size;
}
@@ -44,10 +45,12 @@ const char *TSSegment::onSearchPacketTail(const char *data, size_t len) {
return (char *) pos;
}
if (remainDataSize() > 4 * _size) {
//数据这么多都没ts包，全部清空
// 数据这么多都没ts包，全部清空 [AUTO-TRANSLATED:95bece98]
// So much data but no ts packets, clear all
return data + len;
}
//等待更多数据
// 等待更多数据 [AUTO-TRANSLATED:b47fbc81]
// Wait for more data
return nullptr;
}
@@ -93,7 +96,8 @@ ssize_t TSDecoder::input(const uint8_t *data, size_t bytes) {
try {
_ts_segment.input((char *) data, bytes);
} catch (...) {
//ts解析失败，清空缓存数据
// ts解析失败，清空缓存数据 [AUTO-TRANSLATED:18b3de5b]
// ts parsing failed, clear cache data
_ts_segment.reset();
throw;
}

6
src/Rtp/TSDecoder.h
View File

@@ -20,7 +20,8 @@
namespace mediakit {
//TS包分割器，用于split一个一个的ts包
// TS包分割器，用于split一个一个的ts包 [AUTO-TRANSLATED:a10b66b3]
// TS package splitter, used to split one ts package at a time
class TSSegment : public HttpRequestSplitter {
public:
typedef std::function<void(const char *data,size_t len)> onSegment;
@@ -38,7 +39,8 @@ private:
};
#if defined(ENABLE_HLS)
//ts解析器
// ts解析器 [AUTO-TRANSLATED:f2b9f0cc]
// ts parser
class TSDecoder : public Decoder {
public:
TSDecoder();