greately improved: video is now played but nginx crashed on some (big…

…?) frames
LaxWorks · Mar 16, 2012 · fd9e64c · fd9e64c
1 parent 65cc92e
commit fd9e64c
Show file tree

Hide file tree

Showing 10 changed files with 1,449 additions and 154 deletions.
diff --git a/TODO b/TODO
@@ -1,5 +1,7 @@
 - implement chain-reuse for output
 
+- move shared bufs-related code to a separate .c
+
 - get rid of greedy send_chain
 
 - remove macros hell from ngx_rtmp_send.c

diff --git a/doc/flv.html b/doc/flv.html
diff --git a/ngx_rtmp.c b/ngx_rtmp.c
@@ -272,7 +272,7 @@ ngx_rtmp_init_events(ngx_conf_t *cf, ngx_rtmp_core_main_conf_t *cmcf)
         }
     }
 
-    if (ngx_array_init(&cmcf->calls, cf->pool, 1, 
+    if (ngx_array_init(&cmcf->amf0, cf->pool, 1, 
                 sizeof(ngx_hash_key_t)) != NGX_OK)
     {
         return NGX_ERROR;
@@ -288,6 +288,42 @@ ngx_rtmp_init_events(ngx_conf_t *cf, ngx_rtmp_core_main_conf_t *cmcf)
     return NGX_OK;
 }
 
+/*
+static ngx_int_t
+ngx_rtmp_init_amf0_handler(ngx_rtmp_core_main_conf_t *cmcf, 
+        ngx_int_t id, ngx_array_t *array, ngx_hash_t *hash, 
+        ngx_rtmp_event_handler_pt handler)
+{
+    ngx_hash_init_t             calls_hash;
+    ngx_rtmp_event_handler_pt  *eh;
+    ngx_hash_key_t             *h;
+    size_t                      n;
+
+    eh = ngx_array_push(&cmcf->events[id]);
+    *eh = handler;
+
+    h = array->elts;
+    for(n = 0; n < array->nelts; ++n, ++h) {
+        h->key_hash = ngx_hash_key_lc(h->key.data, h->key.len);
+    }
+
+    calls_hash.hash = hash;
+    calls_hash.key = ngx_hash_key_lc;
+    calls_hash.max_size = 512;
+    calls_hash.bucket_size = ngx_cacheline_size;
+    calls_hash.name = "amf0_hash";
+    calls_hash.pool = cf->pool;
+    calls_hash.temp_pool = NULL;
+
+    if (ngx_hash_init(&calls_hash, array->elts, array->nelts)
+            != NGX_OK)
+    {
+        return NGX_ERROR;
+    }
+
+    return NGX_OK;
+}
+*/
 
 static ngx_int_t
 ngx_rtmp_init_event_handlers(ngx_conf_t *cf, ngx_rtmp_core_main_conf_t *cmcf)
@@ -296,41 +332,48 @@ ngx_rtmp_init_event_handlers(ngx_conf_t *cf, ngx_rtmp_core_main_conf_t *cmcf)
     ngx_rtmp_event_handler_pt  *eh;
     ngx_hash_key_t             *h;
     size_t                      n;
+
     static size_t               pm_events[] = {
         NGX_RTMP_MSG_CHUNK_SIZE,
         NGX_RTMP_MSG_ABORT,
         NGX_RTMP_MSG_ACK,
         NGX_RTMP_MSG_ACK_SIZE,
         NGX_RTMP_MSG_BANDWIDTH
     };
+    static size_t               amf0_events[] = {
+        NGX_RTMP_MSG_AMF0_META,
+        NGX_RTMP_MSG_AMF0_SHARED,
+        NGX_RTMP_MSG_AMF0_CMD
+    };
 
     /* init events */
     for(n = 0; n < sizeof(pm_events) / sizeof(pm_events[0]); ++n) {
         eh = ngx_array_push(&cmcf->events[pm_events[n]]);
         *eh = ngx_rtmp_protocol_message_handler;
     }
 
+    for(n = 0; n < sizeof(amf0_events) / sizeof(amf0_events[0]); ++n) {
+        eh = ngx_array_push(&cmcf->events[amf0_events[n]]);
+        *eh = ngx_rtmp_amf0_message_handler;
+    }
+
     eh = ngx_array_push(&cmcf->events[NGX_RTMP_MSG_USER]);
     *eh = ngx_rtmp_user_message_handler;
 
-    eh = ngx_array_push(&cmcf->events[NGX_RTMP_MSG_AMF0_CMD]);
-    *eh = ngx_rtmp_amf0_message_handler;
-
-    /* init calls */
-    h = cmcf->calls.elts;
-    for(n = 0; n < cmcf->calls.nelts; ++n, ++h) {
+    h = cmcf->amf0.elts;
+    for(n = 0; n < cmcf->amf0.nelts; ++n, ++h) {
         h->key_hash = ngx_hash_key_lc(h->key.data, h->key.len);
     }
 
-    calls_hash.hash = &cmcf->calls_hash;
+    calls_hash.hash = &cmcf->amf0_hash;
     calls_hash.key = ngx_hash_key_lc;
     calls_hash.max_size = 512;
     calls_hash.bucket_size = ngx_cacheline_size;
     calls_hash.name = "calls_hash";
     calls_hash.pool = cf->pool;
     calls_hash.temp_pool = NULL;
 
-    if (ngx_hash_init(&calls_hash, cmcf->calls.elts, cmcf->calls.nelts)
+    if (ngx_hash_init(&calls_hash, cmcf->amf0.elts, cmcf->amf0.nelts)
             != NGX_OK)
     {
         return NGX_ERROR;

diff --git a/ngx_rtmp.h b/ngx_rtmp.h
@@ -158,7 +158,7 @@ typedef struct {
 } ngx_rtmp_header_t;
 
 
-typedef struct ngx_rtmp_stream_t {
+typedef struct {
     ngx_rtmp_header_t       hdr;
     uint32_t                len;        /* current fragment length */
     ngx_chain_t            *in;
@@ -201,8 +201,6 @@ typedef struct ngx_rtmp_session_s {
 
 typedef ngx_int_t (*ngx_rtmp_event_handler_pt)(ngx_rtmp_session_t *s,
         ngx_rtmp_header_t *h, ngx_chain_t *in);
-typedef ngx_int_t (*ngx_rtmp_call_handler_pt)(ngx_rtmp_session_t *s,
-        double trans, ngx_chain_t *in);
 typedef ngx_int_t (*ngx_rtmp_disconnect_handler_pt)(ngx_rtmp_session_t *s);
 
 
@@ -211,8 +209,10 @@ typedef struct {
     ngx_array_t             listen;     /* ngx_rtmp_listen_t */
 
     ngx_array_t             events[NGX_RTMP_MSG_MAX];
-    ngx_hash_t              calls_hash;
-    ngx_array_t             calls;
+
+    ngx_hash_t              amf0_hash;
+    ngx_array_t             amf0;
+
     ngx_array_t             disconnect;
 } ngx_rtmp_core_main_conf_t;
 
@@ -275,6 +275,11 @@ typedef struct {
     ((ngx_rtmp_conf_ctx_t *) cf->ctx)->srv_conf[module.ctx_index]
 
 
+#ifdef NGX_DEBUG
+char* ngx_rtmp_message_type(uint8_t type);
+char* ngx_rtmp_user_message_type(uint16_t evt);
+#endif
+
 void ngx_rtmp_init_connection(ngx_connection_t *c);    
 void ngx_rtmp_close_connection(ngx_connection_t *c);
 u_char * ngx_rtmp_log_error(ngx_log_t *log, u_char *buf, size_t len);
@@ -291,10 +296,14 @@ ngx_int_t ngx_rtmp_amf0_message_handler(ngx_rtmp_session_t *s,
 
 /* Sending messages */
 ngx_chain_t * ngx_rtmp_alloc_shared_buf(ngx_rtmp_session_t *s);
-ngx_int_t ngx_rtmp_release_shared_buf(ngx_rtmp_session_t *s,
+ngx_chain_t * ngx_rtmp_append_shared_bufs(ngx_rtmp_session_t *s, 
+        ngx_chain_t *out, ngx_chain_t *in);
+ngx_int_t ngx_rtmp_addref_shared_bufs(ngx_chain_t *in);
+ngx_int_t ngx_rtmp_free_shared_buf(ngx_rtmp_session_t *s,
         ngx_chain_t *out);
+
 void ngx_rtmp_prepare_message(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h, 
-        ngx_chain_t *out, uint8_t fmt);
+        ngx_rtmp_header_t *lh, ngx_chain_t *out);
 ngx_int_t ngx_rtmp_send_message(ngx_rtmp_session_t *s, ngx_chain_t *out);
 
 #define NGX_RTMP_LIMIT_SOFT         0
@@ -330,8 +339,7 @@ ngx_int_t ngx_rtmp_send_user_ping_response(ngx_rtmp_session_t *s,
         uint32_t timestamp);
 
 /* AMF0 sender/receiver */
-ngx_int_t ngx_rtmp_send_amf0(ngx_rtmp_session_t *s, 
-        uint32_t csid, uint32_t msid,
+ngx_int_t ngx_rtmp_send_amf0(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
         ngx_rtmp_amf0_elt_t *elts, size_t nelts);
 ngx_int_t ngx_rtmp_receive_amf0(ngx_rtmp_session_t *s, ngx_chain_t *in, 
         ngx_rtmp_amf0_elt_t *elts, size_t nelts);

diff --git a/ngx_rtmp_amf0.c b/ngx_rtmp_amf0.c
@@ -223,7 +223,8 @@ ngx_rtmp_amf0_read_object(ngx_rtmp_amf0_ctx_t *ctx, ngx_rtmp_amf0_elt_t *elts,
 #define NGX_RTMP_AMF0_TILL_END_FLAG ((size_t)1 << (sizeof(size_t) * 8 - 1))
 
 ngx_int_t 
-ngx_rtmp_amf0_read(ngx_rtmp_amf0_ctx_t *ctx, ngx_rtmp_amf0_elt_t *elts, size_t nelts)
+ngx_rtmp_amf0_read(ngx_rtmp_amf0_ctx_t *ctx, ngx_rtmp_amf0_elt_t *elts, 
+        size_t nelts)
 {
     void                   *data;
     uint8_t                 type;