RTC: Support statistic for HTTP-API, HTTP-Callback and Security (#2483)

* commit message for your changes. Lines starting

* Update srs_app_rtc_api.cpp

* add SrsRtcConnPlay and SrsRtcConnPublish, in enum SrsRtmpConnType

* Update srs_rtmp_stack.cpp

* Update srs_app_rtc_conn.cpp

* Update srs_app_rtc_api.cpp

* update utest

* Update srs_utest_app.cpp

trunk/src/app/srs_app_rtc_api.cpp

@@ -94,6 +94,9 @@ srs_error_t SrsGoApiRtcPlay::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMe
     if ((prop = req->ensure_property_string("clientip")) != NULL) {
         clientip = prop->to_str();
     }
+    if (clientip.empty()) {
+        clientip = dynamic_cast<SrsHttpMessage*>(r)->connection()->remote_ip();
+    }
 
     string api;
     if ((prop = req->ensure_property_string("api")) != NULL) {
@@ -105,17 +108,19 @@ srs_error_t SrsGoApiRtcPlay::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMe
         tid = prop->to_str();
     }
 
-    // TODO: FIXME: Parse vhost.
+    // The RTC user config object.
-    // Parse app and stream from streamurl.
+    SrsRtcUserConfig ruc;
-    string app;
+    ruc.req_->ip = clientip;
-    string stream_name;
-    if (true) {
-        string tcUrl;
-        srs_parse_rtmp_url(streamurl, tcUrl, stream_name);
 
-        int port;
+    srs_parse_rtmp_url(streamurl, ruc.req_->tcUrl, ruc.req_->stream);
-        string schema, host, vhost, param;
+
-        srs_discovery_tc_url(tcUrl, schema, host, vhost, app, stream_name, port, param);
+    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();
     }
 
     // For client to specifies the candidate(EIP) of server.
@@ -129,12 +134,10 @@ srs_error_t SrsGoApiRtcPlay::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMe
     string dtls = r->query_get("dtls");
 
     srs_trace("RTC play %s, api=%s, tid=%s, clientip=%s, app=%s, stream=%s, offer=%dB, eip=%s, codec=%s, srtp=%s, dtls=%s",
-        streamurl.c_str(), api.c_str(), tid.c_str(), clientip.c_str(), app.c_str(), stream_name.c_str(), remote_sdp_str.length(),
+        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(), srtp.c_str(), dtls.c_str()
     );
 
-    // The RTC user config object.
-    SrsRtcUserConfig ruc;
     ruc.eip_ = eip;
     ruc.codec_ = codec;
     ruc.publish_ = false;
@@ -155,16 +158,6 @@ srs_error_t SrsGoApiRtcPlay::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMe
         return srs_error_wrap(err, "remote sdp check failed");
     }
 
-    ruc.req_->app = app;
-    ruc.req_->stream = stream_name;
-
-    // TODO: FIXME: Parse vhost.
-    // discovery vhost, resolve the vhost from config
-    SrsConfDirective* parsed_vhost = _srs_config->get_vhost("");
-    if (parsed_vhost) {
-        ruc.req_->vhost = parsed_vhost->arg0();
-    }
-
     SrsSdp local_sdp;
 
     // Config for SDP and session.
@@ -288,6 +281,7 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
 
     // 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) {
@@ -318,6 +312,9 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
     if ((prop = req->ensure_property_string("clientip")) != NULL) {
         clientip = prop->to_str();
     }
+    if (clientip.empty()){
+        clientip = dynamic_cast<SrsHttpMessage*>(r)->connection()->remote_ip();
+    }
 
     string api;
     if ((prop = req->ensure_property_string("api")) != NULL) {
@@ -329,16 +326,19 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
         tid = prop->to_str();
     }
 
-    // Parse app and stream from streamurl.
+    // The RTC user config object.
-    string app;
+    SrsRtcUserConfig ruc;
-    string stream_name;
+    ruc.req_->ip = clientip;
-    if (true) {
-        string tcUrl;
-        srs_parse_rtmp_url(streamurl, tcUrl, stream_name);
 
-        int port;
+    srs_parse_rtmp_url(streamurl, ruc.req_->tcUrl, ruc.req_->stream);
-        string schema, host, vhost, param;
+
-        srs_discovery_tc_url(tcUrl, schema, host, vhost, app, stream_name, port, param);
+    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();
     }
 
     // For client to specifies the candidate(EIP) of server.
@@ -349,12 +349,10 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
     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(), app.c_str(), stream_name.c_str(),
+        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()
     );
 
-    // The RTC user config object.
-    SrsRtcUserConfig ruc;
     ruc.eip_ = eip;
     ruc.codec_ = codec;
     ruc.publish_ = true;
@@ -369,16 +367,6 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
         return srs_error_wrap(err, "remote sdp check failed");
     }
 
-    ruc.req_->app = app;
-    ruc.req_->stream = stream_name;
-
-    // TODO: FIXME: Parse vhost.
-    // discovery vhost, resolve the vhost from config
-    SrsConfDirective* parsed_vhost = _srs_config->get_vhost("");
-    if (parsed_vhost) {
-        ruc.req_->vhost = parsed_vhost->arg0();
-    }
-
     SrsSdp local_sdp;
 
     // TODO: FIXME: move to create_session.

trunk/src/app/srs_app_rtc_conn.cpp

@@ -404,6 +404,10 @@ SrsRtcPlayStream::~SrsRtcPlayStream()
             srs_freep(it->second);
         }
     }
+	
+    // update the statistic when client coveried.
+    SrsStatistic* stat = SrsStatistic::instance();
+    stat->on_disconnect(cid_.c_str());
 }
 
 srs_error_t SrsRtcPlayStream::initialize(SrsRequest* req, std::map<uint32_t, SrsRtcTrackDescription*> sub_relations)
@@ -530,6 +534,12 @@ srs_error_t SrsRtcPlayStream::start()
         }
     }
 	
+    // update the statistic when client discoveried.
+    SrsStatistic* stat = SrsStatistic::instance();
+    if ((err = stat->on_client(cid_.c_str(), req_, session_, SrsRtcConnPlay)) != srs_success) {
+	return srs_error_wrap(err, "rtc: stat client");
+    }
+
     is_started = true;
 
     return err;
@@ -998,6 +1008,10 @@ SrsRtcPublishStream::~SrsRtcPublishStream()
     srs_freep(twcc_epp_);
     srs_freep(pli_epp);
     srs_freep(req);
+	
+    // update the statistic when client coveried.
+    SrsStatistic* stat = SrsStatistic::instance();
+    stat->on_disconnect(cid_.c_str());
 }
 
 srs_error_t SrsRtcPublishStream::initialize(SrsRequest* r, SrsRtcSourceDescription* stream_desc)
@@ -1113,6 +1127,12 @@ srs_error_t SrsRtcPublishStream::start()
         }
     }
 	
+    // update the statistic when client discoveried.
+    SrsStatistic* stat = SrsStatistic::instance();
+    if ((err = stat->on_client(cid_.c_str(), req, session_, SrsRtcConnPublish)) != srs_success) {
+        return srs_error_wrap(err, "rtc: stat client");
+    }
+
     is_started = true;
 
     return err;
@@ -1823,6 +1843,11 @@ std::string SrsRtcConnection::desc()
     return "RtcConn";
 }
 
+void SrsRtcConnection::expire()
+{
+    _srs_rtc_manager->remove(this);
+}
+
 void SrsRtcConnection::switch_to_context()
 {
     _srs_context->set_id(cid_);

trunk/src/app/srs_app_rtc_conn.hpp

@@ -404,7 +404,7 @@ private:
 //
 // For performance, we use non-public from resource,
 // see https://stackoverflow.com/questions/3747066/c-cannot-convert-from-base-a-to-derived-type-b-via-virtual-base-a
-class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler
+class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler, public ISrsExpire
 {
     friend class SrsSecurityTransport;
     friend class SrsRtcPlayStream;
@@ -486,6 +486,9 @@ public:
 public:
     virtual const SrsContextId& get_id();
     virtual std::string desc();
+// Interface ISrsExpire.
+public:
+    virtual void expire();
 public:
     void switch_to_context();
     const SrsContextId& context_id();

trunk/src/app/srs_app_rtc_source.cpp

@@ -27,6 +27,7 @@
 #include <srs_app_pithy_print.hpp>
 #include <srs_app_log.hpp>
 #include <srs_app_threads.hpp>
+#include <srs_app_statistic.hpp>
 
 #ifdef SRS_FFMPEG_FIT
 #include <srs_app_rtc_codec.hpp>
@@ -540,7 +541,8 @@ srs_error_t SrsRtcSource::on_publish()
         _srs_hybrid->timer100ms()->subscribe(this);
     }
 
-    // TODO: FIXME: Handle by statistic.
+    SrsStatistic* stat = SrsStatistic::instance();
+    stat->on_stream_publish(req, _source_id.c_str());
 
     return err;
 }
@@ -576,7 +578,8 @@ void SrsRtcSource::on_unpublish()
         srs_freep(bridger_);
     }
 
-    // TODO: FIXME: Handle by statistic.
+    SrsStatistic* stat = SrsStatistic::instance();
+    stat->on_stream_close(req);
 }
 
 void SrsRtcSource::subscribe(ISrsRtcSourceEventHandler* h)

trunk/src/app/srs_app_security.cpp

@@ -72,6 +72,7 @@ srs_error_t SrsSecurity::allow_check(SrsConfDirective* rules, SrsRtmpConnType ty
 
         switch (type) {
             case SrsRtmpConnPlay:
+            case SrsRtcConnPlay:
                 if (rule->arg0() != "play") {
                     break;
                 }
@@ -82,6 +83,7 @@ srs_error_t SrsSecurity::allow_check(SrsConfDirective* rules, SrsRtmpConnType ty
             case SrsRtmpConnFMLEPublish:
             case SrsRtmpConnFlashPublish:
             case SrsRtmpConnHaivisionPublish:
+            case SrsRtcConnPublish:
                 if (rule->arg0() != "publish") {
                     break;
                 }
@@ -112,6 +114,7 @@ srs_error_t SrsSecurity::deny_check(SrsConfDirective* rules, SrsRtmpConnType typ
         
         switch (type) {
             case SrsRtmpConnPlay:
+            case SrsRtcConnPlay:               
                 if (rule->arg0() != "play") {
                     break;
                 }
@@ -122,6 +125,7 @@ srs_error_t SrsSecurity::deny_check(SrsConfDirective* rules, SrsRtmpConnType typ
             case SrsRtmpConnFMLEPublish:
             case SrsRtmpConnFlashPublish:
             case SrsRtmpConnHaivisionPublish:
+            case SrsRtcConnPublish:
                 if (rule->arg0() != "publish") {
                     break;
                 }

trunk/src/protocol/srs_rtmp_stack.cpp

@@ -1628,17 +1628,19 @@ SrsResponse::~SrsResponse()
 string srs_client_type_string(SrsRtmpConnType type)
 {
     switch (type) {
-        case SrsRtmpConnPlay: return "Play";
+        case SrsRtmpConnPlay: return "rtmp-play";
+        case SrsRtcConnPlay: return "rtc-play";
         case SrsRtmpConnFlashPublish: return "flash-publish";
         case SrsRtmpConnFMLEPublish: return "fmle-publish";
         case SrsRtmpConnHaivisionPublish: return "haivision-publish";
+        case SrsRtcConnPublish: return "rtc-publish";
         default: return "Unknown";
     }
 }
 
 bool srs_client_type_is_publish(SrsRtmpConnType type)
 {
-    return type != SrsRtmpConnPlay;
+    return ((type != SrsRtmpConnPlay) && (type != SrsRtcConnPlay));
 }
 
 SrsHandshakeBytes::SrsHandshakeBytes()

trunk/src/protocol/srs_rtmp_stack.hpp

@@ -482,6 +482,8 @@ enum SrsRtmpConnType
     SrsRtmpConnFMLEPublish,
     SrsRtmpConnFlashPublish,
     SrsRtmpConnHaivisionPublish,
+    SrsRtcConnPlay,
+    SrsRtcConnPublish,
 };
 std::string srs_client_type_string(SrsRtmpConnType type);
 bool srs_client_type_is_publish(SrsRtmpConnType type);

trunk/src/utest/srs_utest_app.cpp

@@ -519,6 +519,28 @@ VOID TEST(AppSecurity, CheckSecurity)
         }
         HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtmpConnPlay, "12.13.14.15", &rr));
     }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "play", "all");
+        HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtcConnPlay, "", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "play", "12.13.14.15");
+        HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtcConnPlay, "12.13.14.15", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "play", "11.12.13.14");
+        if (true) {
+            SrsConfDirective* d = new SrsConfDirective();
+            d->name = "deny";
+            d->args.push_back("play");
+            d->args.push_back("12.13.14.15");
+            rules.directives.push_back(d);
+        }
+        HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtcConnPlay, "12.13.14.15", &rr));
+    }
     if (true) {
         SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
         rules.get_or_create("deny", "publish", "12.13.14.15");
@@ -539,6 +561,16 @@ VOID TEST(AppSecurity, CheckSecurity)
         rules.get_or_create("deny", "publish", "12.13.14.15");
         HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtmpConnHaivisionPublish, "12.13.14.15", &rr));
     }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "publish", "12.13.14.15");
+        HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtcConnPublish, "12.13.14.15", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "publish", "all");
+        HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtcConnPublish, "11.12.13.14", &rr));
+    }
 
     // Allowed if not denied.
     if (true) {
@@ -571,6 +603,26 @@ VOID TEST(AppSecurity, CheckSecurity)
         rules.get_or_create("deny", "publish", "12.13.14.15");
         HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtmpConnFlashPublish, "11.12.13.14", &rr));
     }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "play", "12.13.14.15");
+        HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtcConnPlay, "11.12.13.14", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "publish", "12.13.14.15");
+        HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtcConnPlay, "12.13.14.15", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "play", "all");
+        HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtcConnPublish, "12.13.14.15", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("deny", "play", "12.13.14.15");
+        HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtcConnPublish, "12.13.14.15", &rr));
+    }
 
     // Allowed by rule.
     if (true) {
@@ -583,6 +635,26 @@ VOID TEST(AppSecurity, CheckSecurity)
         rules.get_or_create("allow", "play", "all");
         HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtmpConnPlay, "12.13.14.15", &rr));
     }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("allow", "play", "12.13.14.15");
+        HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtcConnPlay, "12.13.14.15", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("allow", "play", "all");
+        HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtcConnPlay, "12.13.14.15", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("allow", "publish", "all");
+        HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtcConnPublish, "12.13.14.15", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("allow", "publish", "12.13.14.15");
+        HELPER_EXPECT_SUCCESS(sec.do_check(&rules, SrsRtcConnPublish, "12.13.14.15", &rr));
+    }
     if (true) {
         SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
         rules.get_or_create("allow", "publish", "all");
@@ -632,6 +704,16 @@ VOID TEST(AppSecurity, CheckSecurity)
         rules.get_or_create("allow", "publish", "12.13.14.15");
         HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtmpConnPlay, "12.13.14.15", &rr));
     }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("allow", "play", "11.12.13.14");
+        HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtcConnPlay, "12.13.14.15", &rr));
+    }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("allow", "publish", "12.13.14.15");
+        HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtcConnPlay, "12.13.14.15", &rr));
+    }
     if (true) {
         SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
         rules.get_or_create("allow", "publish", "11.12.13.14");
@@ -650,6 +732,12 @@ VOID TEST(AppSecurity, CheckSecurity)
         rules.get_or_create("deny", "play", "11.12.13.14");
         HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtmpConnPlay, "11.12.13.14", &rr));
     }
+    if (true) {
+        SrsSecurity sec; SrsRequest rr; SrsConfDirective rules;
+        rules.get_or_create("allow", "play", "11.12.13.14");
+        rules.get_or_create("deny", "play", "11.12.13.14");
+        HELPER_EXPECT_FAILED(sec.do_check(&rules, SrsRtcConnPlay, "11.12.13.14", &rr));
+    }
 
     // SRS apply the following simple strategies one by one:
     //       1. allow all if security disabled.

trunk/src/utest/srs_utest_rtmp.cpp

@@ -1208,16 +1208,20 @@ VOID TEST(ProtocolRTMPTest, RecvMessage3)
     }
 
     if (true) {
-        EXPECT_STREQ("Play", srs_client_type_string(SrsRtmpConnPlay).c_str());
+        EXPECT_STREQ("rtmp-play", srs_client_type_string(SrsRtmpConnPlay).c_str());
+        EXPECT_STREQ("rtc-play", srs_client_type_string(SrsRtcConnPlay).c_str());
+        EXPECT_STREQ("rtc-publish", srs_client_type_string(SrsRtcConnPublish).c_str());
         EXPECT_STREQ("flash-publish", srs_client_type_string(SrsRtmpConnFlashPublish).c_str());
         EXPECT_STREQ("fmle-publish", srs_client_type_string(SrsRtmpConnFMLEPublish).c_str());
         EXPECT_STREQ("haivision-publish", srs_client_type_string(SrsRtmpConnHaivisionPublish).c_str());
         EXPECT_STREQ("Unknown", srs_client_type_string(SrsRtmpConnType(0x0f)).c_str());
 
         EXPECT_TRUE(srs_client_type_is_publish(SrsRtmpConnFlashPublish));
+        EXPECT_TRUE(srs_client_type_is_publish(SrsRtcConnPublish));
         EXPECT_TRUE(srs_client_type_is_publish(SrsRtmpConnFMLEPublish));
         EXPECT_TRUE(srs_client_type_is_publish(SrsRtmpConnHaivisionPublish));
         EXPECT_FALSE(srs_client_type_is_publish(SrsRtmpConnPlay));
+        EXPECT_FALSE(srs_client_type_is_publish(SrsRtcConnPlay));
     }
 }