support decode onMetaData.

trunk/src/core/srs_core_amf0.cpp

@@ -367,8 +367,8 @@ int srs_amf0_read_utf8(SrsStream* stream, std::string& value)
 		char ch = *(str.data() + i);
 		if ((ch & 0x80) != 0) {
 			ret = ERROR_RTMP_AMF0_DECODE;
-			srs_error("only support utf8-1, 0x00-0x7F, actual is %#x. ret=%d", (int)ch, ret);
+			srs_error("ignored. only support utf8-1, 0x00-0x7F, actual is %#x. ret=%d", (int)ch, ret);
-			return ret;
+			ret = ERROR_SUCCESS;
 		}
 	}
 	

trunk/src/core/srs_core_client.cpp

@@ -163,6 +163,22 @@ int SrsClient::streaming_publish()
 
 		SrsAutoFree(SrsMessage, msg, false);
 		
+		// process onMetaData
+		if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
+			if ((ret = msg->decode_packet()) != ERROR_SUCCESS) {
+				srs_error("decode onMetaData message failed. ret=%d", ret);
+				return ret;
+			}
+		
+			SrsPacket* pkt = msg->get_packet();
+			if (dynamic_cast<SrsOnMetaDataPacket*>(pkt)) {
+				SrsOnMetaDataPacket* metadata = dynamic_cast<SrsOnMetaDataPacket*>(pkt);
+			}
+			
+			srs_trace("ignore AMF0/AMF3 data message.");
+			continue;
+		}
+		
 		// process UnPublish event.
 		if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
 			if ((ret = msg->decode_packet()) != ERROR_SUCCESS) {
@@ -177,6 +193,7 @@ int SrsClient::streaming_publish()
 			}
 			
 			srs_trace("ignore AMF0/AMF3 command message.");
+			continue;
 		}
 	}
 	

trunk/src/core/srs_core_protocol.cpp

@@ -202,6 +202,8 @@ messages.
 #define RTMP_AMF0_COMMAND_UNPUBLISH			"FCUnpublish"
 #define RTMP_AMF0_COMMAND_PUBLISH			"publish"
 #define RTMP_AMF0_DATA_SAMPLE_ACCESS		"|RtmpSampleAccess"
+#define RTMP_AMF0_DATA_SET_DATAFRAME		"@setDataFrame"
+#define RTMP_AMF0_DATA_ON_METADATA			"onMetaData"
 
 /****************************************************************************
 *****************************************************************************
@@ -861,11 +863,21 @@ bool SrsMessageHeader::is_amf0_command()
 	return message_type == RTMP_MSG_AMF0CommandMessage;
 }
 
+bool SrsMessageHeader::is_amf0_data()
+{
+	return message_type == RTMP_MSG_AMF0DataMessage;
+}
+
 bool SrsMessageHeader::is_amf3_command()
 {
 	return message_type == RTMP_MSG_AMF3CommandMessage;
 }
 
+bool SrsMessageHeader::is_amf3_data()
+{
+	return message_type == RTMP_MSG_AMF3DataMessage;
+}
+
 bool SrsMessageHeader::is_window_ackledgement_size()
 {
 	return message_type == RTMP_MSG_WindowAcknowledgementSize;
@@ -931,7 +943,7 @@ int SrsMessage::decode_packet()
 	srs_verbose("decode stream initialized success");
 	
 	// decode specified packet type
-	if (header.is_amf0_command() || header.is_amf3_command()) {
+	if (header.is_amf0_command() || header.is_amf3_command() || header.is_amf0_data() || header.is_amf3_data()) {
 		srs_verbose("start to decode AMF0/AMF3 command message.");
 		
 		// skip 1bytes to decode the amf3 command.
@@ -984,6 +996,10 @@ int SrsMessage::decode_packet()
 			srs_info("decode the AMF0/AMF3 command(unpublish message).");
 			packet = new SrsFMLEStartPacket();
 			return packet->decode(stream);
+		} else if(command == RTMP_AMF0_DATA_SET_DATAFRAME || command == RTMP_AMF0_DATA_ON_METADATA) {
+			srs_info("decode the AMF0/AMF3 data(onMetaData message).");
+			packet = new SrsOnMetaDataPacket();
+			return packet->decode(stream);
 		}
 		
 		// default packet to drop message.
@@ -1255,7 +1271,6 @@ int SrsConnectAppResPacket::encode_packet(SrsStream* stream)
 	}
 	srs_verbose("encode info success.");
 	
-	
 	srs_info("encode connect app response packet success.");
 	
 	return ret;
@@ -1881,6 +1896,81 @@ int SrsSampleAccessPacket::encode_packet(SrsStream* stream)
 	return ret;
 }
 
+SrsOnMetaDataPacket::SrsOnMetaDataPacket()
+{
+	name = RTMP_AMF0_DATA_ON_METADATA;
+	metadata = new SrsAmf0Object();
+}
+
+SrsOnMetaDataPacket::~SrsOnMetaDataPacket()
+{
+	srs_freep(metadata);
+}
+
+int SrsOnMetaDataPacket::decode(SrsStream* stream)
+{
+	int ret = ERROR_SUCCESS;
+	
+	if ((ret = srs_amf0_read_string(stream, name)) != ERROR_SUCCESS) {
+		srs_error("decode metadata name failed. ret=%d", ret);
+		return ret;
+	}
+
+	// ignore the @setDataFrame
+	if (name == RTMP_AMF0_DATA_SET_DATAFRAME) {
+		if ((ret = srs_amf0_read_string(stream, name)) != ERROR_SUCCESS) {
+			srs_error("decode metadata name failed. ret=%d", ret);
+			return ret;
+		}
+	}
+	
+	srs_verbose("decode metadata name success. name=%s", name.c_str());
+	
+	if ((ret = srs_amf0_read_object(stream, metadata)) != ERROR_SUCCESS) {
+		srs_error("decode metadata metadata failed. ret=%d", ret);
+		return ret;
+	}
+	
+	srs_info("decode metadata success");
+	
+	return ret;
+}
+
+int SrsOnMetaDataPacket::get_perfer_cid()
+{
+	return RTMP_CID_OverConnection2;
+}
+
+int SrsOnMetaDataPacket::get_message_type()
+{
+	return RTMP_MSG_AMF0DataMessage;
+}
+
+int SrsOnMetaDataPacket::get_size()
+{
+	return srs_amf0_get_string_size(name) + srs_amf0_get_object_size(metadata);
+}
+
+int SrsOnMetaDataPacket::encode_packet(SrsStream* stream)
+{
+	int ret = ERROR_SUCCESS;
+	
+	if ((ret = srs_amf0_write_string(stream, name)) != ERROR_SUCCESS) {
+		srs_error("encode name failed. ret=%d", ret);
+		return ret;
+	}
+	srs_verbose("encode name success.");
+	
+	if ((ret = srs_amf0_write_object(stream, metadata)) != ERROR_SUCCESS) {
+		srs_error("encode metadata failed. ret=%d", ret);
+		return ret;
+	}
+	srs_verbose("encode metadata success.");
+	
+	srs_info("encode onMetaData packet success.");
+	return ret;
+}
+
 SrsSetWindowAckSizePacket::SrsSetWindowAckSizePacket()
 {
 	ackowledgement_window_size = 0;

trunk/src/core/srs_core_protocol.hpp

@@ -180,7 +180,9 @@ struct SrsMessageHeader
 	virtual ~SrsMessageHeader();
 
 	bool is_amf0_command();
+	bool is_amf0_data();
 	bool is_amf3_command();
+	bool is_amf3_data();
 	bool is_window_ackledgement_size();
 	bool is_set_chunk_size();
 };
@@ -692,6 +694,37 @@ protected:
 	virtual int encode_packet(SrsStream* stream);
 };
 
+/**
+* the stream metadata.
+* FMLE: @setDataFrame
+* others: onMetaData
+*/
+class SrsOnMetaDataPacket : public SrsPacket
+{
+private:
+	typedef SrsPacket super;
+protected:
+	virtual const char* get_class_name()
+	{
+		return CLASS_NAME_STRING(SrsOnMetaDataPacket);
+	}
+public:
+	std::string name;
+	SrsAmf0Object* metadata;
+public:
+	SrsOnMetaDataPacket();
+	virtual ~SrsOnMetaDataPacket();
+public:
+	virtual int decode(SrsStream* stream);
+public:
+	virtual int get_perfer_cid();
+public:
+	virtual int get_message_type();
+protected:
+	virtual int get_size();
+	virtual int encode_packet(SrsStream* stream);
+};
+
 /**
 * 5.5. Window Acknowledgement Size (5)
 * The client or the server sends this message to inform the peer which