for bug #237, when recv thread failed, quit the cycle. 2.0.44

@@ -80,6 +80,9 @@ int SrsRecvThread::cycle()
 
 
         // we use no timeout to recv, should never got any error.
         // we use no timeout to recv, should never got any error.
         trd->stop_loop();
         trd->stop_loop();
+        
+        // notice the handler got a recv error.
+        handler->on_recv_error(ret);
 
 
         return ret;
         return ret;
     }
     }
@@ -122,6 +125,7 @@ void SrsRecvThread::on_thread_stop()
 SrsQueueRecvThread::SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms)
 SrsQueueRecvThread::SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms)
     : trd(this, rtmp_sdk, timeout_ms)
     : trd(this, rtmp_sdk, timeout_ms)
 {
 {
+    recv_error_code = ERROR_SUCCESS;
 }
 }
 
 
 SrsQueueRecvThread::~SrsQueueRecvThread()
 SrsQueueRecvThread::~SrsQueueRecvThread()
@@ -168,6 +172,11 @@ SrsMessage* SrsQueueRecvThread::pump()
     return msg;
     return msg;
 }
 }
 
 
+int SrsQueueRecvThread::error_code()
+{
+    return recv_error_code;
+}
+
 bool SrsQueueRecvThread::can_handle()
 bool SrsQueueRecvThread::can_handle()
 {
 {
     // we only recv one message and then process it,
     // we only recv one message and then process it,
@@ -186,6 +195,11 @@ int SrsQueueRecvThread::handle(SrsMessage* msg)
     return ERROR_SUCCESS;
     return ERROR_SUCCESS;
 }
 }
 
 
+void SrsQueueRecvThread::on_recv_error(int ret)
+{
+    recv_error_code = ret;
+}
+
 SrsPublishRecvThread::SrsPublishRecvThread(
 SrsPublishRecvThread::SrsPublishRecvThread(
     SrsRtmpServer* rtmp_sdk, int timeout_ms,
     SrsRtmpServer* rtmp_sdk, int timeout_ms,
     SrsRtmpConn* conn, SrsSource* source, bool is_fmle, bool is_edge
     SrsRtmpConn* conn, SrsSource* source, bool is_fmle, bool is_edge
@@ -254,3 +268,8 @@ int SrsPublishRecvThread::handle(SrsMessage* msg)
     // TODO: FIXME: implements it.
     // TODO: FIXME: implements it.
     return ret;
     return ret;
 }
 }
+
+void SrsPublishRecvThread::on_recv_error(int ret)
+{
+    recv_error_code = ret;
+}

@@ -49,16 +49,20 @@ public:
     virtual ~ISrsMessageHandler();
     virtual ~ISrsMessageHandler();
 public:
 public:
     /**
     /**
-     * whether the handler can handle,
-     * for example, when queue recv handler got an message,
-     * it wait the user to process it, then the recv thread
-     * never recv message util the handler is ok.
-     */
+    * whether the handler can handle,
+    * for example, when queue recv handler got an message,
+    * it wait the user to process it, then the recv thread
+    * never recv message util the handler is ok.
+    */
     virtual bool can_handle() = 0;
     virtual bool can_handle() = 0;
     /**
     /**
-     * process the received message.
-     */
+    * process the received message.
+    */
     virtual int handle(SrsMessage* msg) = 0;
     virtual int handle(SrsMessage* msg) = 0;
+    /**
+    * when recv message error.
+    */
+    virtual void on_recv_error(int ret) = 0;
 };
 };
 
 
 /**
 /**
@@ -95,6 +99,8 @@ class SrsQueueRecvThread : public ISrsMessageHandler
 private:
 private:
     std::vector<SrsMessage*> queue;
     std::vector<SrsMessage*> queue;
     SrsRecvThread trd;
     SrsRecvThread trd;
+    // the recv thread error code.
+    int recv_error_code;
 public:
 public:
     SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms);
     SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms);
     virtual ~SrsQueueRecvThread();
     virtual ~SrsQueueRecvThread();
@@ -105,9 +111,11 @@ public:
     virtual bool empty();
     virtual bool empty();
     virtual int size();
     virtual int size();
     virtual SrsMessage* pump();
     virtual SrsMessage* pump();
+    virtual int error_code();
 public:
 public:
     virtual bool can_handle();
     virtual bool can_handle();
     virtual int handle(SrsMessage* msg);
     virtual int handle(SrsMessage* msg);
+    virtual void on_recv_error(int ret);
 };
 };
 
 
 /**
 /**
@@ -139,6 +147,7 @@ public:
 public:
 public:
     virtual bool can_handle();
     virtual bool can_handle();
     virtual int handle(SrsMessage* msg);
     virtual int handle(SrsMessage* msg);
+    virtual void on_recv_error(int ret);
 };
 };
 
 
 #endif
 #endif

@@ -558,13 +558,21 @@ int SrsRtmpConn::do_playing(SrsSource* source, SrsQueueRecvThread* trd)
             srs_verbose("pump client message to process.");
             srs_verbose("pump client message to process.");
             
             
             if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
             if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
-                if (!srs_is_system_control_error(ret)) {
+                if (!srs_is_system_control_error(ret) && !srs_is_client_gracefully_close(ret)) {
                     srs_error("process play control message failed. ret=%d", ret);
                     srs_error("process play control message failed. ret=%d", ret);
                 }
                 }
                 return ret;
                 return ret;
             }
             }
         }
         }
         
         
+        // quit when recv thread error.
+        if ((ret = trd->error_code()) != ERROR_SUCCESS) {
+            if (!srs_is_client_gracefully_close(ret)) {
+                srs_error("recv thread failed. ret=%d", ret);
+            }
+            return ret;
+        }
+        
         // collect elapse for pithy print.
         // collect elapse for pithy print.
         pithy_print.elapse();
         pithy_print.elapse();
         
         
@@ -744,6 +752,9 @@ int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd)
 
 
             // check the thread error code.
             // check the thread error code.
             if ((ret = trd->error_code()) != ERROR_SUCCESS) {
             if ((ret = trd->error_code()) != ERROR_SUCCESS) {
+                if (!srs_is_client_gracefully_close(ret)) {
+                    srs_error("recv thread failed. ret=%d", ret);
+                }
                 return ret;
                 return ret;
             }
             }
         }
         }

@@ -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       2
 #define VERSION_MAJOR       2
 #define VERSION_MINOR       0
 #define VERSION_MINOR       0
-#define VERSION_REVISION    43
+#define VERSION_REVISION    44
 // server info.
 // server info.
 #define RTMP_SIG_SRS_KEY "SRS"
 #define RTMP_SIG_SRS_KEY "SRS"
 #define RTMP_SIG_SRS_ROLE "origin/edge server"
 #define RTMP_SIG_SRS_ROLE "origin/edge server"