监听HTTP
推拉流HTTP API监听启动
文件位置：src/main/srs_app_rtc_server.cpp
推流是SrsGoApiRtcPublish类来处理，其具体处理函数实在do_serve_http
srs_error_t SrsRtcServer::listen_api()
{
srs_error_t err = srs_success;
// TODO: FIXME: Fetch api from hybrid manager, not from SRS.
SrsHttpServeMux* http_api_mux = _srs_hybrid->srs()->instance()->api_server();
if ((err = http_api_mux->handle("/rtc/v1/play/", new SrsGoApiRtcPlay(this))) != srs_success) {
return srs_error_wrap(err, "handle play");
}
if ((err = http_api_mux->handle("/rtc/v1/publish/", new SrsGoApiRtcPublish(this))) != srs_success) {
return srs_error_wrap(err, "handle publish");
}
#ifdef SRS_SIMULATOR
if ((err = http_api_mux->handle("/rtc/v1/nack/", new SrsGoApiRtcNACK(this))) != srs_success) {
return srs_error_wrap(err, "handle nack");
}
#endif
return err;
}
在do_serve_http中，首先解析http请求参数，其参数含义：
- sdp：请求端的sdp
- streamurl：推流地址
- api：url
- clientip：请求端的ip地址，如果为空会默认设置为http请求端的IP
- tid：应该类似标志的事务ID
然后再srs_parse_rtmp_url解析rtmp的URL，srs_discovery_tc_url解析RTC的URL。
srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, SrsJsonObject* res)
{
srs_error_t err = srs_success;
//解析请求头
// For each RTC session, we use short-term HTTP connection.
SrsHttpHeader* hdr = w->header();
hdr->set("Connection", "Close");
//解析请求体
// Parse req, the request json object, from body.
SrsJsonObject* req = NULL;
SrsAutoFree(SrsJsonObject, req);
if (true) {
string req_json;
if ((err = r->body_read_all(req_json)) != srs_success) {
return srs_error_wrap(err, "read body");
}
SrsJsonAny* json = SrsJsonAny::loads(req_json);
if (!json || !json->is_object()) {
return srs_error_new(ERROR_RTC_API_BODY, "invalid body %s", req_json.c_str());
}
req = json->to_object();
}
//获取请求参数
// Fetch params from req object.
SrsJsonAny* prop = NULL;
if ((prop = req->ensure_property_string("sdp")) == NULL) {
return srs_error_wrap(err, "not sdp");
}
string remote_sdp_str = prop->to_str();
if ((prop = req->ensure_property_string("streamurl")) == NULL) {
return srs_error_wrap(err, "not streamurl");
}
string streamurl = prop->to_str();
string clientip;
if ((prop = req->ensure_property_string("clientip")) != NULL) {
clientip = prop->to_str();
}
if (clientip.empty()){
clientip = dynamic_cast<SrsHttpMessage*>(r)->connection()->remote_ip();
// Overwrite by ip from proxy.
string oip = srs_get_original_ip(r);
if (!oip.empty()) {
clientip = oip;
}
}
string api;
if ((prop = req->ensure_property_string("api")) != NULL) {
api = prop->to_str();
}
string tid;
if ((prop = req->ensure_property_string("tid")) != NULL) {
tid = prop->to_str();
}
// The RTC user config object.
SrsRtcUserConfig ruc;
ruc.req_->ip = clientip;
ruc.api_ = api;
// 解析RTMP URL
srs_parse_rtmp_url(streamurl, ruc.req_->tcUrl, ruc.req_->stream);
// 解析RTC URL
srs_discovery_tc_url(ruc.req_->tcUrl, ruc.req_->schema, ruc.req_->host, ruc.req_->vhost,
ruc.req_->app, ruc.req_->stream, ruc.req_->port, ruc.req_->param);
// discovery vhost, resolve the vhost from config
SrsConfDirective* parsed_vhost = _srs_config->get_vhost(ruc.req_->vhost);
if (parsed_vhost) {
ruc.req_->vhost = parsed_vhost->arg0();
}
if ((err = http_hooks_on_publish(ruc.req_)) != srs_success) {
return srs_error_wrap(err, "RTC: http_hooks_on_publish");
}
// For client to specifies the candidate(EIP) of server.
string eip = r->query_get("eip");
if (eip.empty()) {
eip = r->query_get("candidate");
}
string codec = r->query_get("codec");
srs_trace("RTC publish %s, api=%s, tid=%s, clientip=%s, app=%s, stream=%s, offer=%dB, eip=%s, codec=%s",
streamurl.c_str(), api.c_str(), tid.c_str(), clientip.c_str(), ruc.req_->app.c_str(), ruc.req_->stream.c_str(),
remote_sdp_str.length(), eip.c_str(), codec.c_str()
);
ruc.eip_ = eip;
ruc.codec_ = codec;
ruc.publish_ = true;
ruc.dtls_ = ruc.srtp_ = true;
// TODO: FIXME: It seems remote_sdp doesn't represents the full SDP information.
// 解析sdp
if ((err = ruc.remote_sdp_.parse(remote_sdp_str)) != srs_success) {
return srs_error_wrap(err, "parse sdp failed: %s", remote_sdp_str.c_str());
}
// 校验SDP
if ((err = check_remote_sdp(ruc.remote_sdp_)) != srs_success) {
return srs_error_wrap(err, "remote sdp check failed");
}
SrsSdp local_sdp;
// TODO: FIXME: move to create_session.
// Config for SDP and session.
local_sdp.session_config_.dtls_role = _srs_config->get_rtc_dtls_role(ruc.req_->vhost);
local_sdp.session_config_.dtls_version = _srs_config->get_rtc_dtls_version(ruc.req_->vhost);
// Whether enabled.
bool server_enabled = _srs_config->get_rtc_server_enabled();
bool rtc_enabled = _srs_config->get_rtc_enabled(ruc.req_->vhost);
if (server_enabled && !rtc_enabled) {
srs_warn("RTC disabled in vhost %s", ruc.req_->vhost.c_str());
}
if (!server_enabled || !rtc_enabled) {
return srs_error_new(ERROR_RTC_DISABLED, "Disabled server=%d, rtc=%d, vhost=%s",
server_enabled, rtc_enabled, ruc.req_->vhost.c_str());
}
// TODO: FIXME: When server enabled, but vhost disabled, should report error.
// 创建RTC会话，并写入answer sdp
SrsRtcConnection* session = NULL;
if ((err = server_->create_session(&ruc, local_sdp, &session)) != srs_success) {
return srs_error_wrap(err, "create session");
}
//将answer sdp写入到ostringstream
ostringstream os;
if ((err = local_sdp.encode(os)) != srs_success) {
return srs_error_wrap(err, "encode sdp");
}
string local_sdp_str = os.str();
// Filter the \r\n to \\r\\n for JSON.
string local_sdp_escaped = srs_string_replace(local_sdp_str.c_str(), "\r\n", "\\r\\n");
// 设置响应参数
res->set("code", SrsJsonAny::integer(ERROR_SUCCESS));
res->set("server", SrsJsonAny::str(SrsStatistic::instance()->server_id().c_str()));
// TODO: add candidates in response json?
res->set("sdp", SrsJsonAny::str(local_sdp_str.c_str()));
res->set("sessionid", SrsJsonAny::str(session->username().c_str()));
srs_trace("RTC username=%s, offer=%dB, answer=%dB", session->username().c_str(),
remote_sdp_str.length(), local_sdp_escaped.length());
srs_trace("RTC remote offer: %s", srs_string_replace(remote_sdp_str.c_str(), "\r\n", "\\r\\n").c_str());
srs_trace("RTC local answer: %s", local_sdp_escaped.c_str());
return err;
}
解析sdp
推流解析SDP
然后 ruc.remote_sdp_.parse(remote_sdp_str)解析SDP，SrsSdp::parse首先对读取每一行然后对行调用parse_line进行解析
srs_error_t SrsSdp::parse(const std::string& sdp_str)
{
srs_error_t err = srs_success;
// All webrtc SrsSdp annotated example
// @see: https://tools.ietf.org/html/draft-ietf-rtcweb-SrsSdp-11
// Sdp example
// session info
// v=
// o=
// s=
// t=
// media description
// m=
// a=
// ...
// media description
// m=
// a=
// ...
std::istringstream is(sdp_str);
std::string line;
// 以行读取
while (getline(is, line)) {
srs_verbose("%s", line.c_str());
if (line.size() < 2 || line[1] != '=') {
return srs_error_new(ERROR_RTC_SDP_DECODE, "invalid sdp line=%s", line.c_str());
}
if (!line.empty() && line[line.size()-1] == '\r') {
line.erase(line.size()-1, 1);
}
// Strip the space of line, for pion WebRTC client.
line = srs_string_trim_end(line, " ");
// 解析行
if ((err = parse_line(line)) != srs_success) {
return srs_error_wrap(err, "parse sdp line failed");
}
}
// The msid/tracker/mslabel is optional for SSRC, so we copy it when it's empty.
for (std::vector<SrsMediaDesc>::iterator iter = media_descs_.begin(); iter != media_descs_.end(); ++iter) {
SrsMediaDesc& media_desc = *iter;
for (size_t i = 0; i < media_desc.ssrc_infos_.size(); ++i) {
SrsSSRCInfo& ssrc_info = media_desc.ssrc_infos_.at(i);
if (ssrc_info.msid_.empty()) {
ssrc_info.msid_ = media_desc.msid_;
}
if (ssrc_info.msid_tracker_.empty()) {
ssrc_info.msid_tracker_ = media_desc.msid_tracker_;
}
if (ssrc_info.mslabel_.empty()) {
ssrc_info.mslabel_ = media_desc.msid_;
}
if (ssrc_info.label_.empty()) {
ssrc_info.label_ = media_desc.msid_tracker_;
}
}
}
return err;
}
parse_line
srs_error_t SrsSdp::parse_line(const std::string& line)
{
srs_error_t err = srs_success;
std::string content = line.substr(2);
switch (line[0]) {
case 'o': {
return parse_origin(content);
}
case 'v': {
return parse_version(content);
}
case 's': {
return parse_session_name(content);
}
case 't': {
return parse_timing(content);
}
case 'a': {
if (in_media_session_) {
return media_descs_.back().parse_line(line);
}
return parse_attribute(content);
}
case 'm': {
return parse_media_description(content);
}
case 'c': {
// TODO: process c-line
break;
}
default: {
srs_trace("ignore sdp line=%s", line.c_str());
break;
}
}
return err;
}
校验sdp
check_remote_sdp对SDP进行校验，主要校验是否支持bound媒体复用和媒体能力
srs_error_t SrsGoApiRtcPublish::check_remote_sdp(const SrsSdp& remote_sdp)
{
srs_error_t err = srs_success;
if (remote_sdp.group_policy_ != "BUNDLE") {
return srs_error_new(ERROR_RTC_SDP_EXCHANGE, "now only support BUNDLE, group policy=%s", remote_sdp.group_policy_.c_str());
}
if (remote_sdp.media_descs_.empty()) {
return srs_error_new(ERROR_RTC_SDP_EXCHANGE, "no media descriptions");
}
for (std::vector<SrsMediaDesc>::const_iterator iter = remote_sdp.media_descs_.begin(); iter != remote_sdp.media_descs_.end(); ++iter) {
if (iter->type_ != "audio" && iter->type_ != "video") {
return srs_error_new(ERROR_RTC_SDP_EXCHANGE, "unsupport media type=%s", iter->type_.c_str());
}
if (! iter->rtcp_mux_) {
return srs_error_new(ERROR_RTC_SDP_EXCHANGE, "now only suppor rtcp-mux");
}
if (iter->recvonly_) {
return srs_error_new(ERROR_RTC_SDP_EXCHANGE, "publish API only support sendrecv/sendonly");
}
}
return err;
}
生成SDP
生成本地sdp
server_->create_session进行媒体协商并创建会话信息和生成SDP
srs_error_t SrsRtcServer::create_session(SrsRtcUserConfig* ruc, SrsSdp& local_sdp, SrsRtcConnection** psession)
{
...
SrsRtcSource* source = NULL;
// publish创建SrsRtcSource,即创建流，play绑定SrsRtcSource，即绑定流
if ((err = _srs_rtc_sources->fetch_or_create(req, &source)) != srs_success) {
return srs_error_wrap(err, "create source");
}
...
// TODO: FIXME: add do_create_session to error process.
SrsRtcConnection* session = new SrsRtcConnection(this, cid);
// 创建RTC会话，设置answer sdp
if ((err = do_create_session(ruc, local_sdp, session)) != srs_success) {
srs_freep(session);
return srs_error_wrap(err, "create session");
}
*psession = session;
return err;
}
将sdp编码成字符串
local_sdp.encode(os)将SDP编码成响应的sdp字符串
srs_error_t SrsSdp::encode(std::ostringstream& os)
{
srs_error_t err = srs_success;
os << "v=" << version_ << kCRLF;
os << "o=" << username_ << " " << session_id_ << " " << session_version_ << " " << nettype_ << " " << addrtype_ << " " << unicast_address_ << kCRLF;
os << "s=" << session_name_ << kCRLF;
os << "t=" << start_time_ << " " << end_time_ << kCRLF;
// ice-lite is a minimal version of the ICE specification, intended for servers running on a public IP address.
os << "a=ice-lite" << kCRLF;
if (!groups_.empty()) {
os << "a=group:" << group_policy_;
for (std::vector<std::string>::iterator iter = groups_.begin(); iter != groups_.end(); ++iter) {
os << " " << *iter;
}
os << kCRLF;
}
os << "a=msid-semantic: " << msid_semantic_;
for (std::vector<std::string>::iterator iter = msids_.begin(); iter != msids_.end(); ++iter) {
os << " " << *iter;
}
os << kCRLF;
if ((err = session_info_.encode(os)) != srs_success) {
return srs_error_wrap(err, "encode session info failed");
}
for (std::vector<SrsMediaDesc>::iterator iter = media_descs_.begin(); iter != media_descs_.end(); ++iter) {
if ((err = (*iter).encode(os)) != srs_success) {
return srs_error_wrap(err, "encode media description failed");
}
}
return err;
}
srs_error_t SrsRtcServer::do_create_session(SrsRtcUserConfig* ruc, SrsSdp& local_sdp, SrsRtcConnection* session)
{
srs_error_t err = srs_success;
SrsRequest* req = ruc->req_;
// first add publisher/player for negotiate sdp media info
if (ruc->publish_) {
if ((err = session->add_publisher(ruc, local_sdp)) != srs_success) {
return srs_error_wrap(err, "add publisher");
}
} else {
if ((err = session->add_player(ruc, local_sdp)) != srs_success) {
return srs_error_wrap(err, "add player");
}
}
...
session->set_remote_sdp(ruc->remote_sdp_);
// We must setup the local SDP, then initialize the session object.
session->set_local_sdp(local_sdp);
session->set_state(WAITING_STUN);
// 初始化会话信息
if ((err = session->initialize(req, ruc->dtls_, ruc->srtp_, username)) != srs_success) {
return srs_error_wrap(err, "init");
}
// We allows username is optional, but it never empty here.
_srs_rtc_manager->add_with_name(username, session);
return err;
}
add_publisher
add_publisher主要进行媒体协商并且生成SDP然后创建。
srs_error_t SrsRtcConnection::add_publisher(SrsRtcUserConfig* ruc, SrsSdp& local_sdp)
{
srs_error_t err = srs_success;
SrsRequest* req = ruc->req_;
SrsRtcSourceDescription* stream_desc = new SrsRtcSourceDescription();
SrsAutoFree(SrsRtcSourceDescription, stream_desc);
// TODO: FIXME: Change to api of stream desc.
//媒体协商
if ((err = negotiate_publish_capability(ruc, stream_desc)) != srs_success) {
return srs_error_wrap(err, "publish negotiate");
}
//生成 answer sdp
if ((err = generate_publish_local_sdp(req, local_sdp, stream_desc, ruc->remote_sdp_.is_unified())) != srs_success) {
return srs_error_wrap(err, "generate local sdp");
}
...
source->set_stream_created();
// Apply the SDP to source.
source->set_stream_desc(stream_desc);
// TODO: FIXME: What happends when error?
if ((err = create_publisher(req, stream_desc)) != srs_success) {
return srs_error_wrap(err, "create publish");
}
return err;
}
add_player
总结
主要函数如下图：
