support ingest reload: add new vhost with ingester

@@ -493,8 +493,36 @@ int SrsConfig::reload()
         srs_trace("reload pithy_print success.");
         srs_trace("reload pithy_print success.");
     }
     }
 
 
-    // merge config: vhost added, directly supported.
+    // merge config: vhost added
+    for (int i = 0; i < (int)root->directives.size(); i++) {
+        // ingest need to start if specified.
+        // other features, directly supported.
+        SrsConfDirective* new_vhost = root->at(i);
+        
+        // only process vhost directives.
+        if (new_vhost->name != "vhost") {
+            continue;
+        }
+        
+        std::string vhost = new_vhost->arg0();
+        
+        // not new added vhost, ignore.
+        if (old_root->get("vhost", vhost)) {
+            continue;
+        }
         
         
+        srs_trace("vhost %s added, reload it.", vhost.c_str());
+        for (it = subscribes.begin(); it != subscribes.end(); ++it) {
+            ISrsReloadHandler* subscribe = *it;
+            if ((ret = subscribe->on_reload_vhost_added(vhost)) != ERROR_SUCCESS) {
+                srs_error("notify subscribes pithy_print remove "
+                    "vhost %s failed. ret=%d", vhost.c_str(), ret);
+                return ret;
+            }
+        }
+        srs_trace("reload new vhost %s success.", vhost.c_str());
+    }
+    
     // merge config: vhost removed/disabled/modified.
     // merge config: vhost removed/disabled/modified.
     for (int i = 0; i < (int)old_root->directives.size(); i++) {
     for (int i = 0; i < (int)old_root->directives.size(); i++) {
         SrsConfDirective* old_vhost = old_root->at(i);
         SrsConfDirective* old_vhost = old_root->at(i);

@@ -51,13 +51,16 @@ SrsIngesterFFMPEG::~SrsIngesterFFMPEG()
 
 
 SrsIngester::SrsIngester()
 SrsIngester::SrsIngester()
 {
 {
-    // TODO: FIXME: support reload.
+    _srs_config->subscribe(this);
+    
     pthread = new SrsThread(this, SRS_INGESTER_SLEEP_US);
     pthread = new SrsThread(this, SRS_INGESTER_SLEEP_US);
     pithy_print = new SrsPithyPrint(SRS_STAGE_INGESTER);
     pithy_print = new SrsPithyPrint(SRS_STAGE_INGESTER);
 }
 }
 
 
 SrsIngester::~SrsIngester()
 SrsIngester::~SrsIngester()
 {
 {
+    _srs_config->unsubscribe(this);
+    
     srs_freep(pthread);
     srs_freep(pthread);
     clear_engines();
     clear_engines();
 }
 }
@@ -72,10 +75,8 @@ int SrsIngester::start()
         return ret;
         return ret;
     }
     }
     
     
-    // return for error or no engine.
-    if (ingesters.empty()) {
-        return ret;
-    }
+    // even no ingesters, we must also start it,
+    // for the reload may add more ingesters.
     
     
     // start thread to run all encoding engines.
     // start thread to run all encoding engines.
     if ((ret = pthread->start()) != ERROR_SUCCESS) {
     if ((ret = pthread->start()) != ERROR_SUCCESS) {
@@ -353,4 +354,16 @@ void SrsIngester::ingester()
     }
     }
 }
 }
 
 
+int SrsIngester::on_reload_vhost_added(string vhost)
+{
+    int ret = ERROR_SUCCESS;
+    
+    SrsConfDirective* _vhost = _srs_config->get_vhost(vhost);
+    if ((ret = parse_ingesters(_vhost)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    return ret;
+}
+
 #endif
 #endif

@@ -34,6 +34,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <vector>
 #include <vector>
 
 
 #include <srs_app_thread.hpp>
 #include <srs_app_thread.hpp>
+#include <srs_app_reload.hpp>
 
 
 class SrsFFMPEG;
 class SrsFFMPEG;
 class SrsConfDirective;
 class SrsConfDirective;
@@ -57,7 +58,7 @@ struct SrsIngesterFFMPEG
 * encode with FFMPEG(optional),
 * encode with FFMPEG(optional),
 * push to SRS(or any RTMP server) over RTMP.
 * push to SRS(or any RTMP server) over RTMP.
 */
 */
-class SrsIngester : public ISrsThreadHandler
+class SrsIngester : public ISrsThreadHandler, public ISrsReloadHandler
 {
 {
 private:
 private:
     std::string input_stream_name;
     std::string input_stream_name;
@@ -82,6 +83,9 @@ private:
     virtual int parse_engines(SrsConfDirective* vhost, SrsConfDirective* ingest);
     virtual int parse_engines(SrsConfDirective* vhost, SrsConfDirective* ingest);
     virtual int initialize_ffmpeg(SrsFFMPEG* ffmpeg, SrsConfDirective* vhost, SrsConfDirective* ingest, SrsConfDirective* engine);
     virtual int initialize_ffmpeg(SrsFFMPEG* ffmpeg, SrsConfDirective* vhost, SrsConfDirective* ingest, SrsConfDirective* engine);
     virtual void ingester();
     virtual void ingester();
+// interface ISrsReloadHandler.
+public:
+    virtual int on_reload_vhost_added(std::string vhost);
 };
 };
 
 
 #endif
 #endif

@@ -45,6 +45,11 @@ int ISrsReloadHandler::on_reload_pithy_print()
     return ERROR_SUCCESS;
     return ERROR_SUCCESS;
 }
 }
 
 
+int ISrsReloadHandler::on_reload_vhost_added(string /*vhost*/)
+{
+    return ERROR_SUCCESS;
+}
+
 int ISrsReloadHandler::on_reload_vhost_removed(string /*vhost*/)
 int ISrsReloadHandler::on_reload_vhost_removed(string /*vhost*/)
 {
 {
     return ERROR_SUCCESS;
     return ERROR_SUCCESS;

@@ -33,6 +33,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 /**
 /**
 * the handler for config reload.
 * the handler for config reload.
+* when reload callback, the config is updated yet.
 */
 */
 class ISrsReloadHandler
 class ISrsReloadHandler
 {
 {
@@ -42,6 +43,7 @@ public:
 public:
 public:
     virtual int on_reload_listen();
     virtual int on_reload_listen();
     virtual int on_reload_pithy_print();
     virtual int on_reload_pithy_print();
+    virtual int on_reload_vhost_added(std::string vhost);
     virtual int on_reload_vhost_removed(std::string vhost);
     virtual int on_reload_vhost_removed(std::string vhost);
     virtual int on_reload_gop_cache(std::string vhost);
     virtual int on_reload_gop_cache(std::string vhost);
     virtual int on_reload_queue_length(std::string vhost);
     virtual int on_reload_queue_length(std::string vhost);