support librtmp/ffmpeg without extended timestamp in 0xCX chunk packet. update signautre of server

@@ -46,6 +46,7 @@ url: rtmp://127.0.0.1:1935/live/livestream
 * nginx v1.5.0: 139524 lines <br/>
 * nginx v1.5.0: 139524 lines <br/>
 
 
 ### History
 ### History
+* v0.3, 2013-10-28, support librtmp without extended-timestamp in 0xCX chunk packet.
 * v0.3, 2013-10-27, support cache last gop for client fast startup.
 * v0.3, 2013-10-27, support cache last gop for client fast startup.
 * v0.2, 2013-10-25, v0.2 released. 10125 lines.
 * v0.2, 2013-10-25, v0.2 released. 10125 lines.
 * v0.2, 2013-10-25, support flash publish.
 * v0.2, 2013-10-25, support flash publish.

@@ -62,13 +62,18 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 	} \
 	} \
 	(void)0
 	(void)0
 
 
+// current release version
+#define RTMP_SIG_SRS_VERSION "0.2"
 // server info.
 // server info.
 #define RTMP_SIG_SRS_KEY "srs"
 #define RTMP_SIG_SRS_KEY "srs"
+#define RTMP_SIG_SRS_ROLE "origin server"
 #define RTMP_SIG_SRS_NAME RTMP_SIG_SRS_KEY"(simple rtmp server)"
 #define RTMP_SIG_SRS_NAME RTMP_SIG_SRS_KEY"(simple rtmp server)"
-#define RTMP_SIG_SRS_URL "https://github.com/winlinvip/simple-rtmp-server"
+#define RTMP_SIG_SRS_URL "https://"RTMP_SIG_SRS_URL_SHORT
+#define RTMP_SIG_SRS_URL_SHORT "github.com/winlinvip/simple-rtmp-server"
 #define RTMP_SIG_SRS_WEB "http://blog.csdn.net/win_lin"
 #define RTMP_SIG_SRS_WEB "http://blog.csdn.net/win_lin"
 #define RTMP_SIG_SRS_EMAIL "winterserver@126.com"
 #define RTMP_SIG_SRS_EMAIL "winterserver@126.com"
-#define RTMP_SIG_SRS_VERSION "0.1"
+#define RTMP_SIG_SRS_LICENSE "The MIT License (MIT)"
+#define RTMP_SIG_SRS_COPYRIGHT "Copyright (c) 2013 winlin"
 
 
 // compare
 // compare
 #define srs_min(a, b) ((a < b)? a : b)
 #define srs_min(a, b) ((a < b)? a : b)

@@ -25,6 +25,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 #include <srs_core_error.hpp>
 #include <srs_core_error.hpp>
 #include <srs_core_socket.hpp>
 #include <srs_core_socket.hpp>
+#include <srs_core_log.hpp>
 
 
 #define SOCKET_READ_SIZE 4096
 #define SOCKET_READ_SIZE 4096
 
 
@@ -62,7 +63,11 @@ int SrsBuffer::ensure_buffer_bytes(SrsSocket* skt, int required_size)
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
 
 
-	srs_assert(required_size >= 0);
+	if (required_size < 0) {
+		ret = ERROR_SYSTEM_SIZE_NEGATIVE;
+		srs_error("size is negative. size=%d, ret=%d", required_size, ret);
+		return ret;
+	}
 
 
 	while (size() < required_size) {
 	while (size() < required_size) {
 		char buffer[SOCKET_READ_SIZE];
 		char buffer[SOCKET_READ_SIZE];

@@ -63,8 +63,8 @@ int SrsClient::do_cycle()
 	}
 	}
 	srs_verbose("get peer ip success. ip=%s", ip);
 	srs_verbose("get peer ip success. ip=%s", ip);
 
 
-	rtmp->set_recv_timeout(SRS_SEND_TIMEOUT_MS);
-	rtmp->set_send_timeout(SRS_SEND_TIMEOUT_MS);
+	rtmp->set_recv_timeout(SRS_SEND_TIMEOUT_MS * 1000);
+	rtmp->set_send_timeout(SRS_SEND_TIMEOUT_MS * 1000);
 	
 	
 	if ((ret = rtmp->handshake()) != ERROR_SUCCESS) {
 	if ((ret = rtmp->handshake()) != ERROR_SUCCESS) {
 		srs_error("rtmp handshake failed. ret=%d", ret);
 		srs_error("rtmp handshake failed. ret=%d", ret);
@@ -185,7 +185,7 @@ int SrsClient::streaming_play(SrsSource* source)
 	SrsAutoFree(SrsConsumer, consumer, false);
 	SrsAutoFree(SrsConsumer, consumer, false);
 	srs_verbose("consumer created success.");
 	srs_verbose("consumer created success.");
 	
 	
-	rtmp->set_recv_timeout(SRS_PULSE_TIMEOUT_MS);
+	rtmp->set_recv_timeout(SRS_PULSE_TIMEOUT_MS * 1000);
 	
 	
 	int64_t report_time = 0;
 	int64_t report_time = 0;
 	int64_t reported_time = 0;
 	int64_t reported_time = 0;

@@ -64,11 +64,13 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define ERROR_RTMP_CHUNK_SIZE			310
 #define ERROR_RTMP_CHUNK_SIZE			310
 #define ERROR_RTMP_TRY_SIMPLE_HS		311
 #define ERROR_RTMP_TRY_SIMPLE_HS		311
 #define ERROR_RTMP_CH_SCHEMA			312
 #define ERROR_RTMP_CH_SCHEMA			312
+#define ERROR_RTMP_PACKET_SIZE			313
 
 
 #define ERROR_SYSTEM_STREAM_INIT		400
 #define ERROR_SYSTEM_STREAM_INIT		400
 #define ERROR_SYSTEM_PACKET_INVALID		401
 #define ERROR_SYSTEM_PACKET_INVALID		401
 #define ERROR_SYSTEM_CLIENT_INVALID		402
 #define ERROR_SYSTEM_CLIENT_INVALID		402
 #define ERROR_SYSTEM_ASSERT_FAILED		403
 #define ERROR_SYSTEM_ASSERT_FAILED		403
+#define ERROR_SYSTEM_SIZE_NEGATIVE		404
 
 
 // see librtmp.
 // see librtmp.
 // failed when open ssl create the dh
 // failed when open ssl create the dh

@@ -249,6 +249,9 @@ messages.
 /****************************************************************************
 /****************************************************************************
 *****************************************************************************
 *****************************************************************************
 ****************************************************************************/
 ****************************************************************************/
+// when got a messae header, increase recv timeout to got an entire message.
+#define SRS_MIN_RECV_TIMEOUT_US 3000
+
 SrsProtocol::AckWindowSize::AckWindowSize()
 SrsProtocol::AckWindowSize::AckWindowSize()
 {
 {
 	ack_window_size = acked_size = 0;
 	ack_window_size = acked_size = 0;
@@ -278,14 +281,19 @@ SrsProtocol::~SrsProtocol()
 	srs_freep(skt);
 	srs_freep(skt);
 }
 }
 
 
-void SrsProtocol::set_recv_timeout(int timeout_ms)
+void SrsProtocol::set_recv_timeout(int64_t timeout_us)
+{
+	return skt->set_recv_timeout(timeout_us);
+}
+
+int64_t SrsProtocol::get_recv_timeout()
 {
 {
-	return skt->set_recv_timeout(timeout_ms);
+	return skt->get_recv_timeout();
 }
 }
 
 
-void SrsProtocol::set_send_timeout(int timeout_ms)
+void SrsProtocol::set_send_timeout(int64_t timeout_us)
 {
 {
-	return skt->set_send_timeout(timeout_ms);
+	return skt->set_send_timeout(timeout_us);
 }
 }
 
 
 int SrsProtocol::recv_message(SrsCommonMessage** pmsg)
 int SrsProtocol::recv_message(SrsCommonMessage** pmsg)
@@ -602,17 +610,31 @@ int SrsProtocol::recv_interlaced_message(SrsCommonMessage** pmsg)
 		}
 		}
 		return ret;
 		return ret;
 	}
 	}
-	srs_info("read basic header success. fmt=%d, cid=%d, bh_size=%d", fmt, cid, bh_size);
+	srs_verbose("read basic header success. fmt=%d, cid=%d, bh_size=%d", fmt, cid, bh_size);
+	
+	// once we got the chunk message header, 
+	// that is there is a real message in cache,
+	// increase the timeout to got it.
+	// For example, in the play loop, we set timeout to 100ms,
+	// when we got a chunk header, we should increase the timeout,
+	// or we maybe timeout and disconnect the client.
+	int64_t timeout_us = skt->get_recv_timeout();
+	if (timeout_us != (int64_t)ST_UTIME_NO_TIMEOUT) {
+		int64_t pkt_timeout_us = srs_max(timeout_us, SRS_MIN_RECV_TIMEOUT_US);
+		skt->set_recv_timeout(pkt_timeout_us);
+		srs_verbose("change recv timeout_us "
+			"from %"PRId64" to %"PRId64"", timeout_us, pkt_timeout_us);
+	}
 	
 	
 	// get the cached chunk stream.
 	// get the cached chunk stream.
 	SrsChunkStream* chunk = NULL;
 	SrsChunkStream* chunk = NULL;
 	
 	
 	if (chunk_streams.find(cid) == chunk_streams.end()) {
 	if (chunk_streams.find(cid) == chunk_streams.end()) {
 		chunk = chunk_streams[cid] = new SrsChunkStream(cid);
 		chunk = chunk_streams[cid] = new SrsChunkStream(cid);
-		srs_info("cache new chunk stream: fmt=%d, cid=%d", fmt, cid);
+		srs_verbose("cache new chunk stream: fmt=%d, cid=%d", fmt, cid);
 	} else {
 	} else {
 		chunk = chunk_streams[cid];
 		chunk = chunk_streams[cid];
-		srs_info("cached chunk stream: fmt=%d, cid=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)",
+		srs_verbose("cached chunk stream: fmt=%d, cid=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)",
 			chunk->fmt, chunk->cid, (chunk->msg? chunk->msg->size : 0), chunk->header.message_type, chunk->header.payload_length,
 			chunk->fmt, chunk->cid, (chunk->msg? chunk->msg->size : 0), chunk->header.message_type, chunk->header.payload_length,
 			chunk->header.timestamp, chunk->header.stream_id);
 			chunk->header.timestamp, chunk->header.stream_id);
 	}
 	}
@@ -625,7 +647,7 @@ int SrsProtocol::recv_interlaced_message(SrsCommonMessage** pmsg)
 		}
 		}
 		return ret;
 		return ret;
 	}
 	}
-	srs_info("read message header success. "
+	srs_verbose("read message header success. "
 			"fmt=%d, mh_size=%d, ext_time=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)", 
 			"fmt=%d, mh_size=%d, ext_time=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)", 
 			fmt, mh_size, chunk->extended_timestamp, (chunk->msg? chunk->msg->size : 0), chunk->header.message_type, 
 			fmt, mh_size, chunk->extended_timestamp, (chunk->msg? chunk->msg->size : 0), chunk->header.message_type, 
 			chunk->header.payload_length, chunk->header.timestamp, chunk->header.stream_id);
 			chunk->header.payload_length, chunk->header.timestamp, chunk->header.stream_id);
@@ -640,9 +662,15 @@ int SrsProtocol::recv_interlaced_message(SrsCommonMessage** pmsg)
 		return ret;
 		return ret;
 	}
 	}
 	
 	
+	// reset the recv timeout
+	if (timeout_us != (int64_t)ST_UTIME_NO_TIMEOUT) {
+		skt->set_recv_timeout(timeout_us);
+		srs_verbose("reset recv timeout_us to %"PRId64"", timeout_us);
+	}
+	
 	// not got an entire RTMP message, try next chunk.
 	// not got an entire RTMP message, try next chunk.
 	if (!msg) {
 	if (!msg) {
-		srs_info("get partial message success. chunk_payload_size=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)",
+		srs_verbose("get partial message success. chunk_payload_size=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)",
 				payload_size, (msg? msg->size : (chunk->msg? chunk->msg->size : 0)), chunk->header.message_type, chunk->header.payload_length,
 				payload_size, (msg? msg->size : (chunk->msg? chunk->msg->size : 0)), chunk->header.message_type, chunk->header.payload_length,
 				chunk->header.timestamp, chunk->header.stream_id);
 				chunk->header.timestamp, chunk->header.stream_id);
 		return ret;
 		return ret;
@@ -827,6 +855,15 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz
             pp[0] = *p++;
             pp[0] = *p++;
             pp[3] = 0;
             pp[3] = 0;
             
             
+            // if msg exists in cache, the size must not changed.
+            if (chunk->msg->size > 0 && chunk->msg->size != chunk->header.payload_length) {
+                ret = ERROR_RTMP_PACKET_SIZE;
+                srs_error("msg exists in chunk cache, "
+                	"size=%d cannot change to %d, ret=%d", 
+                	chunk->msg->size, chunk->header.payload_length, ret);
+                return ret;
+            }
+            
             chunk->header.message_type = *p++;
             chunk->header.message_type = *p++;
             
             
             if (fmt == 0) {
             if (fmt == 0) {
@@ -867,11 +904,21 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz
 			return ret;
 			return ret;
 		}
 		}
 
 
-        char* pp = (char*)&chunk->header.timestamp;
+		// ffmpeg/librtmp may donot send this filed, need to detect the value.
+		// @see also: http://blog.csdn.net/win_lin/article/details/13363699
+		int32_t timestamp = 0x00;
+        char* pp = (char*)&timestamp;
         pp[3] = *p++;
         pp[3] = *p++;
         pp[2] = *p++;
         pp[2] = *p++;
         pp[1] = *p++;
         pp[1] = *p++;
         pp[0] = *p++;
         pp[0] = *p++;
+        
+        if (chunk->header.timestamp > RTMP_EXTENDED_TIMESTAMP && chunk->header.timestamp != timestamp) {
+            mh_size -= 4;
+            srs_verbose("ignore the 4bytes extended timestamp. mh_size=%d", mh_size);
+        } else {
+            chunk->header.timestamp = timestamp;
+        }
 		srs_verbose("header read ext_time completed. time=%d", chunk->header.timestamp);
 		srs_verbose("header read ext_time completed. time=%d", chunk->header.timestamp);
 	}
 	}
 	
 	
@@ -897,7 +944,7 @@ int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
 	
 	
 	// empty message
 	// empty message
-	if (chunk->header.payload_length == 0) {
+	if (chunk->header.payload_length <= 0) {
 		// need erase the header in buffer.
 		// need erase the header in buffer.
 		buffer->erase(bh_size + mh_size);
 		buffer->erase(bh_size + mh_size);
 		
 		

@@ -104,11 +104,12 @@ public:
 	virtual ~SrsProtocol();
 	virtual ~SrsProtocol();
 public:
 public:
 	/**
 	/**
-	* set the timeout in ms.
+	* set the timeout in us.
 	* if timeout, recv/send message return ERROR_SOCKET_TIMEOUT.
 	* if timeout, recv/send message return ERROR_SOCKET_TIMEOUT.
 	*/
 	*/
-	virtual void set_recv_timeout(int timeout_ms);
-	virtual void set_send_timeout(int timeout_ms);
+	virtual void set_recv_timeout(int64_t timeout_us);
+	virtual int64_t get_recv_timeout();
+	virtual void set_send_timeout(int64_t timeout_us);
 	/**
 	/**
 	* recv a message with raw/undecoded payload from peer.
 	* recv a message with raw/undecoded payload from peer.
 	* the payload is not decoded, use srs_rtmp_expect_message<T> if requires 
 	* the payload is not decoded, use srs_rtmp_expect_message<T> if requires 

@@ -149,14 +149,19 @@ SrsRtmp::~SrsRtmp()
 	srs_freep(complex_handshake);
 	srs_freep(complex_handshake);
 }
 }
 
 
-void SrsRtmp::set_recv_timeout(int timeout_ms)
+void SrsRtmp::set_recv_timeout(int64_t timeout_us)
 {
 {
-	return protocol->set_recv_timeout(timeout_ms);
+	return protocol->set_recv_timeout(timeout_us);
 }
 }
 
 
-void SrsRtmp::set_send_timeout(int timeout_ms)
+int64_t SrsRtmp::get_recv_timeout()
 {
 {
-	return protocol->set_send_timeout(timeout_ms);
+	return protocol->get_recv_timeout();
+}
+
+void SrsRtmp::set_send_timeout(int64_t timeout_us)
+{
+	return protocol->set_send_timeout(timeout_us);
 }
 }
 
 
 int SrsRtmp::recv_message(SrsCommonMessage** pmsg)
 int SrsRtmp::recv_message(SrsCommonMessage** pmsg)
@@ -176,7 +181,6 @@ int SrsRtmp::handshake()
     ssize_t nsize;
     ssize_t nsize;
     SrsSocket skt(stfd);
     SrsSocket skt(stfd);
     
     
-    // TODO: complex handshake for h264/aac codec.
     char* c0c1 = new char[1537];
     char* c0c1 = new char[1537];
     SrsAutoFree(char, c0c1, true);
     SrsAutoFree(char, c0c1, true);
     if ((ret = skt.read_fully(c0c1, 1537, &nsize)) != ERROR_SUCCESS) {
     if ((ret = skt.read_fully(c0c1, 1537, &nsize)) != ERROR_SUCCESS) {
@@ -325,10 +329,15 @@ int SrsRtmp::response_connect_app(SrsRequest* req)
 	SrsASrsAmf0EcmaArray* data = new SrsASrsAmf0EcmaArray();
 	SrsASrsAmf0EcmaArray* data = new SrsASrsAmf0EcmaArray();
 	pkt->info->set("data", data);
 	pkt->info->set("data", data);
 	
 	
-	data->set("version", new SrsAmf0String(RTMP_SIG_FMS_VER));
-	data->set("server", new SrsAmf0String(RTMP_SIG_SRS_NAME));
+	data->set("srs_version", new SrsAmf0String(RTMP_SIG_FMS_VER));
+	data->set("srs_server", new SrsAmf0String(RTMP_SIG_SRS_NAME));
+	data->set("srs_license", new SrsAmf0String(RTMP_SIG_SRS_LICENSE));
+	data->set("srs_role", new SrsAmf0String(RTMP_SIG_SRS_ROLE));
 	data->set("srs_url", new SrsAmf0String(RTMP_SIG_SRS_URL));
 	data->set("srs_url", new SrsAmf0String(RTMP_SIG_SRS_URL));
 	data->set("srs_version", new SrsAmf0String(RTMP_SIG_SRS_VERSION));
 	data->set("srs_version", new SrsAmf0String(RTMP_SIG_SRS_VERSION));
+	data->set("srs_site", new SrsAmf0String(RTMP_SIG_SRS_WEB));
+	data->set("srs_email", new SrsAmf0String(RTMP_SIG_SRS_EMAIL));
+	data->set("srs_copyright", new SrsAmf0String(RTMP_SIG_SRS_COPYRIGHT));
 	
 	
 	msg->set_packet(pkt, 0);
 	msg->set_packet(pkt, 0);
 	
 	

@@ -105,8 +105,9 @@ public:
 	SrsRtmp(st_netfd_t client_stfd);
 	SrsRtmp(st_netfd_t client_stfd);
 	virtual ~SrsRtmp();
 	virtual ~SrsRtmp();
 public:
 public:
-	virtual void set_recv_timeout(int timeout_ms);
-	virtual void set_send_timeout(int timeout_ms);
+	virtual void set_recv_timeout(int64_t timeout_us);
+	virtual int64_t get_recv_timeout();
+	virtual void set_send_timeout(int64_t timeout_us);
 	virtual int recv_message(SrsCommonMessage** pmsg);
 	virtual int recv_message(SrsCommonMessage** pmsg);
 	virtual int send_message(ISrsMessage* msg);
 	virtual int send_message(ISrsMessage* msg);
 public:
 public:

@@ -36,14 +36,19 @@ SrsSocket::~SrsSocket()
 {
 {
 }
 }
 
 
-void SrsSocket::set_recv_timeout(int timeout_ms)
+void SrsSocket::set_recv_timeout(int64_t timeout_us)
 {
 {
-	recv_timeout = timeout_ms * 1000;
+	recv_timeout = timeout_us;
 }
 }
 
 
-void SrsSocket::set_send_timeout(int timeout_ms)
+int64_t SrsSocket::get_recv_timeout()
 {
 {
-	send_timeout = timeout_ms * 1000;
+	return recv_timeout;
+}
+
+void SrsSocket::set_send_timeout(int64_t timeout_us)
+{
+	send_timeout = timeout_us;
 }
 }
 
 
 int64_t SrsSocket::get_recv_bytes()
 int64_t SrsSocket::get_recv_bytes()

@@ -48,8 +48,9 @@ public:
     SrsSocket(st_netfd_t client_stfd);
     SrsSocket(st_netfd_t client_stfd);
     virtual ~SrsSocket();
     virtual ~SrsSocket();
 public:
 public:
-	virtual void set_recv_timeout(int timeout_ms);
-	virtual void set_send_timeout(int timeout_ms);
+	virtual void set_recv_timeout(int64_t timeout_us);
+	virtual int64_t get_recv_timeout();
+	virtual void set_send_timeout(int64_t timeout_us);
 	virtual int64_t get_recv_bytes();
 	virtual int64_t get_recv_bytes();
 	virtual int64_t get_send_bytes();
 	virtual int64_t get_send_bytes();
 public:
 public:

@@ -179,8 +179,8 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
 	
 	
-	metadata->metadata->set("server", 
-		new SrsAmf0String(RTMP_SIG_SRS_NAME""RTMP_SIG_SRS_VERSION));
+	metadata->metadata->set("server", new SrsAmf0String(
+		RTMP_SIG_SRS_KEY" "RTMP_SIG_SRS_VERSION" ("RTMP_SIG_SRS_URL_SHORT")"));
 	
 	
 	// encode the metadata to payload
 	// encode the metadata to payload
 	int size = metadata->get_payload_length();
 	int size = metadata->get_payload_length();