Skip to content

Z
ZLMediaKit
Commit 2818e371 by xiongziliang
Options

完善startSendRtp接口

parent de0738b1
正在显示 包含 111 行增加 和 73 行删除
api/source/mk_events_objects.cpp
......@@ -211,7 +211,14 @@ API_EXPORT int API_CALL mk_media_source_seek_to(const mk_media_source ctx,uint32
API_EXPORT void API_CALL mk_media_source_start_send_rtp(const mk_media_source ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_source_send_rtp_result cb, void *user_data){
assert(ctx && dst_url && ssrc);
MediaSource *src = (MediaSource *)ctx;
src->startSendRtp(dst_url, dst_port, ssrc, is_udp, 0, [cb, user_data](uint16_t local_port, const SockException &ex){
MediaSourceEvent::SendRtpArgs args;
args.dst_url = dst_url;
args.dst_port = dst_port;
args.ssrc = ssrc;
args.is_udp = is_udp;
src->startSendRtp(args, [cb, user_data](uint16_t local_port, const SockException &ex){
if (cb) {
cb(user_data, local_port, ex.getErrCode(), ex.what());
}
......
api/source/mk_media.cpp
......@@ -239,8 +239,15 @@ API_EXPORT int API_CALL mk_media_input_audio(mk_media ctx, const void* data, int
API_EXPORT void API_CALL mk_media_start_send_rtp(mk_media ctx, const char *dst_url, uint16_t dst_port, const char *ssrc, int is_udp, on_mk_media_send_rtp_result cb, void *user_data){
assert(ctx && dst_url && ssrc);
MediaHelper::Ptr* obj = (MediaHelper::Ptr*) ctx;
MediaSourceEvent::SendRtpArgs args;
args.dst_url = dst_url;
args.dst_port = dst_port;
args.ssrc = ssrc;
args.is_udp = is_udp;
//sender参数无用
(*obj)->getChannel()->startSendRtp(*MediaSource::NullMediaSource, dst_url, dst_port, ssrc, is_udp, 0, [cb, user_data](uint16_t local_port, const SockException &ex){
(*obj)->getChannel()->startSendRtp(*MediaSource::NullMediaSource, args, [cb, user_data](uint16_t local_port, const SockException &ex){
if (cb) {
cb(user_data, local_port, ex.getErrCode(), ex.what());
}
......
postman/ZLMediaKit.postman_collection.json
......@@ -1481,6 +1481,24 @@
"value": "0",
"description": "是否推送本地MP4录像，该参数非必选参数",
"disabled": true
},
{
"key": "use_ps",
"value": "1",
"description": "rtp打包采用ps还是es模式，默认采用ps模式，该参数非必选参数",
"disabled": true
},
{
"key": "pt",
"value": "96",
"description": "rtp payload type，默认96，该参数非必选参数",
"disabled": true
},
{
"key": "only_audio",
"value": "1",
"description": "rtp es方式打包时，是否只打包音频，该参数非必选参数",
"disabled": true
}
]
}
......
server/WebApi.cpp
......@@ -1104,19 +1104,26 @@ void installWebApi() {
if (!src) {
throw ApiRetException("该媒体流不存在", API::OtherFailed);
}
uint8_t pt = allArgs["pt"].empty() ? 96 : allArgs["pt"].as<int>();
bool use_ps = allArgs["use_ps"].empty() ? true : allArgs["use_ps"].as<bool>();
bool only_audio = allArgs["only_audio"].empty() ? true : allArgs["only_audio"].as<bool>();
TraceL << "pt "<<int(pt)<<" ps "<<use_ps<<" audio "<<only_audio;
//src_port为空时，则随机本地端口
src->startSendRtp(allArgs["dst_url"], allArgs["dst_port"], allArgs["ssrc"], allArgs["is_udp"], allArgs["src_port"], [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable{
MediaSourceEvent::SendRtpArgs args;
args.dst_url = allArgs["dst_url"];
args.dst_port = allArgs["dst_port"];
args.ssrc = allArgs["ssrc"];
args.is_udp = allArgs["is_udp"];
args.src_port = allArgs["src_port"];
args.pt = allArgs["pt"].empty() ? 96 : allArgs["pt"].as<int>();
args.use_ps = allArgs["use_ps"].empty() ? true : allArgs["use_ps"].as<bool>();
args.only_audio = allArgs["only_audio"].empty() ? false : allArgs["only_audio"].as<bool>();
TraceL << "pt " << int(args.pt) << " ps " << args.use_ps << " audio " << args.only_audio;
src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable {
if (ex) {
val["code"] = API::OtherFailed;
val["msg"] = ex.what();
}
val["local_port"] = local_port;
invoker(200, headerOut, val.toStyledString());
},pt,use_ps,only_audio);
});
});
api_regist("/index/api/stopSendRtp",[](API_ARGS_MAP){
......
src/Common/MediaSource.cpp
......@@ -237,13 +237,13 @@ bool MediaSource::isRecording(Recorder::type type){
return listener->isRecording(*this, type);
}
void MediaSource::startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(uint16_t local_port, const SockException &ex)> &cb,uint8_t pt, bool use_ps,bool only_audio){
void MediaSource::startSendRtp(const MediaSourceEvent::SendRtpArgs &args, const std::function<void(uint16_t, const toolkit::SockException &)> cb) {
auto listener = _listener.lock();
if (!listener) {
cb(0, SockException(Err_other, "尚未设置事件监听器"));
return;
}
return listener->startSendRtp(*this, dst_url, dst_port, ssrc, is_udp, src_port, cb, pt, use_ps, only_audio);
return listener->startSendRtp(*this, args, cb);
}
bool MediaSource::stopSendRtp(const string &ssrc) {
......@@ -720,12 +720,12 @@ vector<Track::Ptr> MediaSourceEventInterceptor::getMediaTracks(MediaSource &send
return listener->getMediaTracks(sender, trackReady);
}
void MediaSourceEventInterceptor::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(uint16_t local_port, const SockException &ex)> &cb, uint8_t pt, bool use_ps,bool only_audio ){
void MediaSourceEventInterceptor::startSendRtp(MediaSource &sender, const MediaSourceEvent::SendRtpArgs &args, const std::function<void(uint16_t, const toolkit::SockException &)> cb) {
auto listener = _listener.lock();
if (listener) {
listener->startSendRtp(sender, dst_url, dst_port, ssrc, is_udp, src_port, cb, pt, use_ps, only_audio);
listener->startSendRtp(sender, args, cb);
} else {
MediaSourceEvent::startSendRtp(sender, dst_url, dst_port, ssrc, is_udp, src_port, cb, pt, use_ps, only_audio);
MediaSourceEvent::startSendRtp(sender, args, cb);
}
}
......
src/Common/MediaSource.h
......@@ -50,7 +50,7 @@ enum class MediaOriginType : uint8_t {
std::string getOriginTypeString(MediaOriginType type);
class MediaSource;
class MediaSourceEvent{
class MediaSourceEvent {
public:
friend class MediaSource;
MediaSourceEvent(){};
......@@ -85,8 +85,29 @@ public:
virtual bool isRecording(MediaSource &sender, Recorder::type type) { return false; };
// 获取所有track相关信息
virtual std::vector<Track::Ptr> getMediaTracks(MediaSource &sender, bool trackReady = true) const { return std::vector<Track::Ptr>(); };
class SendRtpArgs {
public:
// 是否采用udp方式发送rtp
bool is_udp = true;
// rtp采用ps还是es方式
bool use_ps = true;
//发送es流时指定是否只发送纯音频流
bool only_audio = true;
// rtp payload type
uint8_t pt = 96;
// 指定rtp ssrc
std::string ssrc;
// 指定本地发送端口
uint16_t src_port = 0;
// 发送目标端口
uint16_t dst_port;
// 发送目标主机地址，可以是ip或域名
std::string dst_url;
};
// 开始发送ps-rtp
virtual void startSendRtp(MediaSource &sender, const std::string &dst_url, uint16_t dst_port, const std::string &ssrc, bool is_udp, uint16_t src_port, const std::function<void(uint16_t local_port, const toolkit::SockException &ex)> &cb, uint8_t pt=96, bool use_ps = true,bool only_audio = true) { cb(0, toolkit::SockException(toolkit::Err_other, "not implemented"));};
virtual void startSendRtp(MediaSource &sender, const SendRtpArgs &args, const std::function<void(uint16_t, const toolkit::SockException &)> cb) { cb(0, toolkit::SockException(toolkit::Err_other, "not implemented"));};
// 停止发送ps-rtp
virtual bool stopSendRtp(MediaSource &sender, const std::string &ssrc) {return false; }
......@@ -117,7 +138,7 @@ public:
bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const std::string &custom_path, size_t max_second) override;
bool isRecording(MediaSource &sender, Recorder::type type) override;
std::vector<Track::Ptr> getMediaTracks(MediaSource &sender, bool trackReady = true) const override;
void startSendRtp(MediaSource &sender, const std::string &dst_url, uint16_t dst_port, const std::string &ssrc, bool is_udp, uint16_t src_port, const std::function<void(uint16_t local_port, const toolkit::SockException &ex)> &cb, uint8_t pt=96, bool use_ps = true,bool only_audio = true) override;
void startSendRtp(MediaSource &sender, const SendRtpArgs &args, const std::function<void(uint16_t, const toolkit::SockException &)> cb) override;
bool stopSendRtp(MediaSource &sender, const std::string &ssrc) override;
private:
......@@ -269,7 +290,7 @@ public:
// 获取录制状态
bool isRecording(Recorder::type type);
// 开始发送ps-rtp
void startSendRtp(const std::string &dst_url, uint16_t dst_port, const std::string &ssrc, bool is_udp, uint16_t src_port, const std::function<void(uint16_t local_port, const toolkit::SockException &ex)> &cb , uint8_t pt = 96, bool use_ps = true,bool only_audio = true);
void startSendRtp(const MediaSourceEvent::SendRtpArgs &args, const std::function<void(uint16_t, const toolkit::SockException &)> cb);
// 停止发送ps-rtp
bool stopSendRtp(const std::string &ssrc);
......
src/Common/MultiMediaSourceMuxer.cpp
......@@ -213,11 +213,11 @@ bool MultiMediaSourceMuxer::isRecording(MediaSource &sender, Recorder::type type
}
}
void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(uint16_t local_port, const SockException &ex)> &cb ,uint8_t pt, bool use_ps,bool only_audio){
void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const MediaSourceEvent::SendRtpArgs &args, const std::function<void(uint16_t, const toolkit::SockException &)> cb) {
#if defined(ENABLE_RTPPROXY)
RtpSender::Ptr rtp_sender = std::make_shared<RtpSender>(atoi(ssrc.data()),pt,use_ps,only_audio);
auto rtp_sender = std::make_shared<RtpSender>();
weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this();
rtp_sender->startSend(dst_url, dst_port, is_udp, src_port, [weak_self, rtp_sender, cb, ssrc](uint16_t local_port, const SockException &ex) {
rtp_sender->startSend(args, [args, weak_self, rtp_sender, cb](uint16_t local_port, const SockException &ex) {
cb(local_port, ex);
auto strong_self = weak_self.lock();
if (!strong_self || ex) {
......@@ -228,7 +228,7 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const string &dst_url, u
}
rtp_sender->addTrackCompleted();
lock_guard<mutex> lck(strong_self->_rtp_sender_mtx);
strong_self->_rtp_sender[ssrc] = rtp_sender;
strong_self->_rtp_sender[args.ssrc] = rtp_sender;
});
#else
cb(0, SockException(Err_other, "该功能未启用，编译时请打开ENABLE_RTPPROXY宏"));
......
src/Common/MultiMediaSourceMuxer.h
......@@ -134,7 +134,7 @@ public:
* @param is_udp 是否为udp
* @param cb 启动成功或失败回调
*/
void startSendRtp(MediaSource &sender, const std::string &dst_url, uint16_t dst_port, const std::string &ssrc, bool is_udp, uint16_t src_port, const std::function<void(uint16_t local_port, const toolkit::SockException &ex)> &cb ,uint8_t pt = 96, bool use_ps = true,bool only_audio = true) override;
void startSendRtp(MediaSource &sender, const MediaSourceEvent::SendRtpArgs &args, const std::function<void(uint16_t, const toolkit::SockException &)> cb) override;
/**
* 停止ps-rtp发送
......
src/Rtp/RtpCache.h
......@@ -51,7 +51,7 @@ protected:
class RtpCacheRaw : public RtpCache, public RawEncoderImp{
public:
RtpCacheRaw(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96,bool sendAudio = true) : RtpCache(std::move(cb)), RawEncoderImp(ssrc, payload_type,sendAudio) {};
RtpCacheRaw(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96, bool sendAudio = true) : RtpCache(std::move(cb)), RawEncoderImp(ssrc, payload_type,sendAudio) {};
~RtpCacheRaw() override = default;
protected:
......
src/Rtp/RtpSender.cpp
......@@ -19,37 +19,28 @@ using namespace toolkit;
namespace mediakit{
RtpSender::RtpSender(uint32_t ssrc, uint8_t payload_type,bool use_ps, bool only_audio) {
void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const function<void(uint16_t local_port, const SockException &ex)> &cb){
_args = args;
_poller = EventPollerPool::Instance().getPoller();
if (use_ps) {
_interface = std::make_shared<RtpCachePS>(
[this](std::shared_ptr<List<Buffer::Ptr>> list) { onFlushRtpList(std::move(list)); }, ssrc, payload_type);
}else{
_interface = std::make_shared<RtpCacheRaw>(
[this](std::shared_ptr<List<Buffer::Ptr>> list) { onFlushRtpList(std::move(list)); }, ssrc, payload_type,only_audio);
auto lam = [this](std::shared_ptr<List<Buffer::Ptr>> list) { onFlushRtpList(std::move(list)); };
if (args.use_ps) {
_interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt);
} else {
_interface = std::make_shared<RtpCacheRaw>(lam, atoi(args.ssrc.data()), args.pt, args.only_audio);
}
}
RtpSender::~RtpSender() {}
void RtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp, uint16_t src_port, const function<void(uint16_t local_port, const SockException &ex)> &cb){
_is_udp = is_udp;
_socket = Socket::createSocket(_poller, false);
_dst_url = dst_url;
_dst_port = dst_port;
_src_port = src_port;
weak_ptr<RtpSender> weak_self = shared_from_this();
if (is_udp) {
_socket->bindUdpSock(src_port);
if (args.is_udp) {
_socket->bindUdpSock(args.src_port);
auto poller = _poller;
auto local_port = _socket->get_local_port();
WorkThreadPool::Instance().getPoller()->async([cb, dst_url, dst_port, weak_self, poller, local_port]() {
WorkThreadPool::Instance().getPoller()->async([cb, args, weak_self, poller, local_port]() {
struct sockaddr addr;
//切换线程目的是为了dns解析放在后台线程执行
if (!SockUtil::getDomainIP(dst_url.data(), dst_port, addr)) {
poller->async([dst_url, cb, local_port]() {
if (!SockUtil::getDomainIP(args.dst_url.data(), args.dst_port, addr)) {
poller->async([args, cb, local_port]() {
//切回自己的线程
cb(local_port, SockException(Err_dns, StrPrinter << "dns解析域名失败:" << dst_url));
cb(local_port, SockException(Err_dns, StrPrinter << "dns解析域名失败:" << args.dst_url));
});
return;
}
......@@ -66,7 +57,7 @@ void RtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp,
});
});
} else {
_socket->connect(dst_url, dst_port, [cb, weak_self](const SockException &err) {
_socket->connect(args.dst_url, args.dst_port, [cb, weak_self](const SockException &err) {
auto strong_self = weak_self.lock();
if (strong_self) {
if (!err) {
......@@ -77,7 +68,7 @@ void RtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp,
} else {
cb(0, err);
}
}, 5.0F, "0.0.0.0", src_port);
}, 5.0F, "0.0.0.0", args.src_port);
}
}
......@@ -85,7 +76,7 @@ void RtpSender::onConnect(){
_is_connect = true;
//加大发送缓存,防止udp丢包之类的问题
SockUtil::setSendBuf(_socket->rawFD(), 4 * 1024 * 1024);
if (!_is_udp) {
if (!_args.is_udp) {
//关闭tcp no_delay并开启MSG_MORE, 提高发送性能
SockUtil::setNoDelay(_socket->rawFD(), false);
_socket->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
......@@ -99,8 +90,8 @@ void RtpSender::onConnect(){
}
});
//获取本地端口，断开重连后确保端口不变
_src_port = _socket->get_local_port();
InfoL << "开始发送 rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _is_udp;
_args.src_port = _socket->get_local_port();
InfoL << "开始发送 rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _args.is_udp;
}
bool RtpSender::addTrack(const Track::Ptr &track){
......@@ -128,7 +119,7 @@ void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr> > rtp_list) {
return;
}
auto is_udp = _is_udp;
auto is_udp = _args.is_udp;
auto socket = _socket;
_poller->async([rtp_list, is_udp, socket]() {
size_t i = 0;
......@@ -150,9 +141,9 @@ void RtpSender::onErr(const SockException &ex, bool is_connect) {
//监听socket断开事件，方便重连
if (is_connect) {
WarnL << "重连" << _dst_url << ":" << _dst_port << "失败, 原因为:" << ex.what();
WarnL << "重连" << _args.dst_url << ":" << _args.dst_port << "失败, 原因为:" << ex.what();
} else {
WarnL << "停止发送 rtp:" << _dst_url << ":" << _dst_port << ", 原因为:" << ex.what();
WarnL << "停止发送 rtp:" << _args.dst_url << ":" << _args.dst_port << ", 原因为:" << ex.what();
}
weak_ptr<RtpSender> weak_self = shared_from_this();
......@@ -161,7 +152,7 @@ void RtpSender::onErr(const SockException &ex, bool is_connect) {
if (!strong_self) {
return false;
}
strong_self->startSend(strong_self->_dst_url, strong_self->_dst_port, strong_self->_is_udp, strong_self->_src_port, [weak_self](uint16_t local_port, const SockException &ex){
strong_self->startSend(strong_self->_args, [weak_self](uint16_t local_port, const SockException &ex){
auto strong_self = weak_self.lock();
if (strong_self && ex) {
//连接失败且本对象未销毁，那么重试连接
......
src/Rtp/RtpSender.h
......@@ -21,25 +21,15 @@ class RtpSender : public MediaSinkInterface, public std::enable_shared_from_this
public:
typedef std::shared_ptr<RtpSender> Ptr;
~RtpSender() override;
/**
* 构造函数，创建GB28181 RTP发送客户端
* @param ssrc rtp的ssrc
* @param payload_type 国标中ps-rtp的pt一般为96
* @param use_ps 是否打包为PS然后发送
* @param only_audio use_ps 为false 时有效，指定发送音频还是视频
*/
RtpSender(uint32_t ssrc, uint8_t payload_type = 96,bool use_ps = true,bool only_audio = true);
RtpSender() = default;
~RtpSender() override = default;
/**
* 开始发送ps-rtp包
* @param dst_url 目标ip或域名
* @param dst_port 目标端口
* @param is_udp 是否采用udp方式发送rtp
* @param args 发送参数
* @param cb 连接目标端口是否成功的回调
*/
void startSend(const std::string &dst_url, uint16_t dst_port, bool is_udp, uint16_t src_port, const std::function<void(uint16_t local_port, const toolkit::SockException &ex)> &cb);
void startSend(const MediaSourceEvent::SendRtpArgs &args, const std::function<void(uint16_t local_port, const toolkit::SockException &ex)> &cb);
/**
* 输入帧数据
......@@ -72,11 +62,8 @@ private:
void onErr(const toolkit::SockException &ex, bool is_connect = false);
private:
bool _is_udp;
bool _is_connect = false;
std::string _dst_url;
uint16_t _dst_port;
uint16_t _src_port;
MediaSourceEvent::SendRtpArgs _args;
toolkit::Socket::Ptr _socket;
toolkit::EventPoller::Ptr _poller;
toolkit::Timer::Ptr _connect_timer;
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论