support basic edge(play/publish) RTMP server. 0.9.78

@@ -192,6 +192,7 @@ Supported operating systems and hardware:
 * 2013-10-17, Created.<br/>
 * 2013-10-17, Created.<br/>
 
 
 ## History
 ## History
+* v1.0, 2014-04-27, support basic edge(play/publish) RTMP server. 0.9.78
 * v1.0, 2014-04-25, add donation page. 0.9.76
 * v1.0, 2014-04-25, add donation page. 0.9.76
 * v1.0, 2014-04-24, support live flashP2P(integrated by chnvideo VDN). 0.9.75
 * v1.0, 2014-04-24, support live flashP2P(integrated by chnvideo VDN). 0.9.75
 * v1.0, 2014-04-21, support android app to start srs for internal edge. 0.9.72
 * v1.0, 2014-04-21, support android app to start srs for internal edge. 0.9.72

@@ -49,6 +49,12 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // when edge timeout, retry next.
 // when edge timeout, retry next.
 #define SRS_EDGE_INGESTER_TIMEOUT_US (int64_t)(3*1000*1000LL)
 #define SRS_EDGE_INGESTER_TIMEOUT_US (int64_t)(3*1000*1000LL)
 
 
+// when error, edge ingester sleep for a while and retry.
+#define SRS_EDGE_FORWARDER_SLEEP_US (int64_t)(1*1000*1000LL)
+
+// when edge timeout, retry next.
+#define SRS_EDGE_FORWARDER_TIMEOUT_US (int64_t)(3*1000*1000LL)
+
 SrsEdgeIngester::SrsEdgeIngester()
 SrsEdgeIngester::SrsEdgeIngester()
 {
 {
     io = NULL;
     io = NULL;
@@ -316,6 +322,7 @@ SrsEdgeForwarder::SrsEdgeForwarder()
     origin_index = 0;
     origin_index = 0;
     stream_id = 0;
     stream_id = 0;
     stfd = NULL;
     stfd = NULL;
+    pthread = new SrsThread(this, SRS_EDGE_FORWARDER_SLEEP_US);
 }
 }
 
 
 SrsEdgeForwarder::~SrsEdgeForwarder()
 SrsEdgeForwarder::~SrsEdgeForwarder()
@@ -367,17 +374,54 @@ int SrsEdgeForwarder::start()
         return ret;
         return ret;
     }
     }
     
     
-    return ret;
+    return pthread->start();
 }
 }
 
 
 void SrsEdgeForwarder::stop()
 void SrsEdgeForwarder::stop()
 {
 {
+    pthread->stop();
+    
     close_underlayer_socket();
     close_underlayer_socket();
     
     
     srs_freep(client);
     srs_freep(client);
     srs_freep(io);
     srs_freep(io);
 }
 }
 
 
+int SrsEdgeForwarder::cycle()
+{
+    int ret = ERROR_SUCCESS;
+    
+    client->set_recv_timeout(SRS_EDGE_FORWARDER_TIMEOUT_US);
+    
+    SrsPithyPrint pithy_print(SRS_STAGE_EDGE);
+
+    while (pthread->can_loop()) {
+        // switch to other st-threads.
+        st_usleep(0);
+        
+        pithy_print.elapse();
+        
+        // pithy print
+        if (pithy_print.can_print()) {
+            srs_trace("-> time=%"PRId64", obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
+                pithy_print.age(), client->get_send_bytes(), client->get_recv_bytes(), client->get_send_kbps(), client->get_recv_kbps());
+        }
+
+        // read from client.
+        SrsCommonMessage* msg = NULL;
+        if ((ret = client->recv_message(&msg)) != ERROR_SUCCESS) {
+            srs_info("ignore forwarder recv origin server message failed. ret=%d", ret);
+            continue;
+        }
+        srs_verbose("edge loop recv message. ret=%d", ret);
+        
+        srs_assert(msg);
+        SrsAutoFree(SrsCommonMessage, msg, false);
+    }
+    
+    return ret;
+}
+
 int SrsEdgeForwarder::proxy(SrsCommonMessage* msg)
 int SrsEdgeForwarder::proxy(SrsCommonMessage* msg)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;

@@ -104,7 +104,7 @@ private:
 /**
 /**
 * edge used to forward stream to origin.
 * edge used to forward stream to origin.
 */
 */
-class SrsEdgeForwarder
+class SrsEdgeForwarder : public ISrsThreadHandler
 {
 {
 private:
 private:
     int stream_id;
     int stream_id;
@@ -112,6 +112,7 @@ private:
     SrsSource* _source;
     SrsSource* _source;
     SrsPublishEdge* _edge;
     SrsPublishEdge* _edge;
     SrsRequest* _req;
     SrsRequest* _req;
+    SrsThread* pthread;
     st_netfd_t stfd;
     st_netfd_t stfd;
     ISrsProtocolReaderWriter* io;
     ISrsProtocolReaderWriter* io;
     SrsRtmpClient* client;
     SrsRtmpClient* client;
@@ -123,6 +124,9 @@ public:
     virtual int initialize(SrsSource* source, SrsPublishEdge* edge, SrsRequest* req);
     virtual int initialize(SrsSource* source, SrsPublishEdge* edge, SrsRequest* req);
     virtual int start();
     virtual int start();
     virtual void stop();
     virtual void stop();
+// interface ISrsThreadHandler
+public:
+    virtual int cycle();
 public:
 public:
     virtual int proxy(SrsCommonMessage* msg);
     virtual int proxy(SrsCommonMessage* msg);
 private:
 private:
@@ -134,6 +138,7 @@ private:
 * play edge control service.
 * play edge control service.
 * downloading edge speed-up.
 * downloading edge speed-up.
 */
 */
+// TODO: FIXME: support reload
 class SrsPlayEdge
 class SrsPlayEdge
 {
 {
 private:
 private:
@@ -164,6 +169,7 @@ public:
 * publish edge control service.
 * publish edge control service.
 * uploading edge speed-up.
 * uploading edge speed-up.
 */
 */
+// TODO: FIXME: support reload
 class SrsPublishEdge
 class SrsPublishEdge
 {
 {
 private:
 private:

@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // current release version
 // current release version
 #define VERSION_MAJOR "0"
 #define VERSION_MAJOR "0"
 #define VERSION_MINOR "9"
 #define VERSION_MINOR "9"
-#define VERSION_REVISION "77"
+#define VERSION_REVISION "78"
 #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION
 #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION
 // server info.
 // server info.
 #define RTMP_SIG_SRS_KEY "srs"
 #define RTMP_SIG_SRS_KEY "srs"