fix ossrs#179, dvr support api to start or stop. 2.0.123

trunk/conf/full.conf

@@ -304,7 +304,7 @@ vhost dvr.srs.com {
         #           to start dvr of specified vhost.
         #           request should encode in json, specifies the dvr to create, where:
         #               {path_tmpl:"./[15].[04].[05].[999].flv", 
-        #                   wait_keyframe:true, vhost:"__defaultVhost", callback:"http://dvr/callback"
+        #                   wait_keyframe:true, vhost:"__defaultVhost", callback:"http://127.0.0.1:8085/api/v1/dvrs"
         #               }
         #           response in json, where:
         #               {code:0}
@@ -315,7 +315,12 @@ vhost dvr.srs.com {
         #           response in json, where:
         #               {code:0}
         #       when reap segment, the callback POST request in json:
-        #               {action:"on_dvr_reap_segment"}
+        #               {action:"on_dvr_reap_segment", client_id:100, vhost:"__defaultVhost__", 
+        #                   app:"live", stream:"livestream", cwd:"/home/winlin/srs", file:"./dvr.flv"
+        #               }
+        #           for the dvr http callback, @see http_hooks.on_dvr of vhost hooks.callback.srs.com
+        #           @read https://github.com/winlinvip/simple-rtmp-server/wiki/v2_CN_DVR#http-callback
+        #           @read https://github.com/winlinvip/simple-rtmp-server/wiki/v2_EN_DVR#http-callback
         # default: session
         dvr_plan        session;
         # the dvr output path.

trunk/src/app/srs_app_dvr.cpp

@@ -45,6 +45,9 @@ using namespace std;
 // update the flv duration and filesize every this interval in ms.
 #define __SRS_DVR_UPDATE_DURATION_INTERVAL 60000
 
+// the sleep interval for http async callback.
+#define SRS_AUTO_ASYNC_CALLBACL_SLEEP_US 300000
+
 SrsFlvSegment::SrsFlvSegment(SrsDvrPlan* p)
 {
     req = NULL;
@@ -201,29 +204,6 @@ int SrsFlvSegment::close()
         srs_error("dvr: notify plan to reap segment failed. ret=%d", ret);
         return ret;
     }
-
-#ifdef SRS_AUTO_HTTP_CALLBACK
-    if (_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
-        // HTTP: on_dvr 
-        SrsConfDirective* on_dvr = _srs_config->get_vhost_on_dvr(req->vhost);
-        if (!on_dvr) {
-            srs_info("ignore the empty http callback: on_dvr");
-            return ret;
-        }
-
-        int connection_id = _srs_context->get_id();
-        std::string ip = req->ip;
-        std::string cwd = _srs_config->cwd();
-        std::string file = path;
-        for (int i = 0; i < (int)on_dvr->args.size(); i++) {
-            std::string url = on_dvr->args.at(i);
-            if ((ret = SrsHttpHooks::on_dvr(url, connection_id, ip, req, cwd, file)) != ERROR_SUCCESS) {
-                srs_error("hook client on_dvr failed. url=%s, ret=%d", url.c_str(), ret);
-                return ret;
-            }
-        }
-    }
-#endif
 
     return ret;
 }
@@ -578,18 +558,174 @@ int SrsFlvSegment::on_reload_vhost_dvr(std::string /*vhost*/)
     return ret;
 }
 
+ISrsDvrAsyncCall::ISrsDvrAsyncCall()
+{
+}
+
+ISrsDvrAsyncCall::~ISrsDvrAsyncCall()
+{
+}
+
+SrsDvrAsyncCallOnDvr::SrsDvrAsyncCallOnDvr(SrsRequest* r, string p)
+{
+    req = r;
+    path = p;
+}
+
+SrsDvrAsyncCallOnDvr::~SrsDvrAsyncCallOnDvr()
+{
+}
+
+int SrsDvrAsyncCallOnDvr::call()
+{
+    int ret = ERROR_SUCCESS;
+
+#ifdef SRS_AUTO_HTTP_CALLBACK
+    // http callback for on_dvr in config.
+    if (_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
+        // HTTP: on_dvr 
+        SrsConfDirective* on_dvr = _srs_config->get_vhost_on_dvr(req->vhost);
+        if (!on_dvr) {
+            srs_info("ignore the empty http callback: on_dvr");
+            return ret;
+        }
+
+        int connection_id = _srs_context->get_id();
+        std::string ip = req->ip;
+        std::string cwd = _srs_config->cwd();
+        std::string file = path;
+        for (int i = 0; i < (int)on_dvr->args.size(); i++) {
+            std::string url = on_dvr->args.at(i);
+            if ((ret = SrsHttpHooks::on_dvr(url, connection_id, ip, req, cwd, file)) != ERROR_SUCCESS) {
+                srs_error("hook client on_dvr failed. url=%s, ret=%d", url.c_str(), ret);
+                return ret;
+            }
+        }
+    }
+#endif
+
+    return ret;
+}
+
+string SrsDvrAsyncCallOnDvr::to_string()
+{
+    std::stringstream ss;
+    ss << "vhost=" << req->vhost << ", file=" << path;
+    return ss.str();
+}
+
+SrsDvrAsyncCallOnSegment::SrsDvrAsyncCallOnSegment(SrsRequest* r, string c, string p)
+{
+    req = r;
+    callback = c;
+    path = p;
+}
+
+SrsDvrAsyncCallOnSegment::~SrsDvrAsyncCallOnSegment()
+{
+}
+
+int SrsDvrAsyncCallOnSegment::call()
+{
+    int ret = ERROR_SUCCESS;
+
+#ifdef SRS_AUTO_HTTP_CALLBACK
+    // HTTP: callback
+    if (callback.empty()) {
+        srs_warn("dvr: ignore for callback empty, vhost=%s", req->vhost.c_str());
+        return ret;
+    }
+
+    int connection_id = _srs_context->get_id();
+    std::string cwd = _srs_config->cwd();
+    std::string file = path;
+    std::string url = callback;
+    if ((ret = SrsHttpHooks::on_dvr_reap_segment(url, connection_id, req, cwd, file)) != ERROR_SUCCESS) {
+        srs_error("hook client on_dvr_reap_segment failed. url=%s, ret=%d", url.c_str(), ret);
+        return ret;
+    }
+#endif
+
+    return ret;
+}
+
+string SrsDvrAsyncCallOnSegment::to_string()
+{
+    std::stringstream ss;
+    ss << "vhost=" << req->vhost << ", file=" << path << "callback=" << callback;
+    return ss.str();
+}
+
+SrsDvrAsyncCallThread::SrsDvrAsyncCallThread()
+{
+    pthread = new SrsThread("async", this, SRS_AUTO_ASYNC_CALLBACL_SLEEP_US, true);
+}
+
+SrsDvrAsyncCallThread::~SrsDvrAsyncCallThread()
+{
+    stop();
+    srs_freep(pthread);
+
+    std::vector<ISrsDvrAsyncCall*>::iterator it;
+    for (it = callbacks.begin(); it != callbacks.end(); ++it) {
+        ISrsDvrAsyncCall* call = *it;
+        srs_freep(call);
+    }
+    callbacks.clear();
+}
+
+int SrsDvrAsyncCallThread::call(ISrsDvrAsyncCall* c)
+{
+    int ret = ERROR_SUCCESS;
+
+    callbacks.push_back(c);
+
+    return ret;
+}
+
+int SrsDvrAsyncCallThread::start()
+{
+    return pthread->start();
+}
+
+void SrsDvrAsyncCallThread::stop()
+{
+    pthread->stop();
+}
+
+int SrsDvrAsyncCallThread::cycle()
+{
+    int ret = ERROR_SUCCESS;
+
+    std::vector<ISrsDvrAsyncCall*> copies = callbacks;
+    callbacks.clear();
+
+    std::vector<ISrsDvrAsyncCall*>::iterator it;
+    for (it = copies.begin(); it != copies.end(); ++it) {
+        ISrsDvrAsyncCall* call = *it;
+        if ((ret = call->call()) != ERROR_SUCCESS) {
+            srs_warn("dvr: ignore callback %s, ret=%d", call->to_string().c_str(), ret);
+        }
+        srs_freep(call);
+    }
+
+    return ret;
+}
+
 SrsDvrPlan::SrsDvrPlan()
 {
     source = NULL;
     req = NULL;
 
     dvr_enabled = false;
     segment = new SrsFlvSegment(this);
+    async = new SrsDvrAsyncCallThread();
 }
 
 SrsDvrPlan::~SrsDvrPlan()
 {
     srs_freep(segment);
+    srs_freep(async);
 }
 
 int SrsDvrPlan::initialize(SrsSource* s, SrsRequest* r)
@@ -603,6 +739,10 @@ int SrsDvrPlan::initialize(SrsSource* s, SrsRequest* r)
         return ret;
     }
 
+    if ((ret = async->start()) != ERROR_SUCCESS) {
+        return ret;
+    }
+
     return ret;
 }
 
@@ -671,7 +811,13 @@ int SrsDvrPlan::on_video(SrsSharedPtrMessage* __video)
 
 int SrsDvrPlan::on_reap_segment()
 {
-    return ERROR_SUCCESS;
+    int ret = ERROR_SUCCESS;
+
+    if ((ret = async->call(new SrsDvrAsyncCallOnDvr(req, segment->get_path()))) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    return ret;
 }
 
 SrsDvrPlan* SrsDvrPlan::create_plan(string vhost)
@@ -958,23 +1104,14 @@ int SrsDvrApiPlan::stop()
 int SrsDvrApiPlan::on_reap_segment()
 {
     int ret = ERROR_SUCCESS;
-
-#ifdef SRS_AUTO_HTTP_CALLBACK
-    // HTTP: callback
-    if (callback.empty()) {
-        srs_warn("dvr: ignore for callback empty, vhost=%s", req->vhost.c_str());
+
+    if ((ret = SrsDvrPlan::on_reap_segment()) != ERROR_SUCCESS) {
         return ret;
     }
-
-    int connection_id = _srs_context->get_id();
-    std::string cwd = _srs_config->cwd();
-    std::string file = segment->get_path();
-    std::string url = callback;
-    if ((ret = SrsHttpHooks::on_dvr_reap_segment(url, connection_id, req, cwd, file)) != ERROR_SUCCESS) {
-        srs_error("hook client on_dvr_reap_segment failed. url=%s, ret=%d", url.c_str(), ret);
+
+    if ((ret = async->call(new SrsDvrAsyncCallOnSegment(req, callback, segment->get_path()))) != ERROR_SUCCESS) {
         return ret;
     }
-#endif
 
     return ret;
 }

trunk/src/app/srs_app_dvr.hpp

@@ -44,9 +44,11 @@ class SrsFileWriter;
 class SrsFlvEncoder;
 class SrsDvrPlan;
 class SrsJsonAny;
+class SrsThread;
 
 #include <srs_app_source.hpp>
 #include <srs_app_reload.hpp>
+#include <srs_app_thread.hpp>
 
 /**
 * a piece of flv segment.
@@ -173,6 +175,63 @@ class SrsFlvSegment : public ISrsReloadHandler
     virtual int on_reload_vhost_dvr(std::string vhost);
 };
 
+/**
+* the dvr async call.
+*/
+class ISrsDvrAsyncCall
+{
+public:
+    ISrsDvrAsyncCall();
+    virtual ~ISrsDvrAsyncCall();
+public:
+    virtual int call() = 0;
+    virtual std::string to_string() = 0;
+};
+class SrsDvrAsyncCallOnDvr : public ISrsDvrAsyncCall
+{
+private:
+    std::string path;
+    SrsRequest* req;
+public:
+    SrsDvrAsyncCallOnDvr(SrsRequest* r, std::string p);
+    virtual ~SrsDvrAsyncCallOnDvr();
+public:
+    virtual int call();
+    virtual std::string to_string();
+};
+class SrsDvrAsyncCallOnSegment : public ISrsDvrAsyncCall
+{
+private:
+    std::string callback;
+    std::string path;
+    SrsRequest* req;
+public:
+    SrsDvrAsyncCallOnSegment(SrsRequest* r, std::string c, std::string p);
+    virtual ~SrsDvrAsyncCallOnSegment();
+public:
+    virtual int call();
+    virtual std::string to_string();
+};
+
+/**
+* the async callback for dvr.
+*/
+class SrsDvrAsyncCallThread : public ISrsThreadHandler
+{
+private:
+    SrsThread* pthread;
+    std::vector<ISrsDvrAsyncCall*> callbacks;
+public:
+    SrsDvrAsyncCallThread();
+    virtual ~SrsDvrAsyncCallThread();
+public:
+    virtual int call(ISrsDvrAsyncCall* c);
+public:
+    virtual int start();
+    virtual void stop();
+    virtual int cycle();
+};
+
 /**
 * the plan for dvr.
 * use to control the following dvr params:
@@ -189,6 +248,7 @@ class SrsDvrPlan
 protected:
     SrsSource* source;
     SrsFlvSegment* segment;
+    SrsDvrAsyncCallThread* async;
     bool dvr_enabled;
 public:
     SrsDvrPlan();

trunk/src/app/srs_app_http.cpp

@@ -851,9 +851,18 @@ SrsHttpMessage::~SrsHttpMessage()
 int SrsHttpMessage::initialize()
 {
     int ret = ERROR_SUCCESS;
+
+    std::string host = get_request_header("Host");
+
+    // donot parse the empty host for uri, 
+    // for example, the response contains no host,
+    // ignore it is ok.
+    if (host.empty()) {
+        return ret;
+    }
 
     // parse uri to schema/server:port/path?query
-    std::string uri = "http://" + get_request_header("Host") + _url;
+    std::string uri = "http://" + host + _url;
     if ((ret = _uri->initialize(uri)) != ERROR_SUCCESS) {
         return ret;
     }