refine protocol, add comments. add utest for protocol bytes and timeout.

@@ -61,48 +61,47 @@ public:
     std::string pageUrl;
     std::string pageUrl;
     std::string swfUrl;
     std::string swfUrl;
     double objectEncoding;
     double objectEncoding;
-    
+// data discovery from request.
+public:
+    // discovery from tcUrl and play/publish.
     std::string schema;
     std::string schema;
     std::string vhost;
     std::string vhost;
     std::string host;
     std::string host;
     std::string port;
     std::string port;
     std::string app;
     std::string app;
     std::string stream;
     std::string stream;
-    
     // for play live stream, 
     // for play live stream, 
     // used to specified the stop when exceed the duration.
     // used to specified the stop when exceed the duration.
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/45
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/45
     // in ms.
     // in ms.
     double duration;
     double duration;
-    
     // the token in the connect request,
     // the token in the connect request,
     // used for edge traverse to origin authentication,
     // used for edge traverse to origin authentication,
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/104
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/104
     SrsAmf0Object* args;
     SrsAmf0Object* args;
-    
+public:
     SrsRequest();
     SrsRequest();
     virtual ~SrsRequest();
     virtual ~SrsRequest();
-
+public:
     /**
     /**
     * deep copy the request, for source to use it to support reload,
     * deep copy the request, for source to use it to support reload,
     * for when initialize the source, the request is valid,
     * for when initialize the source, the request is valid,
     * when reload it, the request maybe invalid, so need to copy it.
     * when reload it, the request maybe invalid, so need to copy it.
     */
     */
     virtual SrsRequest* copy();
     virtual SrsRequest* copy();
-
     /**
     /**
     * update the auth info of request,
     * update the auth info of request,
     * to keep the current request ptr is ok,
     * to keep the current request ptr is ok,
     * for many components use the ptr of request.
     * for many components use the ptr of request.
     */
     */
     virtual void update_auth(SrsRequest* req);
     virtual void update_auth(SrsRequest* req);
-    
     /**
     /**
     * get the stream identify, vhost/app/stream.
     * get the stream identify, vhost/app/stream.
     */
     */
     virtual std::string get_stream_url();
     virtual std::string get_stream_url();
-    
-    // strip url, user must strip when update the url.
+    /**
+    * strip url, user must strip when update the url.
+    */
     virtual void strip();
     virtual void strip();
 };
 };
 
 
@@ -112,8 +111,11 @@ public:
 class SrsResponse
 class SrsResponse
 {
 {
 public:
 public:
+    /**
+    * the stream id to response client createStream.
+    */
     int stream_id;
     int stream_id;
-    
+public:
     SrsResponse();
     SrsResponse();
     virtual ~SrsResponse();
     virtual ~SrsResponse();
 };
 };

@@ -29,6 +29,8 @@ using namespace std;
 #include <srs_protocol_utility.hpp>
 #include <srs_protocol_utility.hpp>
 #include <srs_protocol_msg_array.hpp>
 #include <srs_protocol_msg_array.hpp>
 #include <srs_protocol_rtmp_stack.hpp>
 #include <srs_protocol_rtmp_stack.hpp>
+#include <srs_kernel_utility.hpp>
+#include <srs_app_st.hpp>
 
 
 MockEmptyIO::MockEmptyIO()
 MockEmptyIO::MockEmptyIO()
 {
 {
@@ -91,6 +93,114 @@ int MockEmptyIO::read(void* /*buf*/, size_t /*size*/, ssize_t* /*nread*/)
     return ERROR_SUCCESS;
     return ERROR_SUCCESS;
 }
 }
 
 
+MockBufferIO::MockBufferIO()
+{
+    recv_timeout = send_timeout = ST_UTIME_NO_TIMEOUT;
+    recv_bytes = send_bytes = 0;
+}
+
+MockBufferIO::~MockBufferIO()
+{
+}
+
+bool MockBufferIO::is_never_timeout(int64_t timeout_us)
+{
+    return (int64_t)ST_UTIME_NO_TIMEOUT == timeout_us;
+}
+
+int MockBufferIO::read_fully(void* buf, size_t size, ssize_t* nread)
+{
+    if (in_buffer.length() < (int)size) {
+        return ERROR_SOCKET_READ;
+    }
+    memcpy(buf, in_buffer.bytes(), size);
+    
+    recv_bytes += size;
+    if (nread) {
+        *nread = size;
+    }
+    in_buffer.erase(size);
+    return ERROR_SUCCESS;
+}
+
+int MockBufferIO::write(void* buf, size_t size, ssize_t* nwrite)
+{
+    send_bytes += size;
+    if (nwrite) {
+        *nwrite = size;
+    }
+    out_buffer.append((char*)buf, size);
+    return ERROR_SUCCESS;
+}
+
+void MockBufferIO::set_recv_timeout(int64_t timeout_us)
+{
+    recv_timeout = timeout_us;
+}
+
+int64_t MockBufferIO::get_recv_timeout()
+{
+    return recv_timeout;
+}
+
+int64_t MockBufferIO::get_recv_bytes()
+{
+    return recv_bytes;
+}
+
+void MockBufferIO::set_send_timeout(int64_t timeout_us)
+{
+    send_timeout = timeout_us;
+}
+
+int64_t MockBufferIO::get_send_timeout()
+{
+    return send_timeout;
+}
+
+int64_t MockBufferIO::get_send_bytes()
+{
+    return send_bytes;
+}
+
+int MockBufferIO::writev(const iovec *iov, int iov_size, ssize_t* nwrite)
+{
+    int ret = ERROR_SUCCESS;
+    
+    ssize_t total = 0;
+    for (int i = 0; i <iov_size; i++) {
+        const iovec& pi = iov[i];
+        
+        ssize_t writen = 0;
+        if ((ret = write(pi.iov_base, pi.iov_len, &writen)) != ERROR_SUCCESS) {
+            return ret;
+        }
+        total += writen;
+    }
+    
+    if (nwrite) {
+        *nwrite = total;
+    }
+    return ret;
+}
+
+int MockBufferIO::read(void* buf, size_t size, ssize_t* nread)
+{
+    if (in_buffer.length() <= 0) {
+        return ERROR_SOCKET_READ;
+    }
+    
+    size_t available = srs_min(in_buffer.length(), (int)size);
+    memcpy(buf, in_buffer.bytes(), available);
+    
+    recv_bytes += available;
+    if (nread) {
+        *nread = available;
+    }
+    in_buffer.erase(available);
+    return ERROR_SUCCESS;
+}
+
 #ifdef SRS_AUTO_SSL
 #ifdef SRS_AUTO_SSL
 
 
 // verify the sha256
 // verify the sha256
@@ -426,3 +536,31 @@ VOID TEST(ProtocolMsgArrayTest, MessageArray)
     EXPECT_EQ(0, msg.count());
     EXPECT_EQ(0, msg.count());
 }
 }
 
 
+VOID TEST(ProtocolStackTest, ProtocolTimeout)
+{
+    MockBufferIO bio;
+    SrsProtocol proto(&bio);
+    
+    EXPECT_TRUE((int64_t)ST_UTIME_NO_TIMEOUT == proto.get_recv_timeout());
+    EXPECT_TRUE((int64_t)ST_UTIME_NO_TIMEOUT == proto.get_send_timeout());
+    
+    proto.set_recv_timeout(10);
+    EXPECT_TRUE(10 == proto.get_recv_timeout());
+    
+    proto.set_send_timeout(10);
+    EXPECT_TRUE(10 == proto.get_send_timeout());
+}
+
+VOID TEST(ProtocolStackTest, ProtocolBytes)
+{
+    MockBufferIO bio;
+    SrsProtocol proto(&bio);
+    
+    EXPECT_TRUE(0 == proto.get_recv_bytes());
+    EXPECT_TRUE(0 == proto.get_send_bytes());
+    
+    SrsConnectAppPacket* pkt = new SrsConnectAppPacket();
+    proto.send_and_free_packet(pkt, 0);
+    EXPECT_TRUE(0 < proto.get_send_bytes());
+}
+

@@ -34,6 +34,7 @@ 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_handshake.hpp>
 #include <srs_protocol_handshake.hpp>
+#include <srs_kernel_buffer.hpp>
 
 
 #ifdef SRS_AUTO_SSL
 #ifdef SRS_AUTO_SSL
 using namespace _srs_internal;
 using namespace _srs_internal;
@@ -69,4 +70,41 @@ public:
     virtual int read(void* buf, size_t size, ssize_t* nread);
     virtual int read(void* buf, size_t size, ssize_t* nread);
 };
 };
 
 
+class MockBufferIO : public ISrsProtocolReaderWriter
+{
+public:
+    int64_t recv_timeout;
+    int64_t send_timeout;
+    int64_t recv_bytes;
+    int64_t send_bytes;
+    // data source for socket read.
+    SrsBuffer in_buffer;
+    // data buffer for socket send.
+    SrsBuffer out_buffer;
+public:
+    MockBufferIO();
+    virtual ~MockBufferIO();
+// for protocol
+public:
+    virtual bool is_never_timeout(int64_t timeout_us);
+// for handshake.
+public:
+    virtual int read_fully(void* buf, size_t size, ssize_t* nread);
+    virtual int write(void* buf, size_t size, ssize_t* nwrite);
+// for protocol
+public:
+    virtual void set_recv_timeout(int64_t timeout_us);
+    virtual int64_t get_recv_timeout();
+    virtual int64_t get_recv_bytes();
+// for protocol
+public:
+    virtual void set_send_timeout(int64_t timeout_us);
+    virtual int64_t get_send_timeout();
+    virtual int64_t get_send_bytes();
+    virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite);
+// for protocol/amf0/msg-codec
+public:
+    virtual int read(void* buf, size_t size, ssize_t* nread);
+};
+
 #endif
 #endif