for bug #194, add pipe to consumer.

@@ -66,6 +66,11 @@ int SrsPipe::initialize()
     return ret;
     return ret;
 }
 }
 
 
+st_netfd_t SrsPipe::rfd()
+{
+    return read_stfd;
+}
+
 bool SrsPipe::already_written()
 bool SrsPipe::already_written()
 {
 {
     return _already_written;
     return _already_written;

@@ -57,6 +57,10 @@ public:
     * initialize pipes, open fds.
     * initialize pipes, open fds.
     */
     */
     virtual int initialize();
     virtual int initialize();
+    /**
+    * get the read fd to poll.
+    */
+    virtual st_netfd_t rfd();
 public:
 public:
     /**
     /**
     * for event based service, whether already writen data.
     * for event based service, whether already writen data.

@@ -41,6 +41,7 @@ using namespace std;
 #include <srs_app_edge.hpp>
 #include <srs_app_edge.hpp>
 #include <srs_kernel_utility.hpp>
 #include <srs_kernel_utility.hpp>
 #include <srs_app_avc_aac.hpp>
 #include <srs_app_avc_aac.hpp>
+#include <srs_app_pipe.hpp>
 
 
 #define CONST_MAX_JITTER_MS         500
 #define CONST_MAX_JITTER_MS         500
 #define DEFAULT_FRAME_TIME_MS         40
 #define DEFAULT_FRAME_TIME_MS         40
@@ -171,6 +172,11 @@ void SrsMessageQueue::set_queue_size(double queue_size)
     queue_size_ms = (int)(queue_size * 1000);
     queue_size_ms = (int)(queue_size * 1000);
 }
 }
 
 
+bool SrsMessageQueue::empty()
+{
+    return msgs.size() == 0;
+}
+
 int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
 int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
@@ -290,6 +296,7 @@ SrsConsumer::SrsConsumer(SrsSource* _source)
     jitter = new SrsRtmpJitter();
     jitter = new SrsRtmpJitter();
     queue = new SrsMessageQueue();
     queue = new SrsMessageQueue();
     should_update_source_id = false;
     should_update_source_id = false;
+    pipe = new SrsPipe();
 }
 }
 
 
 SrsConsumer::~SrsConsumer()
 SrsConsumer::~SrsConsumer()
@@ -299,6 +306,23 @@ SrsConsumer::~SrsConsumer()
     srs_freep(queue);
     srs_freep(queue);
 }
 }
 
 
+int SrsConsumer::initialize()
+{
+    int ret = ERROR_SUCCESS;
+    
+    if ((ret = pipe->initialize()) != ERROR_SUCCESS) {
+        srs_error("initialize the pipe for consumer failed. ret=%d", ret);
+        return ret;
+    }
+    
+    return ret;
+}
+
+st_netfd_t SrsConsumer::pipe_fd()
+{
+    return pipe->rfd();
+}
+
 void SrsConsumer::set_queue_size(double queue_size)
 void SrsConsumer::set_queue_size(double queue_size)
 {
 {
     queue->set_queue_size(queue_size);
     queue->set_queue_size(queue_size);
@@ -329,11 +353,18 @@ int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, bool atc, int tba, int tbv, S
         return ret;
         return ret;
     }
     }
     
     
+    // notify the rtmp connection to resume to send packet.
+    if (!pipe->already_written()) {
+        pipe->active();
+    }
+    
     return ret;
     return ret;
 }
 }
 
 
 int SrsConsumer::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count)
 int SrsConsumer::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count)
 {
 {
+    int ret = ERROR_SUCCESS;
+    
     srs_assert(max_count > 0);
     srs_assert(max_count > 0);
     
     
     if (should_update_source_id) {
     if (should_update_source_id) {
@@ -346,7 +377,15 @@ int SrsConsumer::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& c
         return ERROR_SUCCESS;
         return ERROR_SUCCESS;
     }
     }
     
     
-    return queue->dump_packets(max_count, pmsgs, count);
+    if ((ret = queue->dump_packets(max_count, pmsgs, count)) != ERROR_SUCCESS) {
+        return ret;
+    }
+    
+    if (queue->empty()) {
+        return pipe->reset();
+    }
+    
+    return ret;
 }
 }
 
 
 int SrsConsumer::on_play_client_pause(bool is_pause)
 int SrsConsumer::on_play_client_pause(bool is_pause)
@@ -1454,7 +1493,13 @@ void SrsSource::on_unpublish()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    consumer = new SrsConsumer(this);
+    SrsConsumer* c = new SrsConsumer(this);
+    if ((ret = c->initialize()) != ERROR_SUCCESS) {
+        srs_freep(c);
+        return ret;
+    }
+    
+    consumer = c;
     consumers.push_back(consumer);
     consumers.push_back(consumer);
     
     
     double queue_size = _srs_config->get_queue_length(_req->vhost);
     double queue_size = _srs_config->get_queue_length(_req->vhost);

@@ -58,6 +58,7 @@ class SrsDvr;
 class SrsEncoder;
 class SrsEncoder;
 #endif
 #endif
 class SrsStream;
 class SrsStream;
+class SrsPipe;
 
 
 /**
 /**
 * the time jitter algorithm:
 * the time jitter algorithm:
@@ -122,6 +123,10 @@ public:
     virtual void set_queue_size(double queue_size);
     virtual void set_queue_size(double queue_size);
 public:
 public:
     /**
     /**
+    * whether queue is empty.
+    */
+    virtual bool empty();
+    /**
     * enqueue the message, the timestamp always monotonically.
     * enqueue the message, the timestamp always monotonically.
     * @param msg, the msg to enqueue, user never free it whatever the return code.
     * @param msg, the msg to enqueue, user never free it whatever the return code.
     */
     */
@@ -148,6 +153,7 @@ private:
 class SrsConsumer
 class SrsConsumer
 {
 {
 private:
 private:
+    SrsPipe* pipe;
     SrsRtmpJitter* jitter;
     SrsRtmpJitter* jitter;
     SrsSource* source;
     SrsSource* source;
     SrsMessageQueue* queue;
     SrsMessageQueue* queue;
@@ -159,6 +165,16 @@ public:
     virtual ~SrsConsumer();
     virtual ~SrsConsumer();
 public:
 public:
     /**
     /**
+    * initialize the consumer.
+    */
+    virtual int initialize();
+    /**
+    * source can use this fd to poll with the read event,
+    * for performance issue, @see: https://github.com/winlinvip/simple-rtmp-server/issues/194
+    */
+    virtual st_netfd_t pipe_fd();
+public:
+    /**
     * set the size of queue.
     * set the size of queue.
     */
     */
     virtual void set_queue_size(double queue_size);
     virtual void set_queue_size(double queue_size);