rewrite #438

@@ -121,6 +121,7 @@ void SrsEdgeIngester::stop()
 int SrsEdgeIngester::cycle()
 int SrsEdgeIngester::cycle()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
+
     _source->on_source_id_changed(_srs_context->get_id());
     _source->on_source_id_changed(_srs_context->get_id());
         
         
     std::string ep_server, ep_port;
     std::string ep_server, ep_port;
@@ -384,8 +385,8 @@ int SrsEdgeIngester::connect_server(string& ep_server, string& ep_port)
     
     
     kbps->set_io(io, io);
     kbps->set_io(io, io);
     
     
-    srs_trace("edge pull connected, can_publish=%d, url=%s/%s, server=%s:%d",
-        _source->can_publish(), _req->tcUrl.c_str(), _req->stream.c_str(), server.c_str(), port);
+    srs_trace("edge pull connected, url=%s/%s, server=%s:%d",
+        _req->tcUrl.c_str(), _req->stream.c_str(), server.c_str(), port);
     
     
     return ret;
     return ret;
 }
 }

@@ -424,13 +424,6 @@ int SrsRtmpConn::stream_service_cycle()
     }
     }
     srs_assert(source != NULL);
     srs_assert(source != NULL);
     
     
-    // check ASAP, to fail it faster if invalid.
-    if (type != SrsRtmpConnPlay) {
-        if ((ret = prepare_publish(source, vhost_is_edge)) != ERROR_SUCCESS) {
-            return ret;
-        }
-    }
-
     // update the statistic when source disconveried.
     // update the statistic when source disconveried.
     SrsStatistic* stat = SrsStatistic::instance();
     SrsStatistic* stat = SrsStatistic::instance();
     if ((ret = stat->on_client(_srs_context->get_id(), req)) != ERROR_SUCCESS) {
     if ((ret = stat->on_client(_srs_context->get_id(), req)) != ERROR_SUCCESS) {
@@ -438,6 +431,19 @@ int SrsRtmpConn::stream_service_cycle()
         return ret;
         return ret;
     }
     }
 
 
+    // check ASAP, to fail it faster if invalid.
+    if (type != SrsRtmpConnPlay) {
+        // check publish available
+        if (!source->can_publish(vhost_is_edge)) {
+            ret = ERROR_SYSTEM_STREAM_BUSY;
+            srs_warn("stream %s is already publishing. ret=%d", 
+                req->get_stream_url().c_str(), ret);
+            // to delay request
+            st_usleep(SRS_STREAM_BUSY_SLEEP_US);
+            return ret;
+        }
+    }
+    
     bool enabled_cache = _srs_config->get_gop_cache(req->vhost);
     bool enabled_cache = _srs_config->get_gop_cache(req->vhost);
     srs_trace("source url=%s, ip=%s, cache=%d, is_edge=%d, source_id=%d[%d]",
     srs_trace("source url=%s, ip=%s, cache=%d, is_edge=%d, source_id=%d[%d]",
         req->get_stream_url().c_str(), ip.c_str(), enabled_cache, vhost_is_edge, 
         req->get_stream_url().c_str(), ip.c_str(), enabled_cache, vhost_is_edge, 
@@ -1253,31 +1259,6 @@ int SrsRtmpConn::do_token_traverse_auth(SrsRtmpClient* client)
     return ret;
     return ret;
 }
 }
 
 
-int SrsRtmpConn::prepare_publish(SrsSource* source, bool vhost_is_edge)
-{
-    int ret = ERROR_SUCCESS;
-    srs_assert(source);
-
-    // check publish available
-    bool can_publish = false;
-    if (vhost_is_edge) {
-        can_publish = source->proxy_can_publish();
-    } else {
-        can_publish = source->can_publish();
-    }
-    
-    if (!can_publish) {
-        ret = ERROR_SYSTEM_STREAM_BUSY;
-        srs_warn("stream %s is already publishing. ret=%d", 
-            req->get_stream_url().c_str(), ret);
-        // to delay request
-        st_usleep(SRS_STREAM_BUSY_SLEEP_US);
-        return ret;
-    }
-
-    return ret;
-}
-
 int SrsRtmpConn::http_hooks_on_connect()
 int SrsRtmpConn::http_hooks_on_connect()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;

@@ -123,7 +123,6 @@ private:
     virtual int check_edge_token_traverse_auth();
     virtual int check_edge_token_traverse_auth();
     virtual int connect_server(int origin_index, st_netfd_t* pstsock);
     virtual int connect_server(int origin_index, st_netfd_t* pstsock);
     virtual int do_token_traverse_auth(SrsRtmpClient* client);
     virtual int do_token_traverse_auth(SrsRtmpClient* client);
-    virtual int prepare_publish(SrsSource* source, bool vhost_is_edge);
 private:
 private:
     virtual int http_hooks_on_connect();
     virtual int http_hooks_on_connect();
     virtual void http_hooks_on_close();
     virtual void http_hooks_on_close();

@@ -1350,14 +1350,13 @@ int SrsSource::source_id()
     return _source_id;
     return _source_id;
 }
 }
 
 
-bool SrsSource::can_publish()
-{
-    return _can_publish;
-}
-
-bool SrsSource::proxy_can_publish()
+bool SrsSource::can_publish(bool is_edge)
 {
 {
+    if (is_edge) {
         return publish_edge->can_publish();
         return publish_edge->can_publish();
+    }
+
+    return _can_publish;
 }
 }
 
 
 int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata)
 int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata)

@@ -537,8 +537,7 @@ public:
     virtual int source_id();
     virtual int source_id();
 // logic data methods
 // logic data methods
 public:
 public:
-    virtual bool can_publish();
-    virtual bool proxy_can_publish();
+    virtual bool can_publish(bool is_edge);
     virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
     virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
 public:
 public:
     virtual int on_audio(SrsCommonMessage* audio);
     virtual int on_audio(SrsCommonMessage* audio);