decode the amf0 command message: connect.

@@ -22,3 +22,69 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */
 */
 
 
 #include <srs_core_amf0.hpp>
 #include <srs_core_amf0.hpp>
+
+#include <srs_core_log.hpp>
+#include <srs_core_stream.hpp>
+
+// AMF0 marker
+#define RTMP_AMF0_Number 					0x00
+#define RTMP_AMF0_Boolean 					0x01
+#define RTMP_AMF0_String 					0x02
+#define RTMP_AMF0_Object 					0x03
+#define RTMP_AMF0_MovieClip 				0x04 // reserved, not supported
+#define RTMP_AMF0_Null 						0x05
+#define RTMP_AMF0_Undefined 				0x06
+#define RTMP_AMF0_Reference 				0x07
+#define RTMP_AMF0_EcmaArray 				0x08
+#define RTMP_AMF0_ObjectEnd 				0x09
+#define RTMP_AMF0_StrictArray 				0x0A
+#define RTMP_AMF0_Date 						0x0B
+#define RTMP_AMF0_LongString 				0x0C
+#define RTMP_AMF0_UnSupported 				0x0D
+#define RTMP_AMF0_RecordSet 				0x0E // reserved, not supported
+#define RTMP_AMF0_XmlDocument 				0x0F
+#define RTMP_AMF0_TypedObject 				0x10
+// AVM+ object is the AMF3 object.
+#define RTMP_AMF0_AVMplusObject 			0x11
+// origin array whos data takes the same form as LengthValueBytes
+#define RTMP_AMF0_OriginStrictArray 		0x20
+
+std::string srs_amf0_read_string(SrsStream* stream)
+{
+	std::string str;
+	
+	// marker
+	if (!stream->require(1)) {
+		return str;
+	}
+	
+	char marker = stream->read_char();
+	if (marker != RTMP_AMF0_String) {
+		return str;
+	}
+	
+	// len
+	if (!stream->require(2)) {
+		return str;
+	}
+	int16_t len = stream->read_2bytes();
+	
+	// data
+	if (!stream->require(len)) {
+		return str;
+	}
+	str = stream->read_string(len);
+	
+	// support utf8-1 only
+	// 1.3.1 Strings and UTF-8
+	// UTF8-1 = %x00-7F
+	for (int i = 0; i < len; i++) {
+		char ch = *(str.data() + i);
+		if ((ch & 0x80) != 0) {
+			srs_warn("only support utf8-1, 0x00-0x7F, actual is %#x", (int)ch);
+			return "";
+		}
+	}
+	
+	return str;
+}

@@ -30,4 +30,13 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 #include <srs_core.hpp>
 #include <srs_core.hpp>
 
 
+#include <string>
+
+class SrsStream;
+
+/**
+* read amf0 string from stream.
+*/
+extern std::string srs_amf0_read_string(SrsStream* stream);
+
 #endif
 #endif

@@ -62,6 +62,8 @@ int SrsBuffer::ensure_buffer_bytes(SrsSocket* skt, int required_size)
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
 
 
+	srs_assert(required_size >= 0);
+
 	while (size() < required_size) {
 	while (size() < required_size) {
 		char buffer[SOCKET_READ_SIZE];
 		char buffer[SOCKET_READ_SIZE];
 		
 		

@@ -52,6 +52,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define ERROR_RTMP_PLAIN_REQUIRED		300
 #define ERROR_RTMP_PLAIN_REQUIRED		300
 #define ERROR_RTMP_CHUNK_START			301
 #define ERROR_RTMP_CHUNK_START			301
 #define ERROR_RTMP_MSG_INVLIAD_SIZE		302
 #define ERROR_RTMP_MSG_INVLIAD_SIZE		302
+#define ERROR_RTMP_AMF0_DECODE			303
 
 
 #define ERROR_SYSTEM_STREAM_INIT		400
 #define ERROR_SYSTEM_STREAM_INIT		400
 
 

@@ -24,9 +24,11 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_core_protocol.hpp>
 #include <srs_core_protocol.hpp>
 
 
 #include <srs_core_log.hpp>
 #include <srs_core_log.hpp>
+#include <srs_core_amf0.hpp>
 #include <srs_core_error.hpp>
 #include <srs_core_error.hpp>
 #include <srs_core_socket.hpp>
 #include <srs_core_socket.hpp>
 #include <srs_core_buffer.hpp>
 #include <srs_core_buffer.hpp>
+#include <srs_core_stream.hpp>
 
 
 /**
 /**
 5. Protocol Control Messages
 5. Protocol Control Messages
@@ -187,6 +189,11 @@ messages.
 */
 */
 #define RTMP_EXTENDED_TIMESTAMP 			0xFFFFFF
 #define RTMP_EXTENDED_TIMESTAMP 			0xFFFFFF
 
 
+/**
+* amf0 command message, command name: "connect"
+*/
+#define RTMP_AMF0_COMMAND_CONNECT			"connect"
+
 SrsMessageHeader::SrsMessageHeader()
 SrsMessageHeader::SrsMessageHeader()
 {
 {
 	message_type = 0;
 	message_type = 0;
@@ -218,6 +225,7 @@ SrsChunkStream::~SrsChunkStream()
 SrsMessage::SrsMessage()
 SrsMessage::SrsMessage()
 {
 {
 	size = 0;
 	size = 0;
+	stream = NULL;
 	payload = NULL;
 	payload = NULL;
 	decoded_payload = NULL;
 	decoded_payload = NULL;
 }
 }
@@ -233,6 +241,11 @@ SrsMessage::~SrsMessage()
 		delete decoded_payload;
 		delete decoded_payload;
 		decoded_payload = NULL;
 		decoded_payload = NULL;
 	}
 	}
+	
+	if (stream) {
+		delete stream;
+		stream = NULL;
+	}
 }
 }
 
 
 SrsPacket* SrsMessage::get_packet()
 SrsPacket* SrsMessage::get_packet()
@@ -249,7 +262,44 @@ int SrsMessage::decode_packet()
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
 	
 	
-	// TODO: decode packet.
+	srs_assert(payload != NULL);
+	srs_assert(size > 0);
+	
+	if (!stream) {
+		srs_verbose("create decode stream for message.");
+		stream = new SrsStream();
+	}
+	
+	if (header.message_type == RTMP_MSG_AMF0CommandMessage) {
+		srs_verbose("start to decode AMF0 command message.");
+		
+		// amf0 command message.
+		// need to read the command name.
+		if ((ret = stream->initialize((char*)payload, size)) != ERROR_SUCCESS) {
+			srs_error("initialize stream failed. ret=%d", ret);
+			return ret;
+		}
+		srs_verbose("decode stream initialized success");
+		
+		std::string command = srs_amf0_read_string(stream);
+		srs_verbose("AMF0 command message, command_name=%s", command.c_str());
+		
+		stream->reset();
+		if (command == RTMP_AMF0_COMMAND_CONNECT) {
+			srs_info("decode the AMF0 command(connect vhost/app message).");
+			decoded_payload = new SrsConnectAppPacket();
+			return decoded_payload->decode(stream);
+		}
+		
+		// default packet to drop message.
+		srs_trace("drop the AMF0 command message, command_name=%s", command.c_str());
+		decoded_payload = new SrsPacket();
+		return ret;
+	}
+	
+	// default packet to drop message.
+	srs_trace("drop the unknown message, type=%d", header.message_type);
+	decoded_payload = new SrsPacket();
 	
 	
 	return ret;
 	return ret;
 }
 }
@@ -262,6 +312,12 @@ SrsPacket::~SrsPacket()
 {
 {
 }
 }
 
 
+int SrsPacket::decode(SrsStream* /*stream*/)
+{
+	int ret = ERROR_SUCCESS;
+	return ret;
+}
+
 SrsConnectAppPacket::SrsConnectAppPacket()
 SrsConnectAppPacket::SrsConnectAppPacket()
 {
 {
 }
 }
@@ -270,6 +326,24 @@ SrsConnectAppPacket::~SrsConnectAppPacket()
 {
 {
 }
 }
 
 
+int SrsConnectAppPacket::decode(SrsStream* stream)
+{
+	int ret = ERROR_SUCCESS;
+	
+	if ((ret = super::decode(stream)) != ERROR_SUCCESS) {
+		return ret;
+	}
+
+	command_name = srs_amf0_read_string(stream);
+	if (command_name.empty()) {
+		ret = ERROR_RTMP_AMF0_DECODE;
+		srs_error("amf0 decode connect command_name failed. ret=%d", ret);
+		return ret;
+	}
+	
+	return ret;
+}
+
 SrsProtocol::SrsProtocol(st_netfd_t client_stfd)
 SrsProtocol::SrsProtocol(st_netfd_t client_stfd)
 {
 {
 	stfd = client_stfd;
 	stfd = client_stfd;
@@ -317,12 +391,21 @@ int SrsProtocol::recv_message(SrsMessage** pmsg)
 			srs_error("recv interlaced message failed. ret=%d", ret);
 			srs_error("recv interlaced message failed. ret=%d", ret);
 			return ret;
 			return ret;
 		}
 		}
+		srs_verbose("entire msg received");
 		
 		
 		if (!msg) {
 		if (!msg) {
 			continue;
 			continue;
 		}
 		}
 		
 		
-		// return the msg with raw/undecoded payload
+		if (msg->size <= 0 || msg->header.payload_length <= 0) {
+			srs_trace("ignore empty message(type=%d, size=%d, time=%d, sid=%d).",
+				msg->header.message_type, msg->header.payload_length,
+				msg->header.timestamp, msg->header.stream_id);
+			delete msg;
+			continue;
+		}
+		
+		srs_verbose("get a msg with raw/undecoded payload");
 		*pmsg = msg;
 		*pmsg = msg;
 		break;
 		break;
 	}
 	}
@@ -598,10 +681,13 @@ int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh
 		// need erase the header in buffer.
 		// need erase the header in buffer.
 		buffer->erase(bh_size + mh_size);
 		buffer->erase(bh_size + mh_size);
 		
 		
-		srs_warn("get an empty RTMP "
+		srs_trace("get an empty RTMP "
 				"message(type=%d, size=%d, time=%d, sid=%d)", chunk->header.message_type, 
 				"message(type=%d, size=%d, time=%d, sid=%d)", 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);
 		
 		
+		*pmsg = chunk->msg;
+		chunk->msg = NULL;
+				
 		return ret;
 		return ret;
 	}
 	}
 	srs_assert(chunk->header.payload_length > 0);
 	srs_assert(chunk->header.payload_length > 0);

@@ -31,6 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_core.hpp>
 #include <srs_core.hpp>
 
 
 #include <map>
 #include <map>
+#include <string>
 
 
 #include <st.h>
 #include <st.h>
 
 
@@ -40,6 +41,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 class SrsSocket;
 class SrsSocket;
 class SrsBuffer;
 class SrsBuffer;
 class SrsPacket;
 class SrsPacket;
+class SrsStream;
 class SrsMessage;
 class SrsMessage;
 class SrsChunkStream;
 class SrsChunkStream;
 
 
@@ -127,6 +129,7 @@ public:
 	int8_t* payload;
 	int8_t* payload;
 // decoded message payload.
 // decoded message payload.
 private:
 private:
+	SrsStream* stream;
 	SrsPacket* decoded_payload;
 	SrsPacket* decoded_payload;
 public:
 public:
 	/**
 	/**
@@ -150,13 +153,21 @@ class SrsPacket
 public:
 public:
 	SrsPacket();
 	SrsPacket();
 	virtual ~SrsPacket();
 	virtual ~SrsPacket();
+public:
+	virtual int decode(SrsStream* stream);
 };
 };
 
 
 class SrsConnectAppPacket : public SrsPacket
 class SrsConnectAppPacket : public SrsPacket
 {
 {
+private:
+	typedef SrsPacket super;
+private:
+	std::string command_name;
 public:
 public:
 	SrsConnectAppPacket();
 	SrsConnectAppPacket();
 	virtual ~SrsConnectAppPacket();
 	virtual ~SrsConnectAppPacket();
+public:
+	virtual int decode(SrsStream* stream);
 };
 };
 
 
 /**
 /**

@@ -28,7 +28,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 SrsStream::SrsStream()
 SrsStream::SrsStream()
 {
 {
-	bytes = NULL;
+	p = bytes = NULL;
 	size = 0;
 	size = 0;
 }
 }
 
 
@@ -53,8 +53,54 @@ int SrsStream::initialize(char* _bytes, int _size)
 	}
 	}
 
 
 	size = _size;
 	size = _size;
-	bytes = _bytes;
+	p = bytes = _bytes;
 
 
 	return ret;
 	return ret;
 }
 }
 
 
+void SrsStream::reset()
+{
+	p = bytes;
+}
+
+bool SrsStream::empty()
+{
+	return !p || !bytes || (p >= bytes + size);
+}
+
+bool SrsStream::require(int required_size)
+{
+	return !empty() && (required_size < bytes + size - p);
+}
+
+char SrsStream::read_char()
+{
+	srs_assert(require(1));
+	
+	return *p++;
+}
+
+int16_t SrsStream::read_2bytes()
+{
+	srs_assert(require(2));
+	
+	int16_t value;
+	pp = (char*)&value;
+	pp[1] = *p++;
+	pp[0] = *p++;
+	
+	return value;
+}
+
+std::string SrsStream::read_string(int len)
+{
+	srs_assert(require(len));
+	
+	std::string value;
+	value.append(p, len);
+	
+	p += len;
+	
+	return value;
+}
+

@@ -30,9 +30,14 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 #include <srs_core.hpp>
 #include <srs_core.hpp>
 
 
+#include <sys/types.h>
+#include <string>
+
 class SrsStream
 class SrsStream
 {
 {
 protected:
 protected:
+	char* p;
+	char* pp;
 	char* bytes;
 	char* bytes;
 	int size;
 	int size;
 public:
 public:
@@ -46,6 +51,24 @@ public:
 	* @remark, stream never free the _bytes, user must free it.
 	* @remark, stream never free the _bytes, user must free it.
 	*/
 	*/
 	virtual int initialize(char* _bytes, int _size);
 	virtual int initialize(char* _bytes, int _size);
+	/**
+	* reset the position to beginning.
+	*/
+	virtual void reset();
+	/**
+	* whether stream is empty.
+	* if empty, never read or write.
+	*/
+	virtual bool empty();
+	/**
+	* whether required size is ok.
+	* @return true if stream can read/write specified required_size bytes.
+	*/
+	virtual bool require(int required_size);
+public:
+	virtual char read_char();
+	virtual int16_t read_2bytes();
+	virtual std::string read_string(int len);
 };
 };
 
 
 #endif
 #endif