From bd24fe7d75a22a46ed40eb5a0baffcff0ad76db2 Mon Sep 17 00:00:00 2001 From: winlin Date: Thu, 22 May 2014 15:08:25 +0800 Subject: [PATCH] support flv parser, add amf0 to librtmp. 0.9.110 --- trunk/research/librtmp/srs_flv_parser.c | 101 +++++++++++++---- trunk/src/app/srs_app_conn.cpp | 1 + trunk/src/core/srs_core.hpp | 2 +- trunk/src/libs/srs_librtmp.cpp | 139 +++++++++++++++++++++++- trunk/src/libs/srs_librtmp.hpp | 33 ++++-- trunk/src/rtmp/srs_protocol_amf0.cpp | 25 +++++ trunk/src/rtmp/srs_protocol_amf0.hpp | 3 + 7 files changed, 268 insertions(+), 36 deletions(-) diff --git a/trunk/research/librtmp/srs_flv_parser.c b/trunk/research/librtmp/srs_flv_parser.c index 53c60720f..7f0aa080b 100644 --- a/trunk/research/librtmp/srs_flv_parser.c +++ b/trunk/research/librtmp/srs_flv_parser.c @@ -78,18 +78,6 @@ int main(int argc, char** argv) return ret; } -int parse_audio_data(char* data, int size) -{ - int ret = 0; - return ret; -} - -int parse_video_data(char* data, int size) -{ - int ret = 0; - return ret; -} - void digit_to_char(char* src, int ssize, char* dst, int dsize) { int i, j; @@ -129,33 +117,96 @@ int parse_bytes(char* data, int size, char* hbuf, int hsize, char* tbuf, int tsi { memset(hbuf, 0, hsize); memset(tbuf, 0, tsize); - if (size > print_size * 2) { + + if (size > 0) { digit_to_char(data, size, hbuf, hsize - 1); + } + + if (size > print_size * 2) { digit_to_char(data + size - print_size, size, tbuf, tsize - 1); } } -int parse_script_data(char* data, int size) +#define FLV_HEADER_SIZE 11 +int parse_script_data(u_int32_t timestamp, char* data, int size, int64_t offset) { int ret = 0; char hbuf[48]; char tbuf[48]; + + int amf0_size = 0; + int nparsed = 0; + + srs_amf0_t amf0_name; + char* amf0_name_str = NULL; + + srs_amf0_t amf0_data; + char* amf0_data_str = NULL; + + // bytes parse_bytes(data, size, hbuf, sizeof(hbuf), tbuf, sizeof(tbuf), 16); - srs_amf0_t amf0 = srs_amf0_parse(data, size); - if (amf0 == NULL) { - trace("invalid amf0 data."); + // amf0 + amf0_name = srs_amf0_parse(data, size, &nparsed); + if (amf0_name == NULL || nparsed >= size) { + trace("invalid amf0 name data."); return -1; } + amf0_data = srs_amf0_parse(data + nparsed, size - nparsed, &nparsed); - trace("details:\n" - "[+00, +15] %s\n[-15, EOF] %s", - hbuf, tbuf); + trace("packet type=%s, time=%d, size=%d, data-size=%d, \n" + "offset=%d\n[+00, +15] %s\n[-15, EOF] %s\n%s%s", + srs_type2string(SRS_RTMP_TYPE_SCRIPT), timestamp, size + FLV_HEADER_SIZE, size, + (int)offset, hbuf, tbuf, + srs_amf0_human_print(amf0_name, &amf0_name_str, &amf0_size), + srs_amf0_human_print(amf0_data, &amf0_data_str, &amf0_size)); + + srs_amf0_free(amf0_name); + srs_amf0_free_bytes(amf0_name_str); + + srs_amf0_free(amf0_data); + srs_amf0_free_bytes(amf0_data_str); return ret; } +int parse_audio_data(u_int32_t timestamp, char* data, int size, int64_t offset) +{ + int ret = 0; + + char hbuf[48]; + char tbuf[48]; + + // bytes + parse_bytes(data, size, hbuf, sizeof(hbuf), tbuf, sizeof(tbuf), 16); + + trace("packet type=%s, time=%d, size=%d, data-size=%d, \n" + "offset=%d\n[+00, +15] %s\n[-15, EOF] %s\n", + srs_type2string(SRS_RTMP_TYPE_AUDIO), timestamp, size + FLV_HEADER_SIZE, size, + (int)offset, hbuf, tbuf); + + return ret; +} + +int parse_video_data(u_int32_t timestamp, char* data, int size, int64_t offset) +{ + int ret = 0; + + char hbuf[48]; + char tbuf[48]; + + // bytes + parse_bytes(data, size, hbuf, sizeof(hbuf), tbuf, sizeof(tbuf), 16); + + trace("packet type=%s, time=%d, size=%d, data-size=%d, \n" + "offset=%d\n[+00, +15] %s\n[-15, EOF] %s\n", + srs_type2string(SRS_RTMP_TYPE_VIDEO), timestamp, size + FLV_HEADER_SIZE, size, + (int)offset, hbuf, tbuf); + + return ret; +} + int parse_flv(int flv_fd) { int ret = 0; @@ -168,9 +219,12 @@ int parse_flv(int flv_fd) int type, size; u_int32_t timestamp = 0; char* data = NULL; + int64_t offset = 0; trace("start parse flv"); for (;;) { + offset = lseek(flv_fd, 0, SEEK_CUR); + if ((ret = flv_read_packet(flv_fd, &type, ×tamp, &data, &size)) != 0) { if (ret == ERROR_FLV_CODEC_EOF) { trace("parse completed."); @@ -179,19 +233,18 @@ int parse_flv(int flv_fd) trace("irtmp get packet failed. ret=%d", ret); return ret; } - trace("flv got packet: type=%s, time=%d, size=%d", srs_type2string(type), timestamp, size); // data tag if (type == SRS_RTMP_TYPE_AUDIO) { - if ((ret = parse_audio_data(data, size)) != 0) { + if ((ret = parse_audio_data(timestamp, data, size, offset)) != 0) { return ret; } } else if (type == SRS_RTMP_TYPE_VIDEO) { - if ((ret = parse_video_data(data, size)) != 0) { + if ((ret = parse_video_data(timestamp, data, size, offset)) != 0) { return ret; } } else { - if ((ret = parse_script_data(data, size)) != 0) { + if ((ret = parse_script_data(timestamp, data, size, offset)) != 0) { return ret; } } diff --git a/trunk/src/app/srs_app_conn.cpp b/trunk/src/app/srs_app_conn.cpp index 539eeb89c..35e637395 100644 --- a/trunk/src/app/srs_app_conn.cpp +++ b/trunk/src/app/srs_app_conn.cpp @@ -80,6 +80,7 @@ int SrsConnection::cycle() void SrsConnection::on_thread_stop() { + // TODO: FIXME: never remove itself, use isolate thread to do cleanup. server->remove(this); } diff --git a/trunk/src/core/srs_core.hpp b/trunk/src/core/srs_core.hpp index e1e26dcd8..57d4eae5c 100644 --- a/trunk/src/core/srs_core.hpp +++ b/trunk/src/core/srs_core.hpp @@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. // current release version #define VERSION_MAJOR "0" #define VERSION_MINOR "9" -#define VERSION_REVISION "109" +#define VERSION_REVISION "110" #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION // server info. #define RTMP_SIG_SRS_KEY "SRS" diff --git a/trunk/src/libs/srs_librtmp.cpp b/trunk/src/libs/srs_librtmp.cpp index c76c96f6a..041819ddd 100644 --- a/trunk/src/libs/srs_librtmp.cpp +++ b/trunk/src/libs/srs_librtmp.cpp @@ -26,6 +26,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. #include #include +#include using namespace std; #include @@ -408,7 +409,7 @@ int64_t srs_get_time_ms() return srs_get_system_time_ms(); } -srs_amf0_t srs_amf0_parse(char* data, int size) +srs_amf0_t srs_amf0_parse(char* data, int size, int* nparsed) { int ret = ERROR_SUCCESS; @@ -430,11 +431,23 @@ srs_amf0_t srs_amf0_parse(char* data, int size) return amf0; } + *nparsed = stream.pos(); amf0 = (srs_amf0_t)any; return amf0; } +void srs_amf0_free(srs_amf0_t amf0) +{ + SrsAmf0Any* any = (SrsAmf0Any*)amf0; + srs_freep(any); +} + +void srs_amf0_free_bytes(char* data) +{ + srs_freep(data); +} + amf0_bool srs_amf0_is_string(srs_amf0_t amf0) { SrsAmf0Any* any = (SrsAmf0Any*)amf0; @@ -471,6 +484,130 @@ amf0_bool srs_amf0_is_ecma_array(srs_amf0_t amf0) return any->is_ecma_array(); } +const char* srs_amf0_to_string(srs_amf0_t amf0) +{ + SrsAmf0Any* any = (SrsAmf0Any*)amf0; + return any->to_str_raw(); +} + +amf0_bool srs_amf0_to_boolean(srs_amf0_t amf0) +{ + SrsAmf0Any* any = (SrsAmf0Any*)amf0; + return any->to_boolean(); +} + +amf0_number srs_amf0_to_number(srs_amf0_t amf0) +{ + SrsAmf0Any* any = (SrsAmf0Any*)amf0; + return any->to_number(); +} + +int srs_amf0_object_property_count(srs_amf0_t amf0) +{ + SrsAmf0Object* obj = (SrsAmf0Object*)amf0; + return obj->count(); +} + +const char* srs_amf0_object_property_name_at(srs_amf0_t amf0, int index) +{ + SrsAmf0Object* obj = (SrsAmf0Object*)amf0; + return obj->key_raw_at(index); +} + +srs_amf0_t srs_amf0_object_property_value_at(srs_amf0_t amf0, int index) +{ + SrsAmf0Object* obj = (SrsAmf0Object*)amf0; + return (srs_amf0_t)obj->value_at(index); +} + +int srs_amf0_array_property_count(srs_amf0_t amf0) +{ + SrsAmf0EcmaArray * obj = (SrsAmf0EcmaArray*)amf0; + return obj->count(); +} + +const char* srs_amf0_array_property_name_at(srs_amf0_t amf0, int index) +{ + SrsAmf0EcmaArray* obj = (SrsAmf0EcmaArray*)amf0; + return obj->key_raw_at(index); +} + +srs_amf0_t srs_amf0_array_property_value_at(srs_amf0_t amf0, int index) +{ + SrsAmf0EcmaArray* obj = (SrsAmf0EcmaArray*)amf0; + return (srs_amf0_t)obj->value_at(index); +} + +void __srs_amf0_do_print(SrsAmf0Any* any, stringstream& ss, int& level) +{ + if (true) { + for (int i = 0; i < level; i++) { + ss << " "; + } + } + + if (any->is_boolean()) { + ss << "Boolean " << (any->to_boolean()? "true":"false") << endl; + } else if (any->is_number()) { + ss << "Number " << std::fixed << any->to_number() << endl; + } else if (any->is_string()) { + ss << "String " << any->to_str() << endl; + } else if (any->is_null()) { + ss << "Null" << endl; + } else if (any->is_ecma_array()) { + SrsAmf0EcmaArray* obj = any->to_ecma_array(); + ss << "EcmaArray " << "(" << obj->count() << " items)" << endl; + for (int i = 0; i < obj->count(); i++) { + ss << " Property '" << obj->key_at(i) << "' "; + if (obj->value_at(i)->is_object() || obj->value_at(i)->is_ecma_array()) { + int next_level = level + 1; + __srs_amf0_do_print(obj->value_at(i), ss, next_level); + } else { + int next_level = 0; + __srs_amf0_do_print(obj->value_at(i), ss, next_level); + } + } + } else if (any->is_object()) { + SrsAmf0Object* obj = any->to_object(); + ss << "Object " << "(" << obj->count() << " items)" << endl; + for (int i = 0; i < obj->count(); i++) { + ss << " Property '" << obj->key_at(i) << "' "; + if (obj->value_at(i)->is_object() || obj->value_at(i)->is_ecma_array()) { + int next_level = level + 1; + __srs_amf0_do_print(obj->value_at(i), ss, next_level); + } else { + int next_level = 0; + __srs_amf0_do_print(obj->value_at(i), ss, next_level); + } + } + } else { + ss << "Unknown" << endl; + } +} + +char* srs_amf0_human_print(srs_amf0_t amf0, char** pdata, int* psize) +{ + stringstream ss; + + ss.precision(1); + + SrsAmf0Any* any = (SrsAmf0Any*)amf0; + + int level = 0; + __srs_amf0_do_print(any, ss, level); + + string str = ss.str(); + if (str.empty()) { + return NULL; + } + + *pdata = new char[str.length()]; + *psize = str.length(); + memcpy(*pdata, str.data(), str.length()); + + return *pdata; +} + #ifdef __cplusplus } #endif diff --git a/trunk/src/libs/srs_librtmp.hpp b/trunk/src/libs/srs_librtmp.hpp index 9aa2a8bb2..e57971b0f 100644 --- a/trunk/src/libs/srs_librtmp.hpp +++ b/trunk/src/libs/srs_librtmp.hpp @@ -165,18 +165,31 @@ int64_t srs_get_time_ms(); /* the output handler. */ typedef void* srs_amf0_t; typedef int amf0_bool; -extern srs_amf0_t srs_amf0_parse(char* data, int size); +typedef double amf0_number; +srs_amf0_t srs_amf0_parse(char* data, int size, int* nparsed); +void srs_amf0_free(srs_amf0_t amf0); +void srs_amf0_free_bytes(char* data); /* type detecter */ -extern amf0_bool srs_amf0_is_string(srs_amf0_t amf0); -extern amf0_bool srs_amf0_is_boolean(srs_amf0_t amf0); -extern amf0_bool srs_amf0_is_number(srs_amf0_t amf0); -extern amf0_bool srs_amf0_is_null(srs_amf0_t amf0); -extern amf0_bool srs_amf0_is_object(srs_amf0_t amf0); -extern amf0_bool srs_amf0_is_ecma_array(srs_amf0_t amf0); +amf0_bool srs_amf0_is_string(srs_amf0_t amf0); +amf0_bool srs_amf0_is_boolean(srs_amf0_t amf0); +amf0_bool srs_amf0_is_number(srs_amf0_t amf0); +amf0_bool srs_amf0_is_null(srs_amf0_t amf0); +amf0_bool srs_amf0_is_object(srs_amf0_t amf0); +amf0_bool srs_amf0_is_ecma_array(srs_amf0_t amf0); /* value converter */ -/*const char* srs_amf0_to_string(srs_amf0_t amf0); -bool srs_amf0_to_boolean(srs_amf0_t amf0); -double srs_amf0_to_number(srs_amf0_t amf0);*/ +const char* srs_amf0_to_string(srs_amf0_t amf0); +amf0_bool srs_amf0_to_boolean(srs_amf0_t amf0); +amf0_number srs_amf0_to_number(srs_amf0_t amf0); +/* object value converter */ +int srs_amf0_object_property_count(srs_amf0_t amf0); +const char* srs_amf0_object_property_name_at(srs_amf0_t amf0, int index); +srs_amf0_t srs_amf0_object_property_value_at(srs_amf0_t amf0, int index); +/* array value converter */ +int srs_amf0_array_property_count(srs_amf0_t amf0); +const char* srs_amf0_array_property_name_at(srs_amf0_t amf0, int index); +srs_amf0_t srs_amf0_array_property_value_at(srs_amf0_t amf0, int index); +/* human readable print */ +char* srs_amf0_human_print(srs_amf0_t amf0, char** pdata, int* psize); #ifdef __cplusplus } diff --git a/trunk/src/rtmp/srs_protocol_amf0.cpp b/trunk/src/rtmp/srs_protocol_amf0.cpp index 95de4351a..6342fd6da 100644 --- a/trunk/src/rtmp/srs_protocol_amf0.cpp +++ b/trunk/src/rtmp/srs_protocol_amf0.cpp @@ -166,6 +166,7 @@ public: virtual int count(); virtual void clear(); virtual std::string key_at(int index); + virtual const char* key_raw_at(int index); virtual SrsAmf0Any* value_at(int index); virtual void set(std::string key, SrsAmf0Any* value); @@ -257,6 +258,13 @@ string SrsAmf0Any::to_str() return p->value; } +const char* SrsAmf0Any::to_str_raw() +{ + __SrsAmf0String* p = dynamic_cast<__SrsAmf0String*>(this); + srs_assert(p != NULL); + return p->value.data(); +} + bool SrsAmf0Any::to_boolean() { __SrsAmf0Boolean* p = dynamic_cast<__SrsAmf0Boolean*>(this); @@ -425,6 +433,13 @@ string __SrsUnSortedHashtable::key_at(int index) return elem.first; } +const char* __SrsUnSortedHashtable::key_raw_at(int index) +{ + srs_assert(index < count()); + SrsAmf0ObjectPropertyType& elem = properties[index]; + return elem.first.data(); +} + SrsAmf0Any* __SrsUnSortedHashtable::value_at(int index) { srs_assert(index < count()); @@ -719,6 +734,11 @@ string SrsAmf0Object::key_at(int index) return properties->key_at(index); } +const char* SrsAmf0Object::key_raw_at(int index) +{ + return properties->key_raw_at(index); +} + SrsAmf0Any* SrsAmf0Object::value_at(int index) { return properties->value_at(index); @@ -906,6 +926,11 @@ string SrsAmf0EcmaArray::key_at(int index) return properties->key_at(index); } +const char* SrsAmf0EcmaArray::key_raw_at(int index) +{ + return properties->key_raw_at(index); +} + SrsAmf0Any* SrsAmf0EcmaArray::value_at(int index) { return properties->value_at(index); diff --git a/trunk/src/rtmp/srs_protocol_amf0.hpp b/trunk/src/rtmp/srs_protocol_amf0.hpp index f9bccaefc..da2d4c06c 100644 --- a/trunk/src/rtmp/srs_protocol_amf0.hpp +++ b/trunk/src/rtmp/srs_protocol_amf0.hpp @@ -102,6 +102,7 @@ public: * user must ensure the type is a string, or assert failed. */ virtual std::string to_str(); + virtual const char* to_str_raw(); /** * get the boolean of any when is_boolean() indicates true. * user must ensure the type is a boolean, or assert failed. @@ -172,6 +173,7 @@ public: virtual int count(); // @remark: max index is count(). virtual std::string key_at(int index); + virtual const char* key_raw_at(int index); // @remark: max index is count(). virtual SrsAmf0Any* value_at(int index); @@ -212,6 +214,7 @@ public: virtual int count(); // @remark: max index is count(). virtual std::string key_at(int index); + virtual const char* key_raw_at(int index); // @remark: max index is count(). virtual SrsAmf0Any* value_at(int index);