implements the ingest framework

@@ -97,7 +97,8 @@ vhost ingest.srs.com {
         # whether enable ingest features
         # whether enable ingest features
         # default: off
         # default: off
         enable      on;
         enable      on;
-        # input file/stream/device, can be multiple input.
+        # input file/stream/device
+        # @remark only support one input.
         input {
         input {
             # the type of input.
             # the type of input.
             # can be file/stream/device, that is,
             # can be file/stream/device, that is,

@@ -1645,6 +1645,17 @@ string SrsConfig::get_ingest_ffmpeg(SrsConfDirective* ingest)
     return conf->arg0();
     return conf->arg0();
 }
 }
 
 
+string SrsConfig::get_ingest_input(SrsConfDirective* ingest)
+{
+    SrsConfDirective* conf = ingest->get("input");
+    
+    if (!conf) {
+        return "";
+    }
+    
+    return conf->arg0();
+}
+
 string SrsConfig::get_srs_log_file()
 string SrsConfig::get_srs_log_file()
 {
 {
     srs_assert(root);
     srs_assert(root);

@@ -190,6 +190,7 @@ public:
     virtual void                get_ingesters(std::string vhost, std::vector<SrsConfDirective*>& ingeters);
     virtual void                get_ingesters(std::string vhost, std::vector<SrsConfDirective*>& ingeters);
     virtual bool                get_ingest_enabled(SrsConfDirective* ingest);
     virtual bool                get_ingest_enabled(SrsConfDirective* ingest);
     virtual std::string         get_ingest_ffmpeg(SrsConfDirective* ingest);
     virtual std::string         get_ingest_ffmpeg(SrsConfDirective* ingest);
+    virtual std::string         get_ingest_input(SrsConfDirective* ingest);
 // log section
 // log section
 public:
 public:
     virtual bool                get_srs_log_tank_file();
     virtual bool                get_srs_log_tank_file();

@@ -49,16 +49,11 @@ int SrsIngester::start()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    // parse ingesters
-    std::vector<SrsConfDirective*> vhosts;
-    _srs_config->get_vhosts(vhosts);
-    
-    for (int i = 0; i < (int)vhosts.size(); i++) {
-        SrsConfDirective* vhost = vhosts[i];
-        if ((ret = parse_ingesters(vhost)) != ERROR_SUCCESS) {
+    if ((ret = parse()) != ERROR_SUCCESS) {
+        clear_engines();
+        ret = ERROR_SUCCESS;
         return ret;
         return ret;
     }
     }
-    }
     
     
     return ret;
     return ret;
 }
 }
@@ -77,18 +72,40 @@ int SrsIngester::parse_ingesters(SrsConfDirective* vhost)
             continue;
             continue;
         }
         }
         
         
+        if ((ret = parse_engines(vhost, ingest)) != ERROR_SUCCESS) {
+            return ret;
+        }
+    }
+    
+    return ret;
+}
+
+int SrsIngester::parse_engines(SrsConfDirective* vhost, SrsConfDirective* ingest)
+{
+    int ret = ERROR_SUCCESS;
+        
     std::string ffmpeg_bin = _srs_config->get_ingest_ffmpeg(ingest);
     std::string ffmpeg_bin = _srs_config->get_ingest_ffmpeg(ingest);
     if (ffmpeg_bin.empty()) {
     if (ffmpeg_bin.empty()) {
-            srs_trace("ignore the empty ffmpeg ingest: %s", ingest->arg0().c_str());
-            continue;
+        ret = ERROR_ENCODER_PARSE;
+        srs_trace("empty ffmpeg ret=%d", ret);
+        return ret;
     }
     }
     
     
     // get all engines.
     // get all engines.
     std::vector<SrsConfDirective*> engines;
     std::vector<SrsConfDirective*> engines;
     _srs_config->get_transcode_engines(ingest, engines);
     _srs_config->get_transcode_engines(ingest, engines);
     if (engines.empty()) {
     if (engines.empty()) {
-            srs_trace("ignore the empty transcode engine: %s", ingest->arg0().c_str());
-            continue;
+        SrsFFMPEG* ffmpeg = new SrsFFMPEG(ffmpeg_bin);
+        if ((ret = initialize_ffmpeg(ffmpeg, ingest, NULL)) != ERROR_SUCCESS) {
+            srs_freep(ffmpeg);
+            if (ret != ERROR_ENCODER_LOOP) {
+                srs_error("invalid ingest engine. ret=%d", ret);
+            }
+            return ret;
+        }
+
+        ffmpegs.push_back(ffmpeg);
+        return ret;
     }
     }
 
 
     // create engine
     // create engine
@@ -105,7 +122,6 @@ int SrsIngester::parse_ingesters(SrsConfDirective* vhost)
 
 
         ffmpegs.push_back(ffmpeg);
         ffmpegs.push_back(ffmpeg);
     }
     }
-    }
     
     
     return ret;
     return ret;
 }
 }
@@ -136,11 +152,36 @@ void SrsIngester::clear_engines()
     ffmpegs.clear();
     ffmpegs.clear();
 }
 }
 
 
+int SrsIngester::parse()
+{
+    int ret = ERROR_SUCCESS;
+    
+    // parse ingesters
+    std::vector<SrsConfDirective*> vhosts;
+    _srs_config->get_vhosts(vhosts);
+    
+    for (int i = 0; i < (int)vhosts.size(); i++) {
+        SrsConfDirective* vhost = vhosts[i];
+        if ((ret = parse_ingesters(vhost)) != ERROR_SUCCESS) {
+            return ret;
+        }
+    }
+    
+    return ret;
+}
+
 int SrsIngester::initialize_ffmpeg(SrsFFMPEG* ffmpeg, SrsConfDirective* ingest, SrsConfDirective* engine)
 int SrsIngester::initialize_ffmpeg(SrsFFMPEG* ffmpeg, SrsConfDirective* ingest, SrsConfDirective* engine)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    if (!_srs_config->get_engine_enabled(engine)) {
+    std::string input = _srs_config->get_ingest_input(ingest);
+    if (input.empty()) {
+        ret = ERROR_ENCODER_NO_INPUT;
+        srs_trace("empty ingest intput. ret=%d", ret);
+        return ret;
+    }
+    
+    if (!engine || !_srs_config->get_engine_enabled(engine)) {
     }
     }
     
     
     return ret;
     return ret;

@@ -61,7 +61,9 @@ public:
     virtual void on_thread_stop();
     virtual void on_thread_stop();
 private:
 private:
     virtual void clear_engines();
     virtual void clear_engines();
+    virtual int parse();
     virtual int parse_ingesters(SrsConfDirective* vhost);
     virtual int parse_ingesters(SrsConfDirective* vhost);
+    virtual int parse_engines(SrsConfDirective* vhost, SrsConfDirective* ingest);
     virtual int initialize_ffmpeg(SrsFFMPEG* ffmpeg, SrsConfDirective* ingest, SrsConfDirective* engine);
     virtual int initialize_ffmpeg(SrsFFMPEG* ffmpeg, SrsConfDirective* ingest, SrsConfDirective* engine);
 };
 };
 
 

@@ -153,6 +153,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define ERROR_ENCODER_LOOP                  714
 #define ERROR_ENCODER_LOOP                  714
 #define ERROR_ENCODER_OPEN                  715
 #define ERROR_ENCODER_OPEN                  715
 #define ERROR_ENCODER_DUP2                  716
 #define ERROR_ENCODER_DUP2                  716
+#define ERROR_ENCODER_PARSE                 717
+#define ERROR_ENCODER_NO_INPUT              718
 
 
 #define ERROR_HTTP_PARSE_URI                800
 #define ERROR_HTTP_PARSE_URI                800
 #define ERROR_HTTP_DATA_INVLIAD             801
 #define ERROR_HTTP_DATA_INVLIAD             801