Revert "for bug #194, add pipe to consumer."

This reverts commit 1e601a6e.

@@ -66,11 +66,6 @@ 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,10 +57,6 @@ 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,7 +41,6 @@ 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
@@ -172,11 +171,6 @@ 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;
@@ -296,7 +290,6 @@ 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()
@@ -306,23 +299,6 @@ 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);
@@ -353,18 +329,11 @@ 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) {
@@ -377,15 +346,7 @@ int SrsConsumer::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& c
         return ERROR_SUCCESS;
         return ERROR_SUCCESS;
     }
     }
     
     
-    if ((ret = queue->dump_packets(max_count, pmsgs, count)) != ERROR_SUCCESS) {
-        return ret;
-    }
-    
-    if (queue->empty()) {
-        return pipe->reset();
-    }
-    
-    return ret;
+    return queue->dump_packets(max_count, pmsgs, count);
 }
 }
 
 
 int SrsConsumer::on_play_client_pause(bool is_pause)
 int SrsConsumer::on_play_client_pause(bool is_pause)
@@ -1493,13 +1454,7 @@ void SrsSource::on_unpublish()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    SrsConsumer* c = new SrsConsumer(this);
-    if ((ret = c->initialize()) != ERROR_SUCCESS) {
-        srs_freep(c);
-        return ret;
-    }
-    
-    consumer = c;
+    consumer = new SrsConsumer(this);
     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,7 +58,6 @@ class SrsDvr;
 class SrsEncoder;
 class SrsEncoder;
 #endif
 #endif
 class SrsStream;
 class SrsStream;
-class SrsPipe;
 
 
 /**
 /**
 * the time jitter algorithm:
 * the time jitter algorithm:
@@ -123,10 +122,6 @@ 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.
     */
     */
@@ -153,7 +148,6 @@ private:
 class SrsConsumer
 class SrsConsumer
 {
 {
 private:
 private:
-    SrsPipe* pipe;
     SrsRtmpJitter* jitter;
     SrsRtmpJitter* jitter;
     SrsSource* source;
     SrsSource* source;
     SrsMessageQueue* queue;
     SrsMessageQueue* queue;
@@ -165,16 +159,6 @@ 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);