refine edge ingester, use upstream adapter.

@@ -62,13 +62,100 @@ using namespace std;
 // when edge error, wait for quit
 // when edge error, wait for quit
 #define SRS_EDGE_FORWARDER_ERROR_US (int64_t)(50*1000LL)
 #define SRS_EDGE_FORWARDER_ERROR_US (int64_t)(50*1000LL)
 
 
+SrsEdgeUpstream::SrsEdgeUpstream()
+{
+}
+
+SrsEdgeUpstream::~SrsEdgeUpstream()
+{
+}
+
+SrsEdgeRtmpUpstream::SrsEdgeRtmpUpstream()
+{
+    sdk = new SrsSimpleRtmpClient();
+}
+
+SrsEdgeRtmpUpstream::~SrsEdgeRtmpUpstream()
+{
+    close();
+    
+    srs_freep(sdk);
+}
+
+int SrsEdgeRtmpUpstream::connect(SrsRequest* r, SrsLbRoundRobin* lb)
+{
+    int ret = ERROR_SUCCESS;
+    
+    SrsRequest* req = r;
+    
+    std::string url;
+    if (true) {
+        SrsConfDirective* conf = _srs_config->get_vhost_edge_origin(req->vhost);
+        
+        // @see https://github.com/ossrs/srs/issues/79
+        // when origin is error, for instance, server is shutdown,
+        // then user remove the vhost then reload, the conf is empty.
+        if (!conf) {
+            ret = ERROR_EDGE_VHOST_REMOVED;
+            srs_warn("vhost %s removed. ret=%d", req->vhost.c_str(), ret);
+            return ret;
+        }
+        
+        // select the origin.
+        std::string server = lb->select(conf->args);
+        int port = SRS_CONSTS_RTMP_DEFAULT_PORT;
+        srs_parse_hostport(server, server, port);
+        
+        // support vhost tranform for edge,
+        // @see https://github.com/ossrs/srs/issues/372
+        std::string vhost = _srs_config->get_vhost_edge_transform_vhost(req->vhost);
+        vhost = srs_string_replace(vhost, "[vhost]", req->vhost);
+        
+        url = srs_generate_rtmp_url(server, port, vhost, req->app, req->stream);
+    }
+    
+    int64_t cto = SRS_EDGE_INGESTER_TIMEOUT_US;
+    int64_t sto = SRS_CONSTS_RTMP_PULSE_TIMEOUT_US;
+    if ((ret = sdk->connect(url, cto, sto)) != ERROR_SUCCESS) {
+        srs_error("edge pull %s failed, cto=%"PRId64", sto=%"PRId64". ret=%d", url.c_str(), cto, sto, ret);
+        return ret;
+    }
+    
+    if ((ret = sdk->play()) != ERROR_SUCCESS) {
+        srs_error("edge pull %s stream failed. ret=%d", url.c_str(), ret);
+        return ret;
+    }
+    
+    return ret;
+}
+
+int SrsEdgeRtmpUpstream::recv_message(SrsCommonMessage** pmsg)
+{
+    return sdk->recv_message(pmsg);
+}
+
+int SrsEdgeRtmpUpstream::decode_message(SrsCommonMessage* msg, SrsPacket** ppacket)
+{
+    return sdk->decode_message(msg, ppacket);
+}
+
+void SrsEdgeRtmpUpstream::close()
+{
+    sdk->close();
+}
+
+void SrsEdgeRtmpUpstream::kbps_sample(const char* label, int64_t age)
+{
+    sdk->kbps_sample(label, age);
+}
+
 SrsEdgeIngester::SrsEdgeIngester()
 SrsEdgeIngester::SrsEdgeIngester()
 {
 {
     source = NULL;
     source = NULL;
     edge = NULL;
     edge = NULL;
     req = NULL;
     req = NULL;
     
     
-    sdk = new SrsSimpleRtmpClient();
+    upstream = new SrsEdgeRtmpUpstream();
     lb = new SrsLbRoundRobin();
     lb = new SrsLbRoundRobin();
     pthread = new SrsReusableThread2("edge-igs", this, SRS_EDGE_INGESTER_SLEEP_US);
     pthread = new SrsReusableThread2("edge-igs", this, SRS_EDGE_INGESTER_SLEEP_US);
 }
 }
@@ -77,7 +164,7 @@ SrsEdgeIngester::~SrsEdgeIngester()
 {
 {
     stop();
     stop();
     
     
-    srs_freep(sdk);
+    srs_freep(upstream);
     srs_freep(lb);
     srs_freep(lb);
     srs_freep(pthread);
     srs_freep(pthread);
 }
 }
@@ -108,7 +195,7 @@ int SrsEdgeIngester::start()
 void SrsEdgeIngester::stop()
 void SrsEdgeIngester::stop()
 {
 {
     pthread->stop();
     pthread->stop();
-    sdk->close();
+    upstream->close();
     
     
     // notice to unpublish.
     // notice to unpublish.
     source->on_unpublish();
     source->on_unpublish();
@@ -122,44 +209,12 @@ string SrsEdgeIngester::get_curr_origin()
 int SrsEdgeIngester::cycle()
 int SrsEdgeIngester::cycle()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
-
-    source->on_source_id_changed(_srs_context->get_id());
-        
-    std::string url;
-    if (true) {
-        SrsConfDirective* conf = _srs_config->get_vhost_edge_origin(req->vhost);
-        
-        // @see https://github.com/ossrs/srs/issues/79
-        // when origin is error, for instance, server is shutdown,
-        // then user remove the vhost then reload, the conf is empty.
-        if (!conf) {
-            ret = ERROR_EDGE_VHOST_REMOVED;
-            srs_warn("vhost %s removed. ret=%d", req->vhost.c_str(), ret);
-            return ret;
-        }
-        
-        // select the origin.
-        std::string server = lb->select(conf->args);
-        int port = SRS_CONSTS_RTMP_DEFAULT_PORT;
-        srs_parse_hostport(server, server, port);
-        
-        // support vhost tranform for edge,
-        // @see https://github.com/ossrs/srs/issues/372
-        std::string vhost = _srs_config->get_vhost_edge_transform_vhost(req->vhost);
-        vhost = srs_string_replace(vhost, "[vhost]", req->vhost);
-        
-        url = srs_generate_rtmp_url(server, port, vhost, req->app, req->stream);
-    }
     
     
-    int64_t cto = SRS_EDGE_INGESTER_TIMEOUT_US;
-    int64_t sto = SRS_CONSTS_RTMP_PULSE_TIMEOUT_US;
-    if ((ret = sdk->connect(url, cto, sto)) != ERROR_SUCCESS) {
-        srs_error("edge pull %s failed, cto=%"PRId64", sto=%"PRId64". ret=%d", url.c_str(), cto, sto, ret);
+    if ((ret = source->on_source_id_changed(_srs_context->get_id())) != ERROR_SUCCESS) {
         return ret;
         return ret;
     }
     }
     
     
-    if ((ret = sdk->play()) != ERROR_SUCCESS) {
-        srs_error("edge pull %s stream failed. ret=%d", url.c_str(), ret);
+    if ((ret = upstream->connect(req, lb)) != ERROR_SUCCESS) {
         return ret;
         return ret;
     }
     }
     
     
@@ -188,12 +243,12 @@ int SrsEdgeIngester::ingest()
         
         
         // pithy print
         // pithy print
         if (pprint->can_print()) {
         if (pprint->can_print()) {
-            sdk->kbps_sample(SRS_CONSTS_LOG_EDGE_PLAY, pprint->age());
+            upstream->kbps_sample(SRS_CONSTS_LOG_EDGE_PLAY, pprint->age());
         }
         }
         
         
         // read from client.
         // read from client.
         SrsCommonMessage* msg = NULL;
         SrsCommonMessage* msg = NULL;
-        if ((ret = sdk->recv_message(&msg)) != ERROR_SUCCESS) {
+        if ((ret = upstream->recv_message(&msg)) != ERROR_SUCCESS) {
             if (!srs_is_client_gracefully_close(ret)) {
             if (!srs_is_client_gracefully_close(ret)) {
                 srs_error("pull origin server message failed. ret=%d", ret);
                 srs_error("pull origin server message failed. ret=%d", ret);
             }
             }
@@ -244,7 +299,7 @@ int SrsEdgeIngester::process_publish_message(SrsCommonMessage* msg)
     // process onMetaData
     // process onMetaData
     if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
     if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
         SrsPacket* pkt = NULL;
         SrsPacket* pkt = NULL;
-        if ((ret = sdk->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
+        if ((ret = upstream->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("decode onMetaData message failed. ret=%d", ret);
             srs_error("decode onMetaData message failed. ret=%d", ret);
             return ret;
             return ret;
         }
         }

@@ -49,6 +49,7 @@ class SrsKbps;
 class SrsLbRoundRobin;
 class SrsLbRoundRobin;
 class SrsTcpClient;
 class SrsTcpClient;
 class SrsSimpleRtmpClient;
 class SrsSimpleRtmpClient;
+class SrsPacket;
 
 
 /**
 /**
 * the state of edge, auto machine
 * the state of edge, auto machine
@@ -76,6 +77,37 @@ enum SrsEdgeUserState
 };
 };
 
 
 /**
 /**
+ * the upstream of edge, can be rtmp or http.
+ */
+class SrsEdgeUpstream
+{
+public:
+    SrsEdgeUpstream();
+    virtual ~SrsEdgeUpstream();
+public:
+    virtual int connect(SrsRequest* r, SrsLbRoundRobin* lb) = 0;
+    virtual int recv_message(SrsCommonMessage** pmsg) = 0;
+    virtual int decode_message(SrsCommonMessage* msg, SrsPacket** ppacket) = 0;
+    virtual void close() = 0;
+    virtual void kbps_sample(const char* label, int64_t age) = 0;
+};
+
+class SrsEdgeRtmpUpstream : public SrsEdgeUpstream
+{
+private:
+    SrsSimpleRtmpClient* sdk;
+public:
+    SrsEdgeRtmpUpstream();
+    virtual ~SrsEdgeRtmpUpstream();
+public:
+    virtual int connect(SrsRequest* r, SrsLbRoundRobin* lb);
+    virtual int recv_message(SrsCommonMessage** pmsg);
+    virtual int decode_message(SrsCommonMessage* msg, SrsPacket** ppacket);
+    virtual void close();
+    virtual void kbps_sample(const char* label, int64_t age);
+};
+
+/**
 * edge used to ingest stream from origin.
 * edge used to ingest stream from origin.
 */
 */
 class SrsEdgeIngester : public ISrsReusableThread2Handler
 class SrsEdgeIngester : public ISrsReusableThread2Handler
@@ -85,8 +117,8 @@ private:
     SrsPlayEdge* edge;
     SrsPlayEdge* edge;
     SrsRequest* req;
     SrsRequest* req;
     SrsReusableThread2* pthread;
     SrsReusableThread2* pthread;
-    SrsSimpleRtmpClient* sdk;
     SrsLbRoundRobin* lb;
     SrsLbRoundRobin* lb;
+    SrsEdgeUpstream* upstream;
 public:
 public:
     SrsEdgeIngester();
     SrsEdgeIngester();
     virtual ~SrsEdgeIngester();
     virtual ~SrsEdgeIngester();

@@ -2054,7 +2054,9 @@ int SrsSource::on_publish()
     
     
     // whatever, the publish thread is the source or edge source,
     // whatever, the publish thread is the source or edge source,
     // save its id to srouce id.
     // save its id to srouce id.
-    on_source_id_changed(_srs_context->get_id());
+    if ((ret = on_source_id_changed(_srs_context->get_id())) != ERROR_SUCCESS) {
+        return ret;
+    }
     
     
     // reset the mix queue.
     // reset the mix queue.
     mix_queue->clear();
     mix_queue->clear();