support flv parser, add amf0 to librtmp. 0.9.110

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);
+    // amf0
-    if (amf0 == NULL) {
+    amf0_name = srs_amf0_parse(data, size, &nparsed);
-        trace("invalid amf0 data.");
+    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"
+    trace("packet type=%s, time=%d, size=%d, data-size=%d, \n"
-        "[+00, +15] %s\n[-15, EOF] %s", 
+        "offset=%d\n[+00, +15] %s\n[-15, EOF] %s\n%s%s", 
-        hbuf, tbuf);
+        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, &timestamp, &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;
             }
         }

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);
 }
 

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"

trunk/src/libs/srs_librtmp.cpp

@@ -26,6 +26,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <stdlib.h>
 
 #include <string>
+#include <sstream>
 using namespace std;
 
 #include <srs_kernel_error.hpp>
@@ -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

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);
+amf0_bool srs_amf0_is_string(srs_amf0_t amf0);
-extern amf0_bool srs_amf0_is_boolean(srs_amf0_t amf0);
+amf0_bool srs_amf0_is_boolean(srs_amf0_t amf0);
-extern amf0_bool srs_amf0_is_number(srs_amf0_t amf0);
+amf0_bool srs_amf0_is_number(srs_amf0_t amf0);
-extern amf0_bool srs_amf0_is_null(srs_amf0_t amf0);
+amf0_bool srs_amf0_is_null(srs_amf0_t amf0);
-extern amf0_bool srs_amf0_is_object(srs_amf0_t amf0);
+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_ecma_array(srs_amf0_t amf0);
 /* value converter */
-/*const char* srs_amf0_to_string(srs_amf0_t amf0);
+const char* srs_amf0_to_string(srs_amf0_t amf0);
-bool srs_amf0_to_boolean(srs_amf0_t amf0);
+amf0_bool srs_amf0_to_boolean(srs_amf0_t amf0);
-double srs_amf0_to_number(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
 }

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);

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);