winlin

refine RTMP protocol stack, rename buffer to in_buffer, change field pp/stream t…

…o local variables. 0.9.154
@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. @@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
31 // current release version 31 // current release version
32 #define VERSION_MAJOR "0" 32 #define VERSION_MAJOR "0"
33 #define VERSION_MINOR "9" 33 #define VERSION_MINOR "9"
34 -#define VERSION_REVISION "153" 34 +#define VERSION_REVISION "154"
35 #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION 35 #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION
36 // server info. 36 // server info.
37 #define RTMP_SIG_SRS_KEY "SRS" 37 #define RTMP_SIG_SRS_KEY "SRS"
@@ -297,8 +297,7 @@ SrsProtocol::AckWindowSize::AckWindowSize() @@ -297,8 +297,7 @@ SrsProtocol::AckWindowSize::AckWindowSize()
297 297
298 SrsProtocol::SrsProtocol(ISrsProtocolReaderWriter* io) 298 SrsProtocol::SrsProtocol(ISrsProtocolReaderWriter* io)
299 { 299 {
300 - buffer = new SrsBuffer();  
301 - decode_stream = new SrsStream(); 300 + in_buffer = new SrsBuffer();
302 skt = io; 301 skt = io;
303 302
304 in_chunk_size = out_chunk_size = RTMP_DEFAULT_CHUNK_SIZE; 303 in_chunk_size = out_chunk_size = RTMP_DEFAULT_CHUNK_SIZE;
@@ -317,8 +316,7 @@ SrsProtocol::~SrsProtocol() @@ -317,8 +316,7 @@ SrsProtocol::~SrsProtocol()
317 chunk_streams.clear(); 316 chunk_streams.clear();
318 } 317 }
319 318
320 - srs_freep(decode_stream);  
321 - srs_freep(buffer); 319 + srs_freep(in_buffer);
322 } 320 }
323 321
324 void SrsProtocol::set_recv_timeout(int64_t timeout_us) 322 void SrsProtocol::set_recv_timeout(int64_t timeout_us)
@@ -407,9 +405,11 @@ int SrsProtocol::decode_message(SrsMessage* msg, SrsPacket** ppacket) @@ -407,9 +405,11 @@ int SrsProtocol::decode_message(SrsMessage* msg, SrsPacket** ppacket)
407 srs_assert(msg->payload != NULL); 405 srs_assert(msg->payload != NULL);
408 srs_assert(msg->size > 0); 406 srs_assert(msg->size > 0);
409 407
  408 + SrsStream stream;
  409 +
410 // initialize the decode stream for all message, 410 // initialize the decode stream for all message,
411 // it's ok for the initialize if fast and without memory copy. 411 // it's ok for the initialize if fast and without memory copy.
412 - if ((ret = decode_stream->initialize((char*)(msg->payload), msg->size)) != ERROR_SUCCESS) { 412 + if ((ret = stream.initialize((char*)(msg->payload), msg->size)) != ERROR_SUCCESS) {
413 srs_error("initialize stream failed. ret=%d", ret); 413 srs_error("initialize stream failed. ret=%d", ret);
414 return ret; 414 return ret;
415 } 415 }
@@ -417,7 +417,7 @@ int SrsProtocol::decode_message(SrsMessage* msg, SrsPacket** ppacket) @@ -417,7 +417,7 @@ int SrsProtocol::decode_message(SrsMessage* msg, SrsPacket** ppacket)
417 417
418 // decode the packet. 418 // decode the packet.
419 SrsPacket* packet = NULL; 419 SrsPacket* packet = NULL;
420 - if ((ret = do_decode_message(msg->header, decode_stream, &packet)) != ERROR_SUCCESS) { 420 + if ((ret = do_decode_message(msg->header, &stream, &packet)) != ERROR_SUCCESS) {
421 srs_freep(packet); 421 srs_freep(packet);
422 return ret; 422 return ret;
423 } 423 }
@@ -445,6 +445,8 @@ int SrsProtocol::do_send_message(SrsMessage* msg, SrsPacket* packet) @@ -445,6 +445,8 @@ int SrsProtocol::do_send_message(SrsMessage* msg, SrsPacket* packet)
445 // p set to current write position, 445 // p set to current write position,
446 // it's ok when payload is NULL and size is 0. 446 // it's ok when payload is NULL and size is 0.
447 char* p = (char*)msg->payload; 447 char* p = (char*)msg->payload;
  448 + // to directly set the field.
  449 + char* pp = NULL;
448 450
449 // always write the header event payload is empty. 451 // always write the header event payload is empty.
450 do { 452 do {
@@ -871,14 +873,14 @@ int SrsProtocol::read_basic_header(char& fmt, int& cid, int& bh_size) @@ -871,14 +873,14 @@ int SrsProtocol::read_basic_header(char& fmt, int& cid, int& bh_size)
871 int ret = ERROR_SUCCESS; 873 int ret = ERROR_SUCCESS;
872 874
873 int required_size = 1; 875 int required_size = 1;
874 - if ((ret = buffer->grow(skt, required_size)) != ERROR_SUCCESS) { 876 + if ((ret = in_buffer->grow(skt, required_size)) != ERROR_SUCCESS) {
875 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { 877 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
876 srs_error("read 1bytes basic header failed. required_size=%d, ret=%d", required_size, ret); 878 srs_error("read 1bytes basic header failed. required_size=%d, ret=%d", required_size, ret);
877 } 879 }
878 return ret; 880 return ret;
879 } 881 }
880 882
881 - char* p = buffer->bytes(); 883 + char* p = in_buffer->bytes();
882 884
883 fmt = (*p >> 6) & 0x03; 885 fmt = (*p >> 6) & 0x03;
884 cid = *p & 0x3f; 886 cid = *p & 0x3f;
@@ -891,7 +893,7 @@ int SrsProtocol::read_basic_header(char& fmt, int& cid, int& bh_size) @@ -891,7 +893,7 @@ int SrsProtocol::read_basic_header(char& fmt, int& cid, int& bh_size)
891 893
892 if (cid == 0) { 894 if (cid == 0) {
893 required_size = 2; 895 required_size = 2;
894 - if ((ret = buffer->grow(skt, required_size)) != ERROR_SUCCESS) { 896 + if ((ret = in_buffer->grow(skt, required_size)) != ERROR_SUCCESS) {
895 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { 897 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
896 srs_error("read 2bytes basic header failed. required_size=%d, ret=%d", required_size, ret); 898 srs_error("read 2bytes basic header failed. required_size=%d, ret=%d", required_size, ret);
897 } 899 }
@@ -904,7 +906,7 @@ int SrsProtocol::read_basic_header(char& fmt, int& cid, int& bh_size) @@ -904,7 +906,7 @@ int SrsProtocol::read_basic_header(char& fmt, int& cid, int& bh_size)
904 srs_verbose("%dbytes basic header parsed. fmt=%d, cid=%d", bh_size, fmt, cid); 906 srs_verbose("%dbytes basic header parsed. fmt=%d, cid=%d", bh_size, fmt, cid);
905 } else if (cid == 1) { 907 } else if (cid == 1) {
906 required_size = 3; 908 required_size = 3;
907 - if ((ret = buffer->grow(skt, 3)) != ERROR_SUCCESS) { 909 + if ((ret = in_buffer->grow(skt, 3)) != ERROR_SUCCESS) {
908 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { 910 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
909 srs_error("read 3bytes basic header failed. required_size=%d, ret=%d", required_size, ret); 911 srs_error("read 3bytes basic header failed. required_size=%d, ret=%d", required_size, ret);
910 } 912 }
@@ -997,13 +999,13 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz @@ -997,13 +999,13 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz
997 srs_verbose("calc chunk message header size. fmt=%d, mh_size=%d", fmt, mh_size); 999 srs_verbose("calc chunk message header size. fmt=%d, mh_size=%d", fmt, mh_size);
998 1000
999 int required_size = bh_size + mh_size; 1001 int required_size = bh_size + mh_size;
1000 - if ((ret = buffer->grow(skt, required_size)) != ERROR_SUCCESS) { 1002 + if ((ret = in_buffer->grow(skt, required_size)) != ERROR_SUCCESS) {
1001 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { 1003 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
1002 srs_error("read %dbytes message header failed. required_size=%d, ret=%d", mh_size, required_size, ret); 1004 srs_error("read %dbytes message header failed. required_size=%d, ret=%d", mh_size, required_size, ret);
1003 } 1005 }
1004 return ret; 1006 return ret;
1005 } 1007 }
1006 - char* p = buffer->bytes() + bh_size; 1008 + char* p = in_buffer->bytes() + bh_size;
1007 1009
1008 /** 1010 /**
1009 * parse the message header. 1011 * parse the message header.
@@ -1107,7 +1109,7 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz @@ -1107,7 +1109,7 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz
1107 mh_size += 4; 1109 mh_size += 4;
1108 required_size = bh_size + mh_size; 1110 required_size = bh_size + mh_size;
1109 srs_verbose("read header ext time. fmt=%d, ext_time=%d, mh_size=%d", fmt, chunk->extended_timestamp, mh_size); 1111 srs_verbose("read header ext time. fmt=%d, ext_time=%d, mh_size=%d", fmt, chunk->extended_timestamp, mh_size);
1110 - if ((ret = buffer->grow(skt, required_size)) != ERROR_SUCCESS) { 1112 + if ((ret = in_buffer->grow(skt, required_size)) != ERROR_SUCCESS) {
1111 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { 1113 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
1112 srs_error("read %dbytes message header failed. required_size=%d, ret=%d", mh_size, required_size, ret); 1114 srs_error("read %dbytes message header failed. required_size=%d, ret=%d", mh_size, required_size, ret);
1113 } 1115 }
@@ -1210,7 +1212,7 @@ int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh @@ -1210,7 +1212,7 @@ int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh
1210 // empty message 1212 // empty message
1211 if (chunk->header.payload_length <= 0) { 1213 if (chunk->header.payload_length <= 0) {
1212 // need erase the header in buffer. 1214 // need erase the header in buffer.
1213 - buffer->erase(bh_size + mh_size); 1215 + in_buffer->erase(bh_size + mh_size);
1214 1216
1215 srs_trace("get an empty RTMP " 1217 srs_trace("get an empty RTMP "
1216 "message(type=%d, size=%d, time=%"PRId64", sid=%d)", chunk->header.message_type, 1218 "message(type=%d, size=%d, time=%"PRId64", sid=%d)", chunk->header.message_type,
@@ -1238,14 +1240,14 @@ int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh @@ -1238,14 +1240,14 @@ int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh
1238 1240
1239 // read payload to buffer 1241 // read payload to buffer
1240 int required_size = bh_size + mh_size + payload_size; 1242 int required_size = bh_size + mh_size + payload_size;
1241 - if ((ret = buffer->grow(skt, required_size)) != ERROR_SUCCESS) { 1243 + if ((ret = in_buffer->grow(skt, required_size)) != ERROR_SUCCESS) {
1242 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { 1244 if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
1243 srs_error("read payload failed. required_size=%d, ret=%d", required_size, ret); 1245 srs_error("read payload failed. required_size=%d, ret=%d", required_size, ret);
1244 } 1246 }
1245 return ret; 1247 return ret;
1246 } 1248 }
1247 - memcpy(chunk->msg->payload + chunk->msg->size, buffer->bytes() + bh_size + mh_size, payload_size);  
1248 - buffer->erase(bh_size + mh_size + payload_size); 1249 + memcpy(chunk->msg->payload + chunk->msg->size, in_buffer->bytes() + bh_size + mh_size, payload_size);
  1250 + in_buffer->erase(bh_size + mh_size + payload_size);
1249 chunk->msg->size += payload_size; 1251 chunk->msg->size += payload_size;
1250 1252
1251 srs_verbose("chunk payload read completed. bh_size=%d, mh_size=%d, payload_size=%d", bh_size, mh_size, payload_size); 1253 srs_verbose("chunk payload read completed. bh_size=%d, mh_size=%d, payload_size=%d", bh_size, mh_size, payload_size);
@@ -97,8 +97,10 @@ private: @@ -97,8 +97,10 @@ private:
97 }; 97 };
98 // peer in/out 98 // peer in/out
99 private: 99 private:
  100 + /**
  101 + * underlayer socket object, send/recv bytes.
  102 + */
100 ISrsProtocolReaderWriter* skt; 103 ISrsProtocolReaderWriter* skt;
101 - char* pp;  
102 /** 104 /**
103 * requests sent out, used to build the response. 105 * requests sent out, used to build the response.
104 * key: transactionId 106 * key: transactionId
@@ -107,14 +109,33 @@ private: @@ -107,14 +109,33 @@ private:
107 std::map<double, std::string> requests; 109 std::map<double, std::string> requests;
108 // peer in 110 // peer in
109 private: 111 private:
  112 + /**
  113 + * chunk stream to decode RTMP messages.
  114 + */
110 std::map<int, SrsChunkStream*> chunk_streams; 115 std::map<int, SrsChunkStream*> chunk_streams;
111 - SrsStream* decode_stream;  
112 - SrsBuffer* buffer; 116 + /**
  117 + * bytes buffer cache, recv from skt, provide services for stream.
  118 + */
  119 + SrsBuffer* in_buffer;
  120 + /**
  121 + * input chunk size, default to 128, set by peer packet.
  122 + */
113 int32_t in_chunk_size; 123 int32_t in_chunk_size;
  124 + /**
  125 + * input ack size, when to send the acked packet.
  126 + */
114 AckWindowSize in_ack_size; 127 AckWindowSize in_ack_size;
115 // peer out 128 // peer out
116 private: 129 private:
  130 + /**
  131 + * output header cache.
  132 + * used for type0, 11bytes(or 15bytes with extended timestamp) header.
  133 + * or for type3, 1bytes(or 5bytes with extended timestamp) header.
  134 + */
117 char out_header_cache[RTMP_MAX_FMT0_HEADER_SIZE]; 135 char out_header_cache[RTMP_MAX_FMT0_HEADER_SIZE];
  136 + /**
  137 + * output chunk size, default to 128, set by config.
  138 + */
118 int32_t out_chunk_size; 139 int32_t out_chunk_size;
119 public: 140 public:
120 /** 141 /**