for bug #237, extract a queue recv thread.

@@ -26,95 +26,69 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_protocol_rtmp.hpp>
 #include <srs_protocol_rtmp.hpp>
 #include <srs_protocol_stack.hpp>
 #include <srs_protocol_stack.hpp>
 
 
-SrsQueueRecvThread::SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk)
+ISrsMessageHandler::ISrsMessageHandler()
 {
 {
-    rtmp = rtmp_sdk;
-    trd = new SrsThread(this, 0, true);
 }
 }
 
 
-SrsQueueRecvThread::~SrsQueueRecvThread()
+ISrsMessageHandler::~ISrsMessageHandler()
 {
 {
-    // stop recv thread.
-    stop();
-    
-    // destroy the thread.
-    srs_freep(trd);
-    
-    // clear all messages.
-    std::vector<SrsMessage*>::iterator it;
-    for (it = queue.begin(); it != queue.end(); ++it) {
-        SrsMessage* msg = *it;
-        srs_freep(msg);
-    }
-    queue.clear();
 }
 }
 
 
-bool SrsQueueRecvThread::empty()
+SrsRecvThread::SrsRecvThread(ISrsMessageHandler* msg_handler, SrsRtmpServer* rtmp_sdk)
 {
 {
-    return queue.empty();
+    handler = msg_handler;
+    rtmp = rtmp_sdk;
+    trd = new SrsThread(this, 0, true);
 }
 }
 
 
-int SrsQueueRecvThread::size()
+SrsRecvThread::~SrsRecvThread()
 {
 {
-    return (int)queue.size();
-}
+    // stop recv thread.
+    stop();
 
 
-SrsMessage* SrsQueueRecvThread::pump()
-{
-    srs_assert(!queue.empty());
-    
-    SrsMessage* msg = *queue.begin();
-    
-    queue.erase(queue.begin());
-    
-    return msg;
+    // destroy the thread.
+    srs_freep(trd);
 }
 }
 
 
-int SrsQueueRecvThread::start()
+int SrsRecvThread::start()
 {
 {
     return trd->start();
     return trd->start();
 }
 }
 
 
-void SrsQueueRecvThread::stop()
+void SrsRecvThread::stop()
 {
 {
     trd->stop();
     trd->stop();
 }
 }
 
 
-int SrsQueueRecvThread::cycle()
+int SrsRecvThread::cycle()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
-    
-    // we only recv one message and then process it,
-    // for the message may cause the thread to stop,
-    // when stop, the thread is freed, so the messages
-    // are dropped.
-    if (!queue.empty()) {
+
+    if (!handler->can_handle()) {
         st_usleep(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US);
         st_usleep(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US);
         return ret;
         return ret;
     }
     }
-    
+
     SrsMessage* msg = NULL;
     SrsMessage* msg = NULL;
-    
+
     if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
     if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
         if (!srs_is_client_gracefully_close(ret)) {
         if (!srs_is_client_gracefully_close(ret)) {
             srs_error("recv client control message failed. ret=%d", ret);
             srs_error("recv client control message failed. ret=%d", ret);
         }
         }
-        
+
         // 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();
-        
+
         return ret;
         return ret;
     }
     }
     srs_verbose("play loop recv message. ret=%d", ret);
     srs_verbose("play loop recv message. ret=%d", ret);
-    
-    // put into queue, the send thread will get and process it,
-    // @see SrsRtmpConn::process_play_control_msg
-    queue.push_back(msg);
-    
+
+    handler->handle(msg);
+
     return ret;
     return ret;
 }
 }
 
 
-void SrsQueueRecvThread::on_thread_start()
+void SrsRecvThread::on_thread_start()
 {
 {
     // the multiple messages writev improve performance large,
     // the multiple messages writev improve performance large,
     // but the timeout recv will cause 33% sys call performance,
     // but the timeout recv will cause 33% sys call performance,
@@ -122,19 +96,75 @@ void SrsQueueRecvThread::on_thread_start()
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/194
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/194
     // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
     // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
     rtmp->set_recv_timeout(ST_UTIME_NO_TIMEOUT);
     rtmp->set_recv_timeout(ST_UTIME_NO_TIMEOUT);
-    
-    // disable the protocol auto response, 
+
+    // disable the protocol auto response,
     // for the isolate recv thread should never send any messages.
     // for the isolate recv thread should never send any messages.
     rtmp->set_auto_response(false);
     rtmp->set_auto_response(false);
 }
 }
 
 
-void SrsQueueRecvThread::on_thread_stop()
+void SrsRecvThread::on_thread_stop()
 {
 {
     // enable the protocol auto response,
     // enable the protocol auto response,
     // for the isolate recv thread terminated.
     // for the isolate recv thread terminated.
     rtmp->set_auto_response(true);
     rtmp->set_auto_response(true);
-    
+
     // reset the timeout to pulse mode.
     // reset the timeout to pulse mode.
     rtmp->set_recv_timeout(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US);
     rtmp->set_recv_timeout(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US);
 }
 }
 
 
+SrsQueueRecvThread::SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk)
+    : SrsRecvThread(this, rtmp_sdk)
+{
+}
+
+SrsQueueRecvThread::~SrsQueueRecvThread()
+{
+    stop();
+
+    // clear all messages.
+    std::vector<SrsMessage*>::iterator it;
+    for (it = queue.begin(); it != queue.end(); ++it) {
+        SrsMessage* msg = *it;
+        srs_freep(msg);
+    }
+    queue.clear();
+}
+
+bool SrsQueueRecvThread::empty()
+{
+    return queue.empty();
+}
+
+int SrsQueueRecvThread::size()
+{
+    return (int)queue.size();
+}
+
+SrsMessage* SrsQueueRecvThread::pump()
+{
+    srs_assert(!queue.empty());
+    
+    SrsMessage* msg = *queue.begin();
+    
+    queue.erase(queue.begin());
+    
+    return msg;
+}
+
+bool SrsQueueRecvThread::can_handle()
+{
+    // we only recv one message and then process it,
+    // for the message may cause the thread to stop,
+    // when stop, the thread is freed, so the messages
+    // are dropped.
+    return empty();
+}
+
+int SrsQueueRecvThread::handle(SrsMessage* msg)
+{
+    // put into queue, the send thread will get and process it,
+    // @see SrsRtmpConn::process_play_control_msg
+    queue.push_back(msg);
+
+    return ERROR_SUCCESS;
+}

@@ -38,16 +38,57 @@ class SrsRtmpServer;
 class SrsMessage;
 class SrsMessage;
 
 
 /**
 /**
+ * for the recv thread to handle the message.
+ */
+class ISrsMessageHandler
+{
+public:
+    ISrsMessageHandler();
+    virtual ~ISrsMessageHandler();
+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.
+     */
+    virtual bool can_handle() = 0;
+    /**
+     * process the received message.
+     */
+    virtual int handle(SrsMessage* msg) = 0;
+};
+
+/**
+ * the recv thread, use message handler to handle each received message.
+ */
+class SrsRecvThread : public ISrsThreadHandler
+{
+protected:
+    SrsThread* trd;
+    ISrsMessageHandler* handler;
+    SrsRtmpServer* rtmp;
+public:
+    SrsRecvThread(ISrsMessageHandler* msg_handler, SrsRtmpServer* rtmp_sdk);
+    virtual ~SrsRecvThread();
+public:
+    virtual int start();
+    virtual void stop();
+    virtual int cycle();
+public:
+    virtual void on_thread_start();
+    virtual void on_thread_stop();
+};
+
+/**
 * the recv thread used to replace the timeout recv,
 * the recv thread used to replace the timeout recv,
 * which hurt performance for the epoll_ctrl is frequently used.
 * which hurt performance for the epoll_ctrl is frequently used.
 * @see: SrsRtmpConn::playing
 * @see: SrsRtmpConn::playing
 * @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
 * @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
 */
 */
-class SrsQueueRecvThread : public ISrsThreadHandler
+class SrsQueueRecvThread : virtual public ISrsMessageHandler, virtual public SrsRecvThread
 {
 {
 private:
 private:
-    SrsThread* trd;
-    SrsRtmpServer* rtmp;
     std::vector<SrsMessage*> queue;
     std::vector<SrsMessage*> queue;
 public:
 public:
     SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk);
     SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk);
@@ -57,12 +98,8 @@ public:
     virtual int size();
     virtual int size();
     virtual SrsMessage* pump();
     virtual SrsMessage* pump();
 public:
 public:
-    virtual int start();
-    virtual void stop();
-    virtual int cycle();
-public:
-    virtual void on_thread_start();
-    virtual void on_thread_stop();
+    virtual bool can_handle();
+    virtual int handle(SrsMessage* msg);
 };
 };
 
 
 #endif
 #endif