for bug #241, merge big chunks for publish, no use.

trunk/src/app/srs_app_recv_thread.cpp

@@ -288,10 +288,16 @@ void SrsPublishRecvThread::on_thread_start()
 {
     // we donot set the auto response to false,
     // for the main thread never send message.
+
+    // notice the protocol stack to merge chunks to big buffer.
+    // for example, the buffer is 64KB=512kb, it's 1s buffer for 500kbps video stream.
+    // so we can use read_fullly(64KB) to merge all chunks in 1s.
+    // @see https://github.com/winlinvip/simple-rtmp-server/issues/241
+    rtmp->set_merge_chunks(true);
 }
 
 void SrsPublishRecvThread::on_thread_stop()
 {
-    // we donot set the auto response to true,
-    // for we donot set to false yet.
+    // revert state
+    rtmp->set_merge_chunks(false);
 }

trunk/src/kernel/srs_kernel_buffer.cpp

@@ -26,7 +26,10 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_kernel_error.hpp>
 #include <srs_kernel_log.hpp>
 
-#define SOCKET_READ_SIZE 4096
+// 4096=4KB
+// 16384=16KB
+// 65536=64KB
+#define SOCKET_READ_SIZE 16384
 
 ISrsBufferReader::ISrsBufferReader()
 {
@@ -38,10 +41,13 @@ ISrsBufferReader::~ISrsBufferReader()
 
 SrsBuffer::SrsBuffer()
 {
+    merge_chunks_in_big_buffer = false;
+    buffer = new char[SOCKET_READ_SIZE];
 }
 
 SrsBuffer::~SrsBuffer()
 {
+    srs_freep(buffer);
 }
 
 int SrsBuffer::length()
@@ -88,12 +94,16 @@ int SrsBuffer::grow(ISrsBufferReader* reader, int required_size)
     }
 
     while (length() < required_size) {
-        char buffer[SOCKET_READ_SIZE];
-        
         ssize_t nread;
+        if (merge_chunks_in_big_buffer) {
+            if ((ret = reader->read_fully(buffer, SOCKET_READ_SIZE, &nread)) != ERROR_SUCCESS) {
+                return ret;
+            }
+        } else {
             if ((ret = reader->read(buffer, SOCKET_READ_SIZE, &nread)) != ERROR_SUCCESS) {
                 return ret;
             }
+        }
         
         srs_assert((int)nread > 0);
         append(buffer, (int)nread);
@@ -102,4 +112,9 @@ int SrsBuffer::grow(ISrsBufferReader* reader, int required_size)
     return ret;
 }
 
+void SrsBuffer::set_merge_chunks(bool v)
+{
+    merge_chunks_in_big_buffer = v;
+}
+
 

trunk/src/kernel/srs_kernel_buffer.hpp

@@ -42,7 +42,16 @@ public:
     virtual ~ISrsBufferReader();
 // for protocol/amf0/msg-codec
 public:
+    /**
+    * read some bytes of data.
+    * @param nread, the actually read size, NULL to ignore.
+    */
     virtual int read(void* buf, size_t size, ssize_t* nread) = 0;
+    /**
+    * read specified size bytes of data
+    * @param nread, the actually read size, NULL to ignore.
+    */
+    virtual int read_fully(void* buf, size_t size, ssize_t* nread) = 0;
 };
 
 /**
@@ -53,6 +62,15 @@ class SrsBuffer
 {
 private:
     std::vector<char> data;
+    /**
+    * notice the protocol stack to merge chunks to big buffer.
+    * for example, the buffer is 64KB=512kb, it's 1s buffer for 500kbps video stream.
+    * so we can use read_fullly(64KB) to merge all chunks in 1s.
+    * @see https://github.com/winlinvip/simple-rtmp-server/issues/241
+    */
+    bool merge_chunks_in_big_buffer;
+    // the socket recv buffer.
+    char* buffer;
 public:
     SrsBuffer();
     virtual ~SrsBuffer();
@@ -89,6 +107,14 @@ public:
     * @remark, we actually maybe read more than required_size, maybe 4k for example.
     */
     virtual int grow(ISrsBufferReader* reader, int required_size);
+public:
+    /**
+    * notice the protocol stack to merge chunks to big buffer.
+    * for example, the buffer is 64KB=512kb, it's 1s buffer for 500kbps video stream.
+    * so we can use read_fullly(64KB) to merge all chunks in 1s.
+    * @see https://github.com/winlinvip/simple-rtmp-server/issues/241
+    */
+    virtual void set_merge_chunks(bool v);
 };
 
 #endif

trunk/src/rtmp/srs_protocol_io.hpp

@@ -43,17 +43,16 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 | IBufferReader |     |    IStatistic      |      | IBufferWriter |
 +---------------+     +--------------------+      +---------------+
 | + read()      |     | + get_recv_bytes() |      | + write()     |
-+------+--------+     | + get_recv_bytes() |      | + writev()    |
-      / \             +---+--------------+-+      +-------+-------+
-       |                 / \            / \              / \
+| + readfully() |     | + get_recv_bytes() |      | + writev()    |
++------+--------+     +---+--------------+-+      +-------+-------+
+      / \                / \            / \              / \
        |                  |              |                |
 +------+------------------+-+      +-----+----------------+--+
 | IProtocolReader           |      | IProtocolWriter         |
 +---------------------------+      +-------------------------+
-| + readfully()             |      | + set_send_timeout()    |
-| + set_recv_timeout()      |      +-------+-----------------+
-+------------+--------------+             / \     
-            / \                            |   
+| + set_recv_timeout()      |      | + set_send_timeout()    |
++------------+--------------+      +-------+-----------------+
+            / \                           / \     
              |                             | 
           +--+-----------------------------+-+
           |       IProtocolReaderWriter      |
@@ -123,13 +122,6 @@ public:
     * get the recv timeout in us.
     */
     virtual int64_t get_recv_timeout() = 0;
-// for handshake.
-public:
-    /**
-    * read specified size bytes of data
-    * @param nread, the actually read size, NULL to ignore.
-    */
-    virtual int read_fully(void* buf, size_t size, ssize_t* nread) = 0;
 };
 
 /**

trunk/src/rtmp/srs_protocol_rtmp.cpp

@@ -745,6 +745,11 @@ void SrsRtmpServer::set_auto_response(bool v)
     protocol->set_auto_response(v);
 }
 
+void SrsRtmpServer::set_merge_chunks(bool v)
+{
+    protocol->set_merge_chunks(v);
+}
+
 void SrsRtmpServer::set_recv_timeout(int64_t timeout_us)
 {
     protocol->set_recv_timeout(timeout_us);

trunk/src/rtmp/srs_protocol_rtmp.hpp

@@ -343,6 +343,13 @@ public:
     */
     virtual void set_auto_response(bool v);
     /**
+    * notice the protocol stack to merge chunks to big buffer.
+    * for example, the buffer is 64KB=512kb, it's 1s buffer for 500kbps video stream.
+    * so we can use read_fullly(64KB) to merge all chunks in 1s.
+    * @see https://github.com/winlinvip/simple-rtmp-server/issues/241
+    */
+    virtual void set_merge_chunks(bool v);
+    /**
     * set/get the recv timeout in us.
     * if timeout, recv/send message return ERROR_SOCKET_TIMEOUT.
     */

trunk/src/rtmp/srs_protocol_stack.cpp

@@ -478,6 +478,11 @@ int SrsProtocol::manual_response_flush()
     return ret;
 }
 
+void SrsProtocol::set_merge_chunks(bool v)
+{
+    in_buffer->set_merge_chunks(v);
+}
+
 void SrsProtocol::set_recv_timeout(int64_t timeout_us)
 {
     return skt->set_recv_timeout(timeout_us);

trunk/src/rtmp/srs_protocol_stack.hpp

@@ -269,6 +269,13 @@ public:
     * @see the auto_response_when_recv and manual_response_queue.
     */
     virtual int manual_response_flush();
+    /**
+    * notice the protocol stack to merge chunks to big buffer.
+    * for example, the buffer is 64KB=512kb, it's 1s buffer for 500kbps video stream.
+    * so we can use read_fullly(64KB) to merge all chunks in 1s.
+    * @see https://github.com/winlinvip/simple-rtmp-server/issues/241
+    */
+    virtual void set_merge_chunks(bool v);
 public:
     /**
     * set/get the recv timeout in us.

trunk/src/utest/srs_utest_kernel.cpp

@@ -199,6 +199,11 @@ int MockBufferReader::read(void* buf, size_t size, ssize_t* nread)
     return ERROR_SUCCESS;
 }
 
+int MockBufferReader::read_fully(void* buf, size_t size, ssize_t* nread)
+{
+    return read(buf, size, nread);
+}
+
 #ifdef ENABLE_UTEST_KERNEL
 
 VOID TEST(KernelBufferTest, DefaultObject)

trunk/src/utest/srs_utest_kernel.hpp

@@ -42,6 +42,7 @@ public:
     virtual ~MockBufferReader();
 public:
     virtual int read(void* buf, size_t size, ssize_t* nread);
+    virtual int read_fully(void* buf, size_t size, ssize_t* nread);
 };
 
 class MockSrsFileWriter : public SrsFileWriter