Merge branch '2.0release' into develop

@@ -7,6 +7,7 @@ RMB 10000+
 * [2015-03-03 13:25] 郭强
 * [2015-03-03 13:25] 郭强
 
 
 RMB 1000-9999
 RMB 1000-9999
+* [2015-04-29 09:20] 王光辉
 * [2015-04-04 16:19] 蔡汉城
 * [2015-04-04 16:19] 蔡汉城
 
 
 RMB 500-999
 RMB 500-999

@@ -780,7 +780,10 @@ vhost hooks.callback.srs.com {
         #           "stream": "livestream",
         #           "stream": "livestream",
         #           "duration": 9.36, // in seconds
         #           "duration": 9.36, // in seconds
         #           "cwd": "/usr/local/srs",
         #           "cwd": "/usr/local/srs",
-        #           "file": "./objs/nginx/html/live/livestream.1420254068776-100.ts",
+        #           "file": "./objs/nginx/html/live/livestream/2015-04-23/01/476584165.ts",
+        #           "url": "live/livestream/2015-04-23/01/476584165.ts",
+        #           "m3u8": "./objs/nginx/html/live/livestream/live.m3u8",
+        #           "m3u8_url": "live/livestream/live.m3u8",
         #           "seq_no": 100
         #           "seq_no": 100
         #       }
         #       }
         # if valid, the hook must return HTTP code 200(Stauts OK) and response
         # if valid, the hook must return HTTP code 200(Stauts OK) and response

+if [[ $# -lt 2 ]]; then
+    echo "Usage: $0 <hls0> <hls1> [keep_ts]"
+    echo "      keep_ts to keep the download ts, default is off"
+    echo "For example:"
+    echo "      $0 http://192.168.1.137:1980/hls/live/stone/live.m3u8 http://192.168.1.137:1984/hls/live/stone/live.m3u8"
+    echo "      $0 http://192.168.1.137:1980/hls/live/livestream/live.m3u8 http://192.168.1.137:1984/hls/live/livestream/live.m3u8"
+    echo "      $0 http://ossrs.net:1984/hls/live/livestream/live.m3u8 http://ossrs.net:1996/hls/live/livestream/live.m3u8"
+    exit 1
+fi
+ 
+hls0=$1
+hls1=$2
+keep_ts=NO
+if [[ $# -gt 2 ]]; then
+    keep_ts=YES
+fi
+#echo "check hls backup of $hls0 vs $hls1, keep_ts=$keep_ts"
+
+hls0_tss=`curl $hls0 2>/dev/null |grep "\.ts"`
+hls1_tss=`curl $hls1 2>/dev/null |grep "\.ts"`
+
+hls0_prefix=`dirname $hls0`
+hls1_prefix=`dirname $hls1`
+work_dir="./hbt_temp_`date +%s`"
+
+md5_tool="md5"
+`md5sum --version >/dev/null 2>&1` && md5_tool="md5sum"
+#echo "use md5 tool: $md5_tool"
+
+CHECKED=NO
+OK=YES
+for ts in $hls0_tss; do
+    match=NO
+    for ts1 in $hls1_tss; do
+        if [[ $ts == $ts1 ]]; then
+            #echo "check ts $ts"
+            match=YES
+            break
+        fi
+    done
+    #echo "check ts $ts, match=$match"
+    
+    if [ $match = NO ]; then
+        echo "skip $ts"
+        continue
+    fi
+    
+    ts0_uri=$hls0_prefix/$ts
+    ts1_uri=$hls1_prefix/$ts
+    ts0_tmp=$work_dir/hls0/`basename $ts`
+    ts1_tmp=$work_dir/hls1/`basename $ts`
+    #echo "start check $ts0_uri($ts0_tmp) vs $ts1_uri($ts1_tmp)"
+    
+    mkdir -p `dirname $ts0_tmp` &&
+    curl $ts0_uri >$ts0_tmp 2>/dev/null &&
+    ret=$?; if [[ $ret -ne 0 ]]; then echo "download $ts0_uri to $ts0_tmp failed. ret=$ret"; exit $ret; fi
+    
+    mkdir -p `dirname $ts1_tmp` &&
+    curl $ts1_uri >$ts1_tmp 2>/dev/null &&
+    ret=$?; if [[ $ret -ne 0 ]]; then echo "download $ts1_uri to $ts1_tmp failed. ret=$ret"; exit $ret; fi
+    
+    if [[ $md5_tool == "md5" ]]; then
+        ts0_cs=`$md5_tool $ts0_tmp|awk '{print $4}'`
+    else
+        ts0_cs=`$md5_tool $ts0_tmp|awk '{print $1}'`
+    fi
+    #echo "hls0: md5sum($ts0_tmp)=$ts0_cs"
+    
+    if [[ $md5_tool == "md5" ]]; then
+        ts1_cs=`$md5_tool $ts1_tmp|awk '{print $4}'`
+    else
+        ts1_cs=`$md5_tool $ts1_tmp|awk '{print $1}'`
+    fi
+    #echo "hls1: md5sum($ts1_tmp)=$ts1_cs"
+    
+    if [[ $ts0_cs != $ts1_cs ]]; then
+        echo "$ts0_uri($ts0_cs) not equals to $ts1_uri($ts1_cs)"
+        OK=NO
+    fi
+    CHECKED=YES
+done
+
+if [ $keep_ts = NO ]; then
+    #echo "clenaup work dir $work_dir"
+    rm -rf $work_dir
+else
+    echo "keep work dir $work_dir"
+fi
+
+#echo "====================================================="
+if [[ $OK = YES && $CHECKED = YES ]]; then
+    echo "OK"
+    exit 0
+else
+    echo "FAILED"
+    exit 1
+fi
+
+exit 0

@@ -2570,7 +2570,7 @@ bool SrsConfig::get_vhost_is_edge(SrsConfDirective* vhost)
         return SRS_CONF_DEFAULT_EDGE_MODE;
         return SRS_CONF_DEFAULT_EDGE_MODE;
     }
     }
     
     
-    return SRS_CONF_PERFER_FALSE(conf->arg0());
+    return conf->arg0() == "remote";
 }
 }
 
 
 SrsConfDirective* SrsConfig::get_vhost_edge_origin(string vhost)
 SrsConfDirective* SrsConfig::get_vhost_edge_origin(string vhost)

@@ -170,10 +170,13 @@ void SrsHlsSegment::update_duration(int64_t current_frame_dts)
     return;
     return;
 }
 }
 
 
-SrsDvrAsyncCallOnHls::SrsDvrAsyncCallOnHls(SrsRequest* r, string p, int s, double d)
+SrsDvrAsyncCallOnHls::SrsDvrAsyncCallOnHls(SrsRequest* r, string p, string t, string m, string mu, int s, double d)
 {
 {
     req = r;
     req = r;
     path = p;
     path = p;
+    ts_url = t;
+    m3u8 = m;
+    m3u8_url = mu;
     seq_no = s;
     seq_no = s;
     duration = d;
     duration = d;
 }
 }
@@ -200,7 +203,7 @@ int SrsDvrAsyncCallOnHls::call()
         int sn = seq_no;
         int sn = seq_no;
         for (int i = 0; i < (int)on_hls->args.size(); i++) {
         for (int i = 0; i < (int)on_hls->args.size(); i++) {
             std::string url = on_hls->args.at(i);
             std::string url = on_hls->args.at(i);
-            if ((ret = SrsHttpHooks::on_hls(url, req, file, sn, duration)) != ERROR_SUCCESS) {
+            if ((ret = SrsHttpHooks::on_hls(url, req, file, ts_url, m3u8, m3u8_url, sn, duration)) != ERROR_SUCCESS) {
                 srs_error("hook client on_hls failed. url=%s, ret=%d", url.c_str(), ret);
                 srs_error("hook client on_hls failed. url=%s, ret=%d", url.c_str(), ret);
                 return ret;
                 return ret;
             }
             }
@@ -361,8 +364,8 @@ int SrsHlsMuxer::update_config(SrsRequest* r, string entry_prefix,
     deviation_ts = 0;
     deviation_ts = 0;
     
     
     // generate the m3u8 dir and path.
     // generate the m3u8 dir and path.
-    m3u8 = path + "/" + m3u8_file;
-    m3u8 = srs_path_build_stream(m3u8, req->vhost, req->app, req->stream);
+    m3u8_url = srs_path_build_stream(m3u8_file, req->vhost, req->app, req->stream);
+    m3u8 = path + "/" + m3u8_url;
 
 
     // we always keep the target duration increasing.
     // we always keep the target duration increasing.
     int max_td = srs_max(target_duration, (int)(fragment * _srs_config->get_hls_td_ratio(r->vhost)));
     int max_td = srs_max(target_duration, (int)(fragment * _srs_config->get_hls_td_ratio(r->vhost)));
@@ -664,7 +667,10 @@ int SrsHlsMuxer::segment_close(string log_desc)
         segments.push_back(current);
         segments.push_back(current);
         
         
         // use async to call the http hooks, for it will cause thread switch.
         // use async to call the http hooks, for it will cause thread switch.
-        if ((ret = async->call(new SrsDvrAsyncCallOnHls(req, current->full_path, current->sequence_no, current->duration))) != ERROR_SUCCESS) {
+        if ((ret = async->call(new SrsDvrAsyncCallOnHls(req,
+            current->full_path, current->uri, m3u8, m3u8_url,
+            current->sequence_no, current->duration))) != ERROR_SUCCESS)
+        {
             return ret;
             return ret;
         }
         }
         
         

@@ -163,11 +163,14 @@ class SrsDvrAsyncCallOnHls : public ISrsDvrAsyncCall
 {
 {
 private:
 private:
     std::string path;
     std::string path;
+    std::string ts_url;
+    std::string m3u8;
+    std::string m3u8_url;
     int seq_no;
     int seq_no;
     SrsRequest* req;
     SrsRequest* req;
     double duration;
     double duration;
 public:
 public:
-    SrsDvrAsyncCallOnHls(SrsRequest* r, std::string p, int s, double d);
+    SrsDvrAsyncCallOnHls(SrsRequest* r, std::string p, std::string t, std::string m, std::string mu, int s, double d);
     virtual ~SrsDvrAsyncCallOnHls();
     virtual ~SrsDvrAsyncCallOnHls();
 public:
 public:
     virtual int call();
     virtual int call();
@@ -227,6 +230,7 @@ private:
     int _sequence_no;
     int _sequence_no;
     int target_duration;
     int target_duration;
     std::string m3u8;
     std::string m3u8;
+    std::string m3u8_url;
 private:
 private:
     ISrsHlsHandler* handler;
     ISrsHlsHandler* handler;
     // TODO: FIXME: supports reload.
     // TODO: FIXME: supports reload.

@@ -292,7 +292,7 @@ int SrsHttpHooks::on_dvr(string url, SrsRequest* req, string file)
     return ret;
     return ret;
 }
 }
 
 
-int SrsHttpHooks::on_hls(string url, SrsRequest* req, string file, int sn, double duration)
+int SrsHttpHooks::on_hls(string url, SrsRequest* req, string file, string ts_url, string m3u8, string m3u8_url, int sn, double duration)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
@@ -310,6 +310,9 @@ int SrsHttpHooks::on_hls(string url, SrsRequest* req, string file, int sn, doubl
         << SRS_JFIELD_ORG("duration", duration) << SRS_JFIELD_CONT
         << SRS_JFIELD_ORG("duration", duration) << SRS_JFIELD_CONT
         << SRS_JFIELD_STR("cwd", cwd) << SRS_JFIELD_CONT
         << SRS_JFIELD_STR("cwd", cwd) << SRS_JFIELD_CONT
         << SRS_JFIELD_STR("file", file) << SRS_JFIELD_CONT
         << SRS_JFIELD_STR("file", file) << SRS_JFIELD_CONT
+        << SRS_JFIELD_STR("url", ts_url) << SRS_JFIELD_CONT
+        << SRS_JFIELD_STR("m3u8", m3u8) << SRS_JFIELD_CONT
+        << SRS_JFIELD_STR("m3u8_url", m3u8_url) << SRS_JFIELD_CONT
         << SRS_JFIELD_ORG("seq_no", sn)
         << SRS_JFIELD_ORG("seq_no", sn)
         << SRS_JOBJECT_END;
         << SRS_JOBJECT_END;
         
         

@@ -101,10 +101,13 @@ public:
     * @param url the api server url, to process the event. 
     * @param url the api server url, to process the event. 
     *         ignore if empty.
     *         ignore if empty.
     * @param file the ts file path, can be relative or absolute path.
     * @param file the ts file path, can be relative or absolute path.
+    * @param ts_url the ts url, which used for m3u8.
+    * @param m3u8 the m3u8 file path, can be relative or absolute path.
+    * @param m3u8_url the m3u8 url, which is used for the http mount path.
     * @param sn the seq_no, the sequence number of ts in hls/m3u8.
     * @param sn the seq_no, the sequence number of ts in hls/m3u8.
     * @param duration the segment duration in seconds.
     * @param duration the segment duration in seconds.
     */
     */
-    static int on_hls(std::string url, SrsRequest* req, std::string file, int sn, double duration);
+    static int on_hls(std::string url, SrsRequest* req, std::string file, std::string ts_url, std::string m3u8, std::string m3u8_url, int sn, double duration);
     /**
     /**
      * when hls reap segment, callback.
      * when hls reap segment, callback.
      * @param url the api server url, to process the event.
      * @param url the api server url, to process the event.

@@ -48,9 +48,6 @@ using namespace std;
 #include <srs_rtmp_amf0.hpp>
 #include <srs_rtmp_amf0.hpp>
 #include <srs_raw_avc.hpp>
 #include <srs_raw_avc.hpp>
 
 
-// the retry timeout in ms.
-#define SRS_INGEST_HLS_ERROR_RETRY_MS 3000
-
 // pre-declare
 // pre-declare
 int proxy_hls2rtmp(std::string hls, std::string rtmp);
 int proxy_hls2rtmp(std::string hls, std::string rtmp);
 
 
@@ -217,7 +214,7 @@ private:
     /**
     /**
      * fetch all ts body.
      * fetch all ts body.
      */
      */
-    virtual void fetch_all_ts(bool fresh_m3u8);
+    virtual int fetch_all_ts(bool fresh_m3u8);
     /**
     /**
      * remove all ts which is dirty.
      * remove all ts which is dirty.
      */
      */
@@ -245,7 +242,10 @@ int SrsIngestSrsInput::connect()
     }
     }
     
     
     // fetch all ts.
     // fetch all ts.
-    fetch_all_ts(fresh_m3u8);
+    if ((ret = fetch_all_ts(fresh_m3u8)) != ERROR_SUCCESS) {
+        srs_error("fetch all ts failed. ret=%d", ret);
+        return ret;
+    }
     
     
     // remove all dirty ts.
     // remove all dirty ts.
     remove_dirty();
     remove_dirty();
@@ -304,15 +304,9 @@ int SrsIngestSrsInput::parseTs(ISrsTsHandler* handler, char* body, int nb_body)
         
         
         // process each ts packet
         // process each ts packet
         if ((ret = context->decode(stream, handler)) != ERROR_SUCCESS) {
         if ((ret = context->decode(stream, handler)) != ERROR_SUCCESS) {
-            // when peer closed, must interrupt parse and reconnect.
-            if (srs_is_client_gracefully_close(ret)) {
-                srs_warn("interrupt parse for peer closed. ret=%d", ret);
+            srs_error("mpegts: ignore parse ts packet failed. ret=%d", ret);
             return ret;
             return ret;
         }
         }
-            
-            srs_warn("mpegts: ignore parse ts packet failed. ret=%d", ret);
-            continue;
-        }
         srs_info("mpegts: parse ts packet completed");
         srs_info("mpegts: parse ts packet completed");
     }
     }
     srs_info("mpegts: parse udp packet completed");
     srs_info("mpegts: parse udp packet completed");
@@ -536,7 +530,7 @@ void SrsIngestSrsInput::dirty_all_ts()
     }
     }
 }
 }
 
 
-void SrsIngestSrsInput::fetch_all_ts(bool fresh_m3u8)
+int SrsIngestSrsInput::fetch_all_ts(bool fresh_m3u8)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
@@ -555,9 +549,9 @@ void SrsIngestSrsInput::fetch_all_ts(bool fresh_m3u8)
         }
         }
         
         
         if ((ret = tp->fetch(in_hls->get_url())) != ERROR_SUCCESS) {
         if ((ret = tp->fetch(in_hls->get_url())) != ERROR_SUCCESS) {
-            srs_warn("ignore ts %s for error. ret=%d", tp->url.c_str(), ret);
+            srs_error("fetch ts %s for error. ret=%d", tp->url.c_str(), ret);
             tp->skip = true;
             tp->skip = true;
-            continue;
+            return ret;
         }
         }
         
         
         // only wait for a duration of last piece.
         // only wait for a duration of last piece.
@@ -565,6 +559,8 @@ void SrsIngestSrsInput::fetch_all_ts(bool fresh_m3u8)
             next_connect_time = srs_update_system_time_ms() + (int)tp->duration * 1000;
             next_connect_time = srs_update_system_time_ms() + (int)tp->duration * 1000;
         }
         }
     }
     }
+    
+    return ret;
 }
 }
 
 
 
 
@@ -779,10 +775,6 @@ int SrsIngestSrsOutput::on_ts_message(SrsTsMessage* msg)
     // we must use queue to cache the msg, then parse it if possible.
     // we must use queue to cache the msg, then parse it if possible.
     queue.insert(std::make_pair(msg->dts, msg->detach()));
     queue.insert(std::make_pair(msg->dts, msg->detach()));
     if ((ret = parse_message_queue()) != ERROR_SUCCESS) {
     if ((ret = parse_message_queue()) != ERROR_SUCCESS) {
-        // when peer closed, close the output and reconnect.
-        if (srs_is_client_gracefully_close(ret)) {
-            close();
-        }
         return ret;
         return ret;
     }
     }
     
     
@@ -1204,6 +1196,7 @@ int SrsIngestSrsOutput::rtmp_write_packet(char type, u_int32_t timestamp, char*
     
     
     // send out encoded msg.
     // send out encoded msg.
     if ((ret = client->send_and_free_message(msg, stream_id)) != ERROR_SUCCESS) {
     if ((ret = client->send_and_free_message(msg, stream_id)) != ERROR_SUCCESS) {
+        srs_error("send RTMP type=%d, dts=%d, size=%d failed. ret=%d", type, timestamp, size, ret);
         return ret;
         return ret;
     }
     }
     
     
@@ -1355,22 +1348,22 @@ public:
         int ret = ERROR_SUCCESS;
         int ret = ERROR_SUCCESS;
         
         
         if ((ret = ic->connect()) != ERROR_SUCCESS) {
         if ((ret = ic->connect()) != ERROR_SUCCESS) {
-            srs_warn("connect oc failed. ret=%d", ret);
+            srs_error("connect oc failed. ret=%d", ret);
             return ret;
             return ret;
         }
         }
         
         
         if ((ret = oc->connect()) != ERROR_SUCCESS) {
         if ((ret = oc->connect()) != ERROR_SUCCESS) {
-            srs_warn("connect ic failed. ret=%d", ret);
+            srs_error("connect ic failed. ret=%d", ret);
             return ret;
             return ret;
         }
         }
         
         
         if ((ret = ic->parse(oc, oc)) != ERROR_SUCCESS) {
         if ((ret = ic->parse(oc, oc)) != ERROR_SUCCESS) {
-            srs_warn("proxy ts to rtmp failed. ret=%d", ret);
+            srs_error("proxy ts to rtmp failed. ret=%d", ret);
             return ret;
             return ret;
         }
         }
         
         
         if ((ret = oc->flush_message_queue()) != ERROR_SUCCESS) {
         if ((ret = oc->flush_message_queue()) != ERROR_SUCCESS) {
-            srs_warn("flush oc message failed. ret=%d", ret);
+            srs_error("flush oc message failed. ret=%d", ret);
             return ret;
             return ret;
         }
         }
         
         
@@ -1400,12 +1393,10 @@ int proxy_hls2rtmp(string hls, string rtmp)
     
     
     SrsIngestSrsContext context(&hls_uri, &rtmp_uri);
     SrsIngestSrsContext context(&hls_uri, &rtmp_uri);
     for (;;) {
     for (;;) {
-        if ((ret = context.proxy()) == ERROR_SUCCESS) {
-            continue;
+        if ((ret = context.proxy()) != ERROR_SUCCESS) {
+            srs_error("proxy hls to rtmp failed. ret=%d", ret);
+            return ret;
         }
         }
-        
-        srs_warn("proxy hls to rtmp failed. ret=%d", ret);
-        st_usleep(SRS_INGEST_HLS_ERROR_RETRY_MS * 1000);
     }
     }
     
     
     return ret;
     return ret;