feat: update negotiateSdp and WebRtcArgs (#3371)

- update negotiateSdp
- update HttpAllArgs and alias
- update onRtcConfigure
- define setWebRtcArgs, handle set_webrtc_cands and setLocalIp

---------

Co-authored-by: xiongziliang <771730766@qq.com>
Co-authored-by: KkemChen <kkemchen@qq.com>
This commit is contained in:
johzzy 2024-03-23 11:46:30 -03:00 committed by GitHub
parent 2e2823d4cf
commit 029813402d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 171 additions and 219 deletions

View File

@ -304,10 +304,10 @@ API_EXPORT void API_CALL mk_webrtc_get_answer_sdp2(void *user_data, on_user_data
std::string offer_str = offer; std::string offer_str = offer;
std::shared_ptr<void> ptr(user_data, user_data_free ? user_data_free : [](void *) {}); std::shared_ptr<void> ptr(user_data, user_data_free ? user_data_free : [](void *) {});
auto args = std::make_shared<WebRtcArgsUrl>(url); auto args = std::make_shared<WebRtcArgsUrl>(url);
WebRtcPluginManager::Instance().getAnswerSdp(*session, type, *args, WebRtcPluginManager::Instance().negotiateSdp(*session, type, *args, [offer_str, session, ptr, cb, args](const WebRtcInterface &exchanger) mutable {
[offer_str, session, ptr, cb](const WebRtcInterface &exchanger) mutable { auto &handler = const_cast<WebRtcInterface &>(exchanger);
try { try {
auto sdp_answer = exchangeSdp(exchanger, offer_str); auto sdp_answer = handler.getAnswerSdp(offer_str);
cb(ptr.get(), sdp_answer.data(), nullptr); cb(ptr.get(), sdp_answer.data(), nullptr);
} catch (std::exception &ex) { } catch (std::exception &ex) {
cb(ptr.get(), nullptr, ex.what()); cb(ptr.get(), nullptr, ex.what());

View File

@ -119,7 +119,7 @@ static HttpApi toApi(const function<void(API_ARGS_MAP_ASYNC)> &cb) {
//参数解析成map //参数解析成map
auto args = getAllArgs(parser); auto args = getAllArgs(parser);
cb(sender, headerOut, HttpAllArgs<decltype(args)>(parser, args), val, invoker); cb(sender, headerOut, ArgsMap(parser, args), val, invoker);
}; };
} }
@ -147,7 +147,7 @@ static HttpApi toApi(const function<void(API_ARGS_JSON_ASYNC)> &cb) {
Json::Reader reader; Json::Reader reader;
reader.parse(parser.content(), args); reader.parse(parser.content(), args);
cb(sender, headerOut, HttpAllArgs<decltype(args)>(parser, args), val, invoker); cb(sender, headerOut, ArgsJson(parser, args), val, invoker);
}; };
} }
@ -167,7 +167,7 @@ static HttpApi toApi(const function<void(API_ARGS_STRING_ASYNC)> &cb) {
Json::Value val; Json::Value val;
val["code"] = API::Success; val["code"] = API::Success;
cb(sender, headerOut, HttpAllArgs<string>(parser, (string &)parser.content()), val, invoker); cb(sender, headerOut, ArgsString(parser, (string &)parser.content()), val, invoker);
}; };
} }
@ -662,13 +662,6 @@ void addStreamPusherProxy(const string &schema,
pusher->publish(url); pusher->publish(url);
} }
template <typename Type>
static void getArgsValue(const HttpAllArgs<ApiArgsType> &allArgs, const string &key, Type &value) {
auto val = allArgs[key];
if (!val.empty()) {
value = (Type)val;
}
}
/** /**
* api接口 * api接口
@ -735,7 +728,7 @@ void installWebApi() {
CHECK_SECRET(); CHECK_SECRET();
auto &ini = mINI::Instance(); auto &ini = mINI::Instance();
int changed = API::Success; int changed = API::Success;
for (auto &pr : allArgs.getArgs()) { for (auto &pr : allArgs.args) {
if (ini.find(pr.first) == ini.end()) { if (ini.find(pr.first) == ini.end()) {
#if 1 #if 1
//没有这个key //没有这个key
@ -1093,7 +1086,7 @@ void installWebApi() {
CHECK_ARGS("vhost","app","stream","url"); CHECK_ARGS("vhost","app","stream","url");
mINI args; mINI args;
for (auto &pr : allArgs.getArgs()) { for (auto &pr : allArgs.args) {
args.emplace(pr.first, pr.second); args.emplace(pr.first, pr.second);
} }
@ -1190,7 +1183,7 @@ void installWebApi() {
//测试url http://127.0.0.1/index/api/downloadBin //测试url http://127.0.0.1/index/api/downloadBin
api_regist("/index/api/downloadBin",[](API_ARGS_MAP_ASYNC){ api_regist("/index/api/downloadBin",[](API_ARGS_MAP_ASYNC){
CHECK_SECRET(); CHECK_SECRET();
invoker.responseFile(allArgs.getParser().getHeader(),StrCaseMap(),exePath()); invoker.responseFile(allArgs.parser.getHeader(), StrCaseMap(), exePath());
}); });
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
@ -1697,7 +1690,7 @@ void installWebApi() {
//截图存在,且未过期,那么返回之 //截图存在,且未过期,那么返回之
res_old_snap = true; res_old_snap = true;
responseSnap(path, allArgs.getParser().getHeader(), invoker); responseSnap(path, allArgs.parser.getHeader(), invoker);
//中断遍历 //中断遍历
return false; return false;
}); });
@ -1728,7 +1721,7 @@ void installWebApi() {
File::delete_file(new_snap); File::delete_file(new_snap);
rename(new_snap_tmp.data(), new_snap.data()); rename(new_snap_tmp.data(), new_snap.data());
} }
responseSnap(new_snap, allArgs.getParser().getHeader(), invoker, err_msg); responseSnap(new_snap, allArgs.parser.getHeader(), invoker, err_msg);
}); });
}); });
@ -1743,7 +1736,7 @@ void installWebApi() {
#ifdef ENABLE_WEBRTC #ifdef ENABLE_WEBRTC
class WebRtcArgsImp : public WebRtcArgs { class WebRtcArgsImp : public WebRtcArgs {
public: public:
WebRtcArgsImp(const HttpAllArgs<string> &args, std::string session_id) WebRtcArgsImp(const ArgsString &args, std::string session_id)
: _args(args) : _args(args)
, _session_id(std::move(session_id)) {} , _session_id(std::move(session_id)) {}
~WebRtcArgsImp() override = default; ~WebRtcArgsImp() override = default;
@ -1761,40 +1754,26 @@ void installWebApi() {
CHECK_ARGS("app", "stream"); CHECK_ARGS("app", "stream");
return StrPrinter << "rtc://" << _args["Host"] << "/" << _args["app"] << "/" return StrPrinter << "rtc://" << _args["Host"] << "/" << _args["app"] << "/"
<< _args["stream"] << "?" << _args.getParser().params() + "&session=" + _session_id; << _args["stream"] << "?" << _args.parser.params() + "&session=" + _session_id;
} }
private: private:
HttpAllArgs<string> _args; ArgsString _args;
std::string _session_id; std::string _session_id;
}; };
api_regist("/index/api/webrtc",[](API_ARGS_STRING_ASYNC){ api_regist("/index/api/webrtc",[](API_ARGS_STRING_ASYNC){
CHECK_ARGS("type"); CHECK_ARGS("type");
auto type = allArgs["type"]; auto type = allArgs["type"];
auto offer = allArgs.getArgs(); auto offer = allArgs.args;
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 &session = static_cast<Session&>(sender);
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, [invoker, val, offer, headerOut, localIp](const WebRtcInterface &exchanger) mutable { WebRtcPluginManager::Instance().negotiateSdp(session, type, *args, [invoker, val, offer, headerOut, args](const WebRtcInterface &exchanger) mutable {
//设置返回类型 auto &handler = const_cast<WebRtcInterface &>(exchanger);
headerOut["Content-Type"] = HttpFileManager::getContentType(".json");
//设置跨域
headerOut["Access-Control-Allow-Origin"] = "*";
try { try {
setLocalIp(exchanger,localIp); val["sdp"] = handler.getAnswerSdp(offer);
val["sdp"] = exchangeSdp(exchanger, offer);
val["id"] = exchanger.getIdentifier(); val["id"] = exchanger.getIdentifier();
val["type"] = "answer"; val["type"] = "answer";
invoker(200, headerOut, val.toStyledString()); invoker(200, headerOut, val.toStyledString());
@ -1808,21 +1787,19 @@ void installWebApi() {
static constexpr char delete_webrtc_url [] = "/index/api/delete_webrtc"; static constexpr char delete_webrtc_url [] = "/index/api/delete_webrtc";
static auto whip_whep_func = [](const char *type, API_ARGS_STRING_ASYNC) { static auto whip_whep_func = [](const char *type, API_ARGS_STRING_ASYNC) {
auto offer = allArgs.getArgs(); auto offer = allArgs.args;
CHECK(!offer.empty(), "http body(webrtc offer sdp) is empty"); CHECK(!offer.empty(), "http body(webrtc offer sdp) is empty");
auto &session = static_cast<Session&>(sender); auto &session = static_cast<Session&>(sender);
auto location = std::string("http") + (session.overSsl() ? "s" : "") + "://" + allArgs["host"] + delete_webrtc_url; auto location = std::string(session.overSsl() ? "https://" : "http://") + allArgs["host"] + delete_webrtc_url;
auto args = std::make_shared<WebRtcArgsImp>(allArgs, sender.getIdentifier()); auto args = std::make_shared<WebRtcArgsImp>(allArgs, sender.getIdentifier());
WebRtcPluginManager::Instance().getAnswerSdp(session, type, *args, WebRtcPluginManager::Instance().negotiateSdp(session, type, *args, [invoker, offer, headerOut, location, args](const WebRtcInterface &exchanger) mutable {
[invoker, offer, headerOut, location](const WebRtcInterface &exchanger) mutable { auto &handler = const_cast<WebRtcInterface &>(exchanger);
// 设置跨域
headerOut["Access-Control-Allow-Origin"] = "*";
try { try {
// 设置返回类型 // 设置返回类型
headerOut["Content-Type"] = "application/sdp"; headerOut["Content-Type"] = "application/sdp";
headerOut["Location"] = location + "?id=" + exchanger.getIdentifier() + "&token=" + exchanger.deleteRandStr(); headerOut["Location"] = location + "?id=" + exchanger.getIdentifier() + "&token=" + exchanger.deleteRandStr();
invoker(201, headerOut, exchangeSdp(exchanger, offer)); invoker(201, headerOut, handler.getAnswerSdp(offer));
} catch (std::exception &ex) { } catch (std::exception &ex) {
headerOut["Content-Type"] = "text/plain"; headerOut["Content-Type"] = "text/plain";
invoker(406, headerOut, ex.what()); invoker(406, headerOut, ex.what());
@ -1835,7 +1812,7 @@ void installWebApi() {
api_regist(delete_webrtc_url, [](API_ARGS_MAP_ASYNC) { api_regist(delete_webrtc_url, [](API_ARGS_MAP_ASYNC) {
CHECK_ARGS("id", "token"); CHECK_ARGS("id", "token");
CHECK(allArgs.getParser().method() == "DELETE", "http method is not DELETE: " + allArgs.getParser().method()); CHECK(allArgs.parser.method() == "DELETE", "http method is not DELETE: " + allArgs.parser.method());
auto obj = WebRtcTransportManager::Instance().getItem(allArgs["id"]); auto obj = WebRtcTransportManager::Instance().getItem(allArgs["id"]);
if (!obj) { if (!obj) {
invoker(404, headerOut, "id not found"); invoker(404, headerOut, "id not found");
@ -1921,11 +1898,11 @@ void installWebApi() {
if (!save_name.empty()) { if (!save_name.empty()) {
res_header.emplace("Content-Disposition", "attachment;filename=\"" + save_name + "\""); res_header.emplace("Content-Disposition", "attachment;filename=\"" + save_name + "\"");
} }
invoker.responseFile(allArgs.getParser().getHeader(), res_header, allArgs["file_path"]); invoker.responseFile(allArgs.parser.getHeader(), res_header, allArgs["file_path"]);
} }
}; };
bool flag = NOTICE_EMIT(BroadcastHttpAccessArgs, Broadcast::kBroadcastHttpAccess, allArgs.getParser(), file_path, false, file_invoker, sender); bool flag = NOTICE_EMIT(BroadcastHttpAccessArgs, Broadcast::kBroadcastHttpAccess, allArgs.parser, file_path, false, file_invoker, sender);
if (!flag) { if (!flag) {
// 文件下载鉴权事件无人监听,不允许下载 // 文件下载鉴权事件无人监听,不允许下载
invoker(401, StrCaseMap {}, "None http access event listener"); invoker(401, StrCaseMap {}, "None http access event listener");

View File

@ -115,72 +115,41 @@ std::string getValue(const mediakit::Parser &parser, Args &args, const First &fi
template<typename Args> template<typename Args>
class HttpAllArgs { class HttpAllArgs {
mediakit::Parser* _parser = nullptr;
Args* _args = nullptr;
public: public:
HttpAllArgs(const mediakit::Parser &parser, Args &args) { const mediakit::Parser& parser;
_get_args = [&args]() { Args& args;
return (void *) &args;
};
_get_parser = [&parser]() -> const mediakit::Parser & {
return parser;
};
_get_value = [](HttpAllArgs &that, const std::string &key) {
return getValue(that.getParser(), that.getArgs(), key);
};
_clone = [&](HttpAllArgs &that) {
that._get_args = [args]() {
return (void *) &args;
};
that._get_parser = [parser]() -> const mediakit::Parser & {
return parser;
};
that._get_value = [](HttpAllArgs &that, const std::string &key) {
return getValue(that.getParser(), that.getArgs(), key);
};
that._cache_able = true;
};
}
HttpAllArgs(const HttpAllArgs &that) { HttpAllArgs(const mediakit::Parser &p, Args &a): parser(p), args(a) {}
if (that._cache_able) {
_get_args = that._get_args; HttpAllArgs(const HttpAllArgs &that): _parser(new mediakit::Parser(that.parser)),
_get_parser = that._get_parser; _args(new Args(that.args)),
_get_value = that._get_value; parser(*_parser), args(*_args) {}
_cache_able = true; ~HttpAllArgs() {
} else { if (_parser) {
that._clone(*this); delete _parser;
}
if (_args) {
delete _args;
} }
} }
template<typename Key> template<typename Key>
toolkit::variant operator[](const Key &key) const { toolkit::variant operator[](const Key &key) const {
return (toolkit::variant)_get_value(*(HttpAllArgs*)this, key); return (toolkit::variant)getValue(parser, args, key);
} }
const mediakit::Parser &getParser() const {
return _get_parser();
}
Args &getArgs() {
return *((Args *) _get_args());
}
const Args &getArgs() const {
return *((Args *) _get_args());
}
private:
bool _cache_able = false;
std::function<void *() > _get_args;
std::function<const mediakit::Parser &() > _get_parser;
std::function<std::string(HttpAllArgs &that, const std::string &key)> _get_value;
std::function<void(HttpAllArgs &that) > _clone;
}; };
#define API_ARGS_MAP toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const HttpAllArgs<ApiArgsType> &allArgs, Json::Value &val using ArgsMap = HttpAllArgs<ApiArgsType>;
using ArgsJson = HttpAllArgs<Json::Value>;
using ArgsString = HttpAllArgs<std::string>;
#define API_ARGS_MAP toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsMap &allArgs, Json::Value &val
#define API_ARGS_MAP_ASYNC API_ARGS_MAP, const mediakit::HttpSession::HttpResponseInvoker &invoker #define API_ARGS_MAP_ASYNC API_ARGS_MAP, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_JSON toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const HttpAllArgs<Json::Value> &allArgs, Json::Value &val #define API_ARGS_JSON toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsJson &allArgs, Json::Value &val
#define API_ARGS_JSON_ASYNC API_ARGS_JSON, const mediakit::HttpSession::HttpResponseInvoker &invoker #define API_ARGS_JSON_ASYNC API_ARGS_JSON, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_STRING toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const HttpAllArgs<std::string> &allArgs, Json::Value &val #define API_ARGS_STRING toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsString &allArgs, Json::Value &val
#define API_ARGS_STRING_ASYNC API_ARGS_STRING, const mediakit::HttpSession::HttpResponseInvoker &invoker #define API_ARGS_STRING_ASYNC API_ARGS_STRING, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_VALUE sender, headerOut, allArgs, val #define API_ARGS_VALUE sender, headerOut, allArgs, val

View File

@ -136,6 +136,15 @@ private:
toolkit::Timer::Ptr _async_close_timer; toolkit::Timer::Ptr _async_close_timer;
}; };
template <typename MAP, typename KEY, typename TYPE>
static void getArgsValue(const MAP &allArgs, const KEY &key, TYPE &value) {
auto val = ((MAP &)allArgs)[key];
if (!val.empty()) {
value = (TYPE)val;
}
}
class ProtocolOption { class ProtocolOption {
public: public:
ProtocolOption(); ProtocolOption();
@ -243,15 +252,6 @@ public:
GET_OPT_VALUE(stream_replace); GET_OPT_VALUE(stream_replace);
GET_OPT_VALUE(max_track); GET_OPT_VALUE(max_track);
} }
private:
template <typename MAP, typename KEY, typename TYPE>
static void getArgsValue(const MAP &allArgs, const KEY &key, TYPE &value) {
auto val = ((MAP &)allArgs)[key];
if (!val.empty()) {
value = (TYPE)val;
}
}
}; };
//该对象用于拦截感兴趣的MediaSourceEvent事件 //该对象用于拦截感兴趣的MediaSourceEvent事件

View File

@ -30,7 +30,7 @@ struct StrCaseCompare {
class StrCaseMap : public std::multimap<std::string, std::string, StrCaseCompare> { class StrCaseMap : public std::multimap<std::string, std::string, StrCaseCompare> {
public: public:
using Super = multimap<std::string, std::string, StrCaseCompare>; using Super = std::multimap<std::string, std::string, StrCaseCompare>;
std::string &operator[](const std::string &k) { std::string &operator[](const std::string &k) {
auto it = find(k); auto it = find(k);

View File

@ -27,7 +27,6 @@ protected:
void onRtp(const char *buf, size_t len, uint64_t stamp_ms) override; void onRtp(const char *buf, size_t len, uint64_t stamp_ms) override;
void onRtcp(const char *buf, size_t len) override; void onRtcp(const char *buf, size_t len) override;
void onRecvRtp(MediaTrack &track, const std::string &rid, RtpPacket::Ptr rtp) override {};
void onBeforeEncryptRtp(const char *buf, int &len, void *ctx) override {}; void onBeforeEncryptRtp(const char *buf, int &len, void *ctx) override {};
void onBeforeEncryptRtcp(const char *buf, int &len, void *ctx) override {}; void onBeforeEncryptRtcp(const char *buf, int &len, void *ctx) override {};

View File

@ -17,9 +17,8 @@ namespace mediakit {
WebRtcPlayer::Ptr WebRtcPlayer::create(const EventPoller::Ptr &poller, WebRtcPlayer::Ptr WebRtcPlayer::create(const EventPoller::Ptr &poller,
const RtspMediaSource::Ptr &src, const RtspMediaSource::Ptr &src,
const MediaInfo &info, const MediaInfo &info) {
bool preferred_tcp) { WebRtcPlayer::Ptr ret(new WebRtcPlayer(poller, src, info), [](WebRtcPlayer *ptr) {
WebRtcPlayer::Ptr ret(new WebRtcPlayer(poller, src, info, preferred_tcp), [](WebRtcPlayer *ptr) {
ptr->onDestory(); ptr->onDestory();
delete ptr; delete ptr;
}); });
@ -29,8 +28,7 @@ WebRtcPlayer::Ptr WebRtcPlayer::create(const EventPoller::Ptr &poller,
WebRtcPlayer::WebRtcPlayer(const EventPoller::Ptr &poller, WebRtcPlayer::WebRtcPlayer(const EventPoller::Ptr &poller,
const RtspMediaSource::Ptr &src, const RtspMediaSource::Ptr &src,
const MediaInfo &info, const MediaInfo &info) : WebRtcTransportImp(poller) {
bool preferred_tcp) : WebRtcTransportImp(poller,preferred_tcp) {
_media_info = info; _media_info = info;
_play_src = src; _play_src = src;
CHECK(src); CHECK(src);

View File

@ -19,7 +19,7 @@ namespace mediakit {
class WebRtcPlayer : public WebRtcTransportImp { class WebRtcPlayer : public WebRtcTransportImp {
public: public:
using Ptr = std::shared_ptr<WebRtcPlayer>; using Ptr = std::shared_ptr<WebRtcPlayer>;
static Ptr create(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src, const MediaInfo &info, bool preferred_tcp = false); static Ptr create(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src, const MediaInfo &info);
MediaInfo getMediaInfo() { return _media_info; } MediaInfo getMediaInfo() { return _media_info; }
protected: protected:
@ -27,10 +27,9 @@ protected:
void onStartWebRTC() override; void onStartWebRTC() override;
void onDestory() override; void onDestory() override;
void onRtcConfigure(RtcConfigure &configure) const override; void onRtcConfigure(RtcConfigure &configure) const override;
void onRecvRtp(MediaTrack &track, const std::string &rid, RtpPacket::Ptr rtp) override {};
private: private:
WebRtcPlayer(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src, const MediaInfo &info, bool preferred_tcp); WebRtcPlayer(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src, const MediaInfo &info);
private: private:
//媒体相关元数据 //媒体相关元数据

View File

@ -20,9 +20,8 @@ WebRtcPusher::Ptr WebRtcPusher::create(const EventPoller::Ptr &poller,
const RtspMediaSource::Ptr &src, const RtspMediaSource::Ptr &src,
const std::shared_ptr<void> &ownership, const std::shared_ptr<void> &ownership,
const MediaInfo &info, const MediaInfo &info,
const ProtocolOption &option, const ProtocolOption &option) {
bool preferred_tcp) { WebRtcPusher::Ptr ret(new WebRtcPusher(poller, src, ownership, info, option), [](WebRtcPusher *ptr) {
WebRtcPusher::Ptr ret(new WebRtcPusher(poller, src, ownership, info, option,preferred_tcp), [](WebRtcPusher *ptr) {
ptr->onDestory(); ptr->onDestory();
delete ptr; delete ptr;
}); });
@ -34,8 +33,7 @@ WebRtcPusher::WebRtcPusher(const EventPoller::Ptr &poller,
const RtspMediaSource::Ptr &src, const RtspMediaSource::Ptr &src,
const std::shared_ptr<void> &ownership, const std::shared_ptr<void> &ownership,
const MediaInfo &info, const MediaInfo &info,
const ProtocolOption &option, const ProtocolOption &option) : WebRtcTransportImp(poller) {
bool preferred_tcp) : WebRtcTransportImp(poller,preferred_tcp) {
_media_info = info; _media_info = info;
_push_src = src; _push_src = src;
_push_src_ownership = ownership; _push_src_ownership = ownership;

View File

@ -20,8 +20,7 @@ class WebRtcPusher : public WebRtcTransportImp, public MediaSourceEvent {
public: public:
using Ptr = std::shared_ptr<WebRtcPusher>; using Ptr = std::shared_ptr<WebRtcPusher>;
static Ptr create(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src, static Ptr create(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src,
const std::shared_ptr<void> &ownership, const MediaInfo &info, const ProtocolOption &option, bool preferred_tcp = false); const std::shared_ptr<void> &ownership, const MediaInfo &info, const ProtocolOption &option);
protected: protected:
///////WebRtcTransportImp override/////// ///////WebRtcTransportImp override///////
@ -53,7 +52,7 @@ protected:
private: private:
WebRtcPusher(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src, WebRtcPusher(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src,
const std::shared_ptr<void> &ownership, const MediaInfo &info, const ProtocolOption &option, bool preferred_tcp); const std::shared_ptr<void> &ownership, const MediaInfo &info, const ProtocolOption &option);
private: private:
bool _simulcast = false; bool _simulcast = false;

View File

@ -378,6 +378,12 @@ void WebRtcTransport::setRemoteDtlsFingerprint(const RtcSession &remote) {
} }
void WebRtcTransport::onRtcConfigure(RtcConfigure &configure) const { void WebRtcTransport::onRtcConfigure(RtcConfigure &configure) const {
SdpAttrFingerprint fingerprint;
fingerprint.algorithm = _offer_sdp->media[0].fingerprint.algorithm;
fingerprint.hash = getFingerprint(fingerprint.algorithm, _dtls_transport);
configure.setDefaultSetting(
_ice_server->GetUsernameFragment(), _ice_server->GetPassword(), RtpDirection::sendrecv, fingerprint);
// 开启remb后关闭twcc因为开启twcc后remb无效 // 开启remb后关闭twcc因为开启twcc后remb无效
GET_CONFIG(size_t, remb_bit_rate, Rtc::kRembBitRate); GET_CONFIG(size_t, remb_bit_rate, Rtc::kRembBitRate);
configure.enableTWCC(!remb_bit_rate); configure.enableTWCC(!remb_bit_rate);
@ -407,12 +413,7 @@ std::string WebRtcTransport::getAnswerSdp(const string &offer) {
setRemoteDtlsFingerprint(*_offer_sdp); setRemoteDtlsFingerprint(*_offer_sdp);
//// sdp 配置 //// //// sdp 配置 ////
SdpAttrFingerprint fingerprint;
fingerprint.algorithm = _offer_sdp->media[0].fingerprint.algorithm;
fingerprint.hash = getFingerprint(fingerprint.algorithm, _dtls_transport);
RtcConfigure configure; RtcConfigure configure;
configure.setDefaultSetting(
_ice_server->GetUsernameFragment(), _ice_server->GetPassword(), RtpDirection::sendrecv, fingerprint);
onRtcConfigure(configure); onRtcConfigure(configure);
//// 生成answer sdp //// //// 生成answer sdp ////
@ -431,10 +432,6 @@ static bool isDtls(char *buf) {
return ((*buf > 19) && (*buf < 64)); return ((*buf > 19) && (*buf < 64));
} }
static string getPeerAddress(RTC::TransportTuple *tuple) {
return tuple->get_peer_ip();
}
void WebRtcTransport::inputSockData(char *buf, int len, RTC::TransportTuple *tuple) { void WebRtcTransport::inputSockData(char *buf, int len, RTC::TransportTuple *tuple) {
if (RTC::StunPacket::IsStun((const uint8_t *)buf, len)) { if (RTC::StunPacket::IsStun((const uint8_t *)buf, len)) {
std::unique_ptr<RTC::StunPacket> packet(RTC::StunPacket::Parse((const uint8_t *)buf, len)); std::unique_ptr<RTC::StunPacket> packet(RTC::StunPacket::Parse((const uint8_t *)buf, len));
@ -451,7 +448,7 @@ void WebRtcTransport::inputSockData(char *buf, int len, RTC::TransportTuple *tup
} }
if (isRtp(buf, len)) { if (isRtp(buf, len)) {
if (!_srtp_session_recv) { if (!_srtp_session_recv) {
WarnL << "received rtp packet when dtls not completed from:" << getPeerAddress(tuple); WarnL << "received rtp packet when dtls not completed from:" << tuple->get_peer_ip();
return; return;
} }
if (_srtp_session_recv->DecryptSrtp((uint8_t *)buf, &len)) { if (_srtp_session_recv->DecryptSrtp((uint8_t *)buf, &len)) {
@ -461,7 +458,7 @@ void WebRtcTransport::inputSockData(char *buf, int len, RTC::TransportTuple *tup
} }
if (isRtcp(buf, len)) { if (isRtcp(buf, len)) {
if (!_srtp_session_recv) { if (!_srtp_session_recv) {
WarnL << "received rtcp packet when dtls not completed from:" << getPeerAddress(tuple); WarnL << "received rtcp packet when dtls not completed from:" << tuple->get_peer_ip();
return; return;
} }
if (_srtp_session_recv->DecryptSrtcp((uint8_t *)buf, &len)) { if (_srtp_session_recv->DecryptSrtcp((uint8_t *)buf, &len)) {
@ -533,8 +530,7 @@ void WebRtcTransportImp::OnDtlsTransportApplicationDataReceived(const RTC::DtlsT
#endif #endif
} }
WebRtcTransportImp::WebRtcTransportImp(const EventPoller::Ptr &poller,bool preferred_tcp) WebRtcTransportImp::WebRtcTransportImp(const EventPoller::Ptr &poller) : WebRtcTransport(poller) {
: WebRtcTransport(poller), _preferred_tcp(preferred_tcp) {
InfoL << getIdentifier(); InfoL << getIdentifier();
} }
@ -674,7 +670,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() ? _localIp.empty() ? SockUtil::get_local_ip() : _localIp : extern_ips[0]; m.addr.address = extern_ips.empty() ? _local_ip.empty() ? SockUtil::get_local_ip() : _local_ip : 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;
@ -769,7 +765,7 @@ void WebRtcTransportImp::onRtcConfigure(RtcConfigure &configure) const {
return ret; return ret;
}); });
if (extern_ips.empty()) { if (extern_ips.empty()) {
std::string local_ip = _localIp.empty() ? SockUtil::get_local_ip() : _localIp; std::string local_ip = _local_ip.empty() ? SockUtil::get_local_ip() : _local_ip;
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 {
@ -783,12 +779,16 @@ void WebRtcTransportImp::onRtcConfigure(RtcConfigure &configure) const {
} }
} }
void WebRtcTransportImp::setIceCandidate(vector<SdpAttrCandidate> cands) { void WebRtcTransportImp::setPreferredTcp(bool flag) {
_cands = std::move(cands); _preferred_tcp = flag;
} }
void WebRtcTransportImp::setLocalIp(const std::string &localIp) { void WebRtcTransportImp::setLocalIp(std::string local_ip) {
_localIp = localIp; _local_ip = std::move(local_ip);
}
void WebRtcTransportImp::setIceCandidate(vector<SdpAttrCandidate> cands) {
_cands = std::move(cands);
} }
/////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////
@ -1278,21 +1278,14 @@ void WebRtcPluginManager::registerPlugin(const string &type, Plugin cb) {
_map_creator[type] = std::move(cb); _map_creator[type] = std::move(cb);
} }
std::string exchangeSdp(const WebRtcInterface &exchanger, const std::string& 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);
} }
void WebRtcPluginManager::getAnswerSdp(Session &sender, const string &type, const WebRtcArgs &args, const onCreateRtc &cb_in) { void WebRtcPluginManager::negotiateSdp(Session &sender, const string &type, const WebRtcArgs &args, const onCreateWebRtc &cb_in) {
onCreateRtc cb; onCreateWebRtc cb;
lock_guard<mutex> lck(_mtx_creator); lock_guard<mutex> lck(_mtx_creator);
if (_listener) { if (_listener) {
auto listener = _listener; auto listener = _listener;
@ -1308,21 +1301,19 @@ void WebRtcPluginManager::getAnswerSdp(Session &sender, const string &type, cons
auto it = _map_creator.find(type); auto it = _map_creator.find(type);
if (it == _map_creator.end()) { if (it == _map_creator.end()) {
cb(WebRtcException(SockException(Err_other, "the type can not supported"))); cb_in(WebRtcException(SockException(Err_other, "the type can not supported")));
return; return;
} }
it->second(sender, args, cb); it->second(sender, args, cb);
} }
void echo_plugin(Session &sender, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) { void echo_plugin(Session &sender, const WebRtcArgs &args, const onCreateWebRtc &cb) {
cb(*WebRtcEchoTest::create(EventPollerPool::Instance().getPoller())); cb(*WebRtcEchoTest::create(EventPollerPool::Instance().getPoller()));
} }
void push_plugin(Session &sender, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) { void push_plugin(Session &sender, const WebRtcArgs &args, const onCreateWebRtc &cb) {
MediaInfo info(args["url"]); MediaInfo info(args["url"]);
bool preferred_tcp = args["preferred_tcp"]; Broadcast::PublishAuthInvoker invoker = [cb, info](const string &err, const ProtocolOption &option) mutable {
Broadcast::PublishAuthInvoker invoker = [cb, info, preferred_tcp](const string &err, const ProtocolOption &option) mutable {
if (!err.empty()) { if (!err.empty()) {
cb(WebRtcException(SockException(Err_other, err))); cb(WebRtcException(SockException(Err_other, err)));
return; return;
@ -1361,7 +1352,7 @@ void push_plugin(Session &sender, const WebRtcArgs &args, const WebRtcPluginMana
push_src_ownership = push_src->getOwnership(); push_src_ownership = push_src->getOwnership();
push_src->setProtocolOption(option); push_src->setProtocolOption(option);
} }
auto rtc = WebRtcPusher::create(EventPollerPool::Instance().getPoller(), push_src, push_src_ownership, info, option, preferred_tcp); auto rtc = WebRtcPusher::create(EventPollerPool::Instance().getPoller(), push_src, push_src_ownership, info, option);
push_src->setListener(rtc); push_src->setListener(rtc);
cb(*rtc); cb(*rtc);
}; };
@ -1374,12 +1365,10 @@ void push_plugin(Session &sender, const WebRtcArgs &args, const WebRtcPluginMana
} }
} }
void play_plugin(Session &sender, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) { void play_plugin(Session &sender, const WebRtcArgs &args, const onCreateWebRtc &cb) {
MediaInfo info(args["url"]); MediaInfo info(args["url"]);
bool preferred_tcp = args["preferred_tcp"];
auto session_ptr = static_pointer_cast<Session>(sender.shared_from_this()); auto session_ptr = static_pointer_cast<Session>(sender.shared_from_this());
Broadcast::AuthInvoker invoker = [cb, info, session_ptr, preferred_tcp](const string &err) mutable { Broadcast::AuthInvoker invoker = [cb, info, session_ptr](const string &err) mutable {
if (!err.empty()) { if (!err.empty()) {
cb(WebRtcException(SockException(Err_other, err))); cb(WebRtcException(SockException(Err_other, err)));
return; return;
@ -1395,7 +1384,7 @@ void play_plugin(Session &sender, const WebRtcArgs &args, const WebRtcPluginMana
} }
// 还原成rtc目的是为了hook时识别哪种播放协议 // 还原成rtc目的是为了hook时识别哪种播放协议
info.schema = "rtc"; info.schema = "rtc";
auto rtc = WebRtcPlayer::create(EventPollerPool::Instance().getPoller(), src, info, preferred_tcp); auto rtc = WebRtcPlayer::create(EventPollerPool::Instance().getPoller(), src, info);
cb(*rtc); cb(*rtc);
}); });
}; };
@ -1408,14 +1397,35 @@ void play_plugin(Session &sender, const WebRtcArgs &args, const WebRtcPluginMana
} }
} }
static void set_webrtc_cands(const WebRtcArgs &args, const WebRtcInterface &rtc) { static void setWebRtcArgs(const WebRtcArgs &args, WebRtcInterface &rtc) {
{
static auto is_vaild_ip = [](const std::string &ip) -> bool {
int a, b, c, d;
return sscanf(ip.c_str(), "%d.%d.%d.%d", &a, &b, &c, &d) == 4;
};
std::string host = args["Host"];
if (!host.empty()) {
auto local_ip = host.substr(0, host.find(':'));
if (!is_vaild_ip(local_ip) || local_ip == "127.0.0.1") {
local_ip = "";
}
rtc.setLocalIp(std::move(local_ip));
}
}
bool preferred_tcp = args["preferred_tcp"];
{
rtc.setPreferredTcp(preferred_tcp);
}
{
vector<SdpAttrCandidate> cands; vector<SdpAttrCandidate> cands;
{ {
auto cand_str = trim(args["cand_udp"]); auto cand_str = trim(args["cand_udp"]);
auto ip_port = toolkit::split(cand_str, ":"); auto ip_port = toolkit::split(cand_str, ":");
if (ip_port.size() == 2) { if (ip_port.size() == 2) {
// udp优先 // udp优先
auto ice_cand = makeIceCandidate(ip_port[0], atoi(ip_port[1].data()), 120, "udp"); auto ice_cand = makeIceCandidate(ip_port[0], atoi(ip_port[1].data()), preferred_tcp ? 100 : 120, "udp");
cands.emplace_back(std::move(*ice_cand)); cands.emplace_back(std::move(*ice_cand));
} }
} }
@ -1424,23 +1434,26 @@ static void set_webrtc_cands(const WebRtcArgs &args, const WebRtcInterface &rtc)
auto ip_port = toolkit::split(cand_str, ":"); auto ip_port = toolkit::split(cand_str, ":");
if (ip_port.size() == 2) { if (ip_port.size() == 2) {
// tcp模式 // tcp模式
auto ice_cand = makeIceCandidate(ip_port[0], atoi(ip_port[1].data()), 100, "tcp"); auto ice_cand = makeIceCandidate(ip_port[0], atoi(ip_port[1].data()), preferred_tcp ? 120 : 100, "tcp");
cands.emplace_back(std::move(*ice_cand)); cands.emplace_back(std::move(*ice_cand));
} }
} }
if (!cands.empty()) { if (!cands.empty()) {
// udp优先 // udp优先
const_cast<WebRtcInterface &>(rtc).setIceCandidate(std::move(cands)); rtc.setIceCandidate(std::move(cands));
}
} }
} }
static onceToken s_rtc_auto_register([]() { static onceToken s_rtc_auto_register([]() {
#if !defined (NDEBUG)
// debug模式才开启echo插件
WebRtcPluginManager::Instance().registerPlugin("echo", echo_plugin); WebRtcPluginManager::Instance().registerPlugin("echo", echo_plugin);
#endif
WebRtcPluginManager::Instance().registerPlugin("push", push_plugin); WebRtcPluginManager::Instance().registerPlugin("push", push_plugin);
WebRtcPluginManager::Instance().registerPlugin("play", play_plugin); WebRtcPluginManager::Instance().registerPlugin("play", play_plugin);
WebRtcPluginManager::Instance().setListener([](Session &sender, const std::string &type, const WebRtcArgs &args, const WebRtcInterface &rtc) { WebRtcPluginManager::Instance().setListener([](Session &sender, const std::string &type, const WebRtcArgs &args, const WebRtcInterface &rtc) {
set_webrtc_cands(args, rtc); setWebRtcArgs(args, const_cast<WebRtcInterface&>(rtc));
}); });
}); });

View File

@ -35,6 +35,8 @@ extern const std::string kTcpPort;
extern const std::string kTimeOutSec; extern const std::string kTimeOutSec;
}//namespace RTC }//namespace RTC
class WebRtcArgs;
class WebRtcInterface { class WebRtcInterface {
public: public:
virtual ~WebRtcInterface() = default; virtual ~WebRtcInterface() = default;
@ -42,13 +44,10 @@ 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) {} virtual void setLocalIp(std::string localIp) {}
virtual void setPreferredTcp(bool flag) {}
}; };
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) {};
@ -88,7 +87,7 @@ public:
* @param offer offer sdp * @param offer offer sdp
* @return answer sdp * @return answer sdp
*/ */
std::string getAnswerSdp(const std::string &offer) override; std::string getAnswerSdp(const std::string &offer) override final;
/** /**
* id * id
@ -252,14 +251,16 @@ public:
void onSendRtp(const RtpPacket::Ptr &rtp, bool flush, bool rtx = false); void onSendRtp(const RtpPacket::Ptr &rtp, bool flush, bool rtx = false);
void createRtpChannel(const std::string &rid, uint32_t ssrc, MediaTrack &track); void createRtpChannel(const std::string &rid, uint32_t ssrc, MediaTrack &track);
void setIceCandidate(std::vector<SdpAttrCandidate> cands) override;
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; void setPreferredTcp(bool flag) override;
void setLocalIp(std::string local_ip) override;
void setIceCandidate(std::vector<SdpAttrCandidate> cands) 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);
void OnDtlsTransportApplicationDataReceived(const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) override; void OnDtlsTransportApplicationDataReceived(const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) override;
void onStartWebRTC() override; void onStartWebRTC() override;
void onSendSockData(Buffer::Ptr buf, bool flush = true, RTC::TransportTuple *tuple = nullptr) override; void onSendSockData(Buffer::Ptr buf, bool flush = true, RTC::TransportTuple *tuple = nullptr) override;
@ -273,7 +274,7 @@ protected:
void onCreate() override; void onCreate() override;
void onDestory() override; void onDestory() override;
void onShutdown(const SockException &ex) override; void onShutdown(const SockException &ex) override;
virtual void onRecvRtp(MediaTrack &track, const std::string &rid, RtpPacket::Ptr rtp) = 0; virtual void onRecvRtp(MediaTrack &track, const std::string &rid, RtpPacket::Ptr rtp) {}
void updateTicker(); void updateTicker();
float getLossRate(TrackType type); float getLossRate(TrackType type);
void onRtcpBye() override; void onRtcpBye() override;
@ -289,7 +290,7 @@ private:
void onCheckAnswer(RtcSession &sdp); void onCheckAnswer(RtcSession &sdp);
private: private:
bool _preferred_tcp; bool _preferred_tcp = false;
uint16_t _rtx_seq[2] = {0, 0}; uint16_t _rtx_seq[2] = {0, 0};
//用掉的总流量 //用掉的总流量
uint64_t _bytes_usage = 0; uint64_t _bytes_usage = 0;
@ -310,8 +311,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 //http访问时的host ip
std::string _localIp; std::string _local_ip;
}; };
class WebRtcTransportManager { class WebRtcTransportManager {
@ -333,21 +334,20 @@ private:
class WebRtcArgs : public std::enable_shared_from_this<WebRtcArgs> { class WebRtcArgs : public std::enable_shared_from_this<WebRtcArgs> {
public: public:
virtual ~WebRtcArgs() = default; virtual ~WebRtcArgs() = default;
virtual variant operator[](const std::string &key) const = 0; virtual variant operator[](const std::string &key) const = 0;
}; };
using onCreateWebRtc = std::function<void(const WebRtcInterface &rtc)>;
class WebRtcPluginManager { class WebRtcPluginManager {
public: public:
using onCreateRtc = std::function<void(const WebRtcInterface &rtc)>; using Plugin = std::function<void(Session &sender, const WebRtcArgs &args, const onCreateWebRtc &cb)>;
using Plugin = std::function<void(Session &sender, const WebRtcArgs &args, const onCreateRtc &cb)>;
using Listener = std::function<void(Session &sender, const std::string &type, const WebRtcArgs &args, const WebRtcInterface &rtc)>; using Listener = std::function<void(Session &sender, const std::string &type, const WebRtcArgs &args, const WebRtcInterface &rtc)>;
static WebRtcPluginManager &Instance(); static WebRtcPluginManager &Instance();
void registerPlugin(const std::string &type, Plugin cb); void registerPlugin(const std::string &type, Plugin cb);
void getAnswerSdp(Session &sender, const std::string &type, const WebRtcArgs &args, const onCreateRtc &cb);
void setListener(Listener cb); void setListener(Listener cb);
void negotiateSdp(Session &sender, const std::string &type, const WebRtcArgs &args, const onCreateWebRtc &cb);
private: private:
WebRtcPluginManager() = default; WebRtcPluginManager() = default;