for bug #293, http live streaming framework.

@@ -41,6 +41,7 @@ using namespace std;
 #include <srs_kernel_utility.hpp>
 #include <srs_kernel_utility.hpp>
 #include <srs_kernel_file.hpp>
 #include <srs_kernel_file.hpp>
 #include <srs_kernel_flv.hpp>
 #include <srs_kernel_flv.hpp>
+#include <srs_protocol_rtmp.hpp>
 
 
 SrsVodStream::SrsVodStream(string root_dir)
 SrsVodStream::SrsVodStream(string root_dir)
     : SrsGoHttpFileServer(root_dir)
     : SrsGoHttpFileServer(root_dir)
@@ -177,6 +178,39 @@ int SrsHttpServer::initialize()
     return ret;
     return ret;
 }
 }
 
 
+int SrsHttpServer::mount(SrsSource* s, SrsRequest* r)
+{
+    int ret = ERROR_SUCCESS;
+    
+    if (flvs.empty()) {
+        srs_info("ignore mount, no flv stream configed.");
+        return ret;
+    }
+    
+    if (flvs.find(r->vhost) == flvs.end()) {
+        srs_info("ignore mount flv stream for disabled");
+        return ret;
+    }
+    
+    // TODO: FIXME: implements it.
+    return ret;
+}
+
+void SrsHttpServer::unmount(SrsSource* s, SrsRequest* r)
+{
+    if (flvs.empty()) {
+        srs_info("ignore unmount, no flv stream configed.");
+        return;
+    }
+    
+    if (flvs.find(r->vhost) == flvs.end()) {
+        srs_info("ignore unmount flv stream for disabled");
+        return;
+    }
+    
+    // TODO: FIXME: implements it.
+}
+
 int SrsHttpServer::on_reload_vhost_http_updated()
 int SrsHttpServer::on_reload_vhost_http_updated()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;

@@ -37,6 +37,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_app_http.hpp>
 #include <srs_app_http.hpp>
 #include <srs_app_reload.hpp>
 #include <srs_app_reload.hpp>
 
 
+class SrsSource;
+class SrsRequest;
 class SrsStSocket;
 class SrsStSocket;
 class SrsHttpParser;
 class SrsHttpParser;
 class SrsHttpMessage;
 class SrsHttpMessage;
@@ -85,6 +87,9 @@ public:
     virtual ~SrsHttpServer();
     virtual ~SrsHttpServer();
 public:
 public:
     virtual int initialize();
     virtual int initialize();
+public:
+    virtual int mount(SrsSource* s, SrsRequest* r);
+    virtual void unmount(SrsSource* s, SrsRequest* r);
 // interface ISrsThreadHandler.
 // interface ISrsThreadHandler.
 public:
 public:
     virtual int on_reload_vhost_http_updated();
     virtual int on_reload_vhost_http_updated();

@@ -393,7 +393,7 @@ int SrsRtmpConn::stream_service_cycle()
     
     
     // find a source to serve.
     // find a source to serve.
     SrsSource* source = NULL;
     SrsSource* source = NULL;
-    if ((ret = SrsSource::find(req, &source)) != ERROR_SUCCESS) {
+    if ((ret = SrsSource::find(req, server, &source)) != ERROR_SUCCESS) {
         return ret;
         return ret;
     }
     }
     srs_assert(source != NULL);
     srs_assert(source != NULL);

@@ -1074,3 +1074,23 @@ int SrsServer::on_reload_http_stream_updated()
     return ret;
     return ret;
 }
 }
 
 
+int SrsServer::on_publish(SrsSource* s, SrsRequest* r)
+{
+    int ret = ERROR_SUCCESS;
+    
+#ifdef SRS_AUTO_HTTP_SERVER
+    if ((ret = http_stream_mux->mount(s, r)) != ERROR_SUCCESS) {
+        return ret;
+    }
+#endif
+    
+    return ret;
+}
+
+void SrsServer::on_unpublish(SrsSource* s, SrsRequest* r)
+{
+#ifdef SRS_AUTO_HTTP_SERVER
+    http_stream_mux->unmount(s, r);
+#endif
+}
+

@@ -35,6 +35,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_app_st.hpp>
 #include <srs_app_st.hpp>
 #include <srs_app_reload.hpp>
 #include <srs_app_reload.hpp>
 #include <srs_app_thread.hpp>
 #include <srs_app_thread.hpp>
+#include <srs_app_source.hpp>
 
 
 class SrsServer;
 class SrsServer;
 class SrsConnection;
 class SrsConnection;
@@ -113,7 +114,8 @@ private:
 * SRS RTMP server, initialize and listen, 
 * SRS RTMP server, initialize and listen, 
 * start connection service thread, destroy client.
 * start connection service thread, destroy client.
 */
 */
-class SrsServer : public ISrsReloadHandler
+class SrsServer : virtual public ISrsReloadHandler
+    , virtual public ISrsSourceHandler
 {
 {
 private:
 private:
 #ifdef SRS_AUTO_HTTP_API
 #ifdef SRS_AUTO_HTTP_API
@@ -241,6 +243,10 @@ public:
     virtual int on_reload_http_stream_enabled();
     virtual int on_reload_http_stream_enabled();
     virtual int on_reload_http_stream_disabled();
     virtual int on_reload_http_stream_disabled();
     virtual int on_reload_http_stream_updated();
     virtual int on_reload_http_stream_updated();
+// interface ISrsSourceHandler
+public:
+    virtual int on_publish(SrsSource* s, SrsRequest* r);
+    virtual void on_unpublish(SrsSource* s, SrsRequest* r);
 };
 };
 
 
 #endif
 #endif

@@ -672,24 +672,33 @@ bool SrsGopCache::pure_audio()
     return cached_video_count == 0;
     return cached_video_count == 0;
 }
 }
 
 
+ISrsSourceHandler::ISrsSourceHandler()
+{
+}
+
+ISrsSourceHandler::~ISrsSourceHandler()
+{
+}
+
 std::map<std::string, SrsSource*> SrsSource::pool;
 std::map<std::string, SrsSource*> SrsSource::pool;
 
 
-int SrsSource::find(SrsRequest* req, SrsSource** ppsource)
+int SrsSource::find(SrsRequest* r, ISrsSourceHandler* h, SrsSource** pps)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    string stream_url = req->get_stream_url();
-    string vhost = req->vhost;
+    string stream_url = r->get_stream_url();
+    string vhost = r->vhost;
     
     
     if (pool.find(stream_url) == pool.end()) {
     if (pool.find(stream_url) == pool.end()) {
-        SrsSource* source = new SrsSource(req);
-        if ((ret = source->initialize()) != ERROR_SUCCESS) {
+        SrsSource* source = new SrsSource();
+        if ((ret = source->initialize(r, h)) != ERROR_SUCCESS) {
             srs_freep(source);
             srs_freep(source);
             return ret;
             return ret;
         }
         }
         
         
         pool[stream_url] = source;
         pool[stream_url] = source;
-        srs_info("create new source for url=%s, vhost=%s", stream_url.c_str(), vhost.c_str());
+        srs_info("create new source for url=%s, vhost=%s", 
+            stream_url.c_str(), vhost.c_str());
     }
     }
     
     
     // we always update the request of resource, 
     // we always update the request of resource, 
@@ -697,8 +706,8 @@ int SrsSource::find(SrsRequest* req, SrsSource** ppsource)
     // and we only need to update the token of request, it's simple.
     // and we only need to update the token of request, it's simple.
     if (true) {
     if (true) {
         SrsSource* source = pool[stream_url];
         SrsSource* source = pool[stream_url];
-        source->_req->update_auth(req);
-        *ppsource = source;
+        source->_req->update_auth(r);
+        *pps = source;
     }
     }
     
     
     return ret;
     return ret;
@@ -714,9 +723,9 @@ void SrsSource::destroy()
     pool.clear();
     pool.clear();
 }
 }
 
 
-SrsSource::SrsSource(SrsRequest* req)
+SrsSource::SrsSource()
 {
 {
-    _req = req->copy();
+    _req = NULL;
     jitter_algorithm = SrsRtmpJitterAlgorithmOFF;
     jitter_algorithm = SrsRtmpJitterAlgorithmOFF;
     
     
 #ifdef SRS_AUTO_HLS
 #ifdef SRS_AUTO_HLS
@@ -741,7 +750,7 @@ SrsSource::SrsSource(SrsRequest* req)
     aggregate_stream = new SrsStream();
     aggregate_stream = new SrsStream();
     
     
     _srs_config->subscribe(this);
     _srs_config->subscribe(this);
-    atc = _srs_config->get_atc(_req->vhost);
+    atc = false;
 }
 }
 
 
 SrsSource::~SrsSource()
 SrsSource::~SrsSource()
@@ -783,10 +792,14 @@ SrsSource::~SrsSource()
     srs_freep(_req);
     srs_freep(_req);
 }
 }
 
 
-int SrsSource::initialize()
+int SrsSource::initialize(SrsRequest* r, ISrsSourceHandler* h)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
+    handler = h;
+    _req = r->copy();
+    atc = _srs_config->get_atc(_req->vhost);
+    
 #ifdef SRS_AUTO_DVR
 #ifdef SRS_AUTO_DVR
     if ((ret = dvr->initialize(_req)) != ERROR_SUCCESS) {
     if ((ret = dvr->initialize(_req)) != ERROR_SUCCESS) {
         return ret;
         return ret;
@@ -1643,6 +1656,13 @@ int SrsSource::on_publish()
     }
     }
 #endif
 #endif
 
 
+    // notify the handler.
+    srs_assert(handler);
+    if ((ret = handler->on_publish(this, _req)) != ERROR_SUCCESS) {
+        srs_error("handle on publish failed. ret=%d", ret);
+        return ret;
+    }
+
     return ret;
     return ret;
 }
 }
 
 
@@ -1676,6 +1696,10 @@ void SrsSource::on_unpublish()
     
     
     _can_publish = true;
     _can_publish = true;
     _source_id = -1;
     _source_id = -1;
+
+    // notify the handler.
+    srs_assert(handler);
+    handler->on_unpublish(this, _req);
 }
 }
 
 
 int SrsSource::create_consumer(SrsConsumer*& consumer)
 int SrsSource::create_consumer(SrsConsumer*& consumer)

@@ -337,6 +337,27 @@ public:
 };
 };
 
 
 /**
 /**
+* the handler to handle the event of srs source.
+* for example, the http flv streaming module handle the event and 
+* mount http when rtmp start publishing.
+*/
+class ISrsSourceHandler
+{
+public:
+    ISrsSourceHandler();
+    virtual ~ISrsSourceHandler();
+public:
+    /**
+    * when stream start publish, mount stream.
+    */
+    virtual int on_publish(SrsSource* s, SrsRequest* r) = 0;
+    /**
+    * when stream stop publish, unmount stream.
+    */
+    virtual void on_unpublish(SrsSource* s, SrsRequest* r) = 0;
+};
+
+/**
 * live streaming source.
 * live streaming source.
 */
 */
 class SrsSource : public ISrsReloadHandler
 class SrsSource : public ISrsReloadHandler
@@ -346,11 +367,11 @@ private:
 public:
 public:
     /**
     /**
     * find stream by vhost/app/stream.
     * find stream by vhost/app/stream.
-    * @param req the client request.
-    * @param ppsource the matched source, if success never be NULL.
-    * @remark stream_url should without port and schema.
+    * @param r the client request.
+    * @param h the event handler for source.
+    * @param pps the matched source, if success never be NULL.
     */
     */
-    static int find(SrsRequest* req, SrsSource** ppsource);
+    static int find(SrsRequest* r, ISrsSourceHandler* h, SrsSource** pps);
     /**
     /**
     * when system exit, destroy the sources,
     * when system exit, destroy the sources,
     * for gmc to analysis mem leaks.
     * for gmc to analysis mem leaks.
@@ -390,6 +411,8 @@ private:
     std::vector<SrsForwarder*> forwarders;
     std::vector<SrsForwarder*> forwarders;
     // for aggregate message
     // for aggregate message
     SrsStream* aggregate_stream;
     SrsStream* aggregate_stream;
+    // the event handler.
+    ISrsSourceHandler* handler;
 private:
 private:
     /**
     /**
     * the sample rate of audio in metadata.
     * the sample rate of audio in metadata.
@@ -421,10 +444,11 @@ public:
     * @param _req the client request object, 
     * @param _req the client request object, 
     *     this object will deep copy it for reload.
     *     this object will deep copy it for reload.
     */
     */
-    SrsSource(SrsRequest* req);
+    SrsSource();
     virtual ~SrsSource();
     virtual ~SrsSource();
+// initialize, get and setter.
 public:
 public:
-    virtual int initialize();
+    virtual int initialize(SrsRequest* r, ISrsSourceHandler* h);
 // interface ISrsReloadHandler
 // interface ISrsReloadHandler
 public:
 public:
     virtual int on_reload_vhost_atc(std::string vhost);
     virtual int on_reload_vhost_atc(std::string vhost);