this version compiles ok

drothchild8 · Oct 23, 2012 · eacfd8f · eacfd8f
1 parent f65f07d
commit eacfd8f
Show file tree

Hide file tree

Showing 5 changed files with 380 additions and 52 deletions.
diff --git a/ngx_rtmp_netcall_module.c b/ngx_rtmp_netcall_module.c
@@ -20,12 +20,9 @@ static void ngx_rtmp_netcall_recv(ngx_event_t *rev);
 static void ngx_rtmp_netcall_send(ngx_event_t *wev);
 
 
-ngx_str_t   ngx_rtmp_netcall_content_type_urlencoded = 
-    ngx_string("application/x-www-form-urlencoded");
-
-
 typedef struct {
     ngx_msec_t                                  timeout;
+    size_t                                      bufsize;
     ngx_log_t                                  *log;
 } ngx_rtmp_netcall_app_conf_t;
 
@@ -38,11 +35,13 @@ typedef struct ngx_rtmp_netcall_session_s {
     void                                       *arg;
     ngx_rtmp_netcall_handle_pt                  handle;
     ngx_rtmp_netcall_filter_pt                  filter;
+    ngx_rtmp_netcall_sink_pt                    sink;
     ngx_chain_t                                *in;
     ngx_chain_t                                *inlast;
     ngx_chain_t                                *out;
     ngx_msec_t                                  timeout;
-    ngx_int_t                                   detached;
+    unsigned                                    detached:1;
+    size_t                                      bufsize;
 } ngx_rtmp_netcall_session_t;
 
 
@@ -60,6 +59,13 @@ static ngx_command_t  ngx_rtmp_netcall_commands[] = {
       offsetof(ngx_rtmp_netcall_app_conf_t, timeout),
       NULL },
 
+    { ngx_string("netcall_buffer"),
+      NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_CONF_TAKE1,
+      ngx_conf_set_size_slot,
+      NGX_RTMP_SRV_CONF_OFFSET,
+      offsetof(ngx_rtmp_netcall_app_conf_t, bufsize),
+      NULL },
+
       ngx_null_command
 };
 
@@ -103,6 +109,8 @@ ngx_rtmp_netcall_create_app_conf(ngx_conf_t *cf)
     }
 
     nacf->timeout = NGX_CONF_UNSET_MSEC;
+    nacf->bufsize = NGX_CONF_UNSET_SIZE;
+
     nacf->log = &cf->cycle->new_log;
 
     return nacf;
@@ -116,6 +124,7 @@ ngx_rtmp_netcall_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
     ngx_rtmp_netcall_app_conf_t *conf = child;
 
     ngx_conf_merge_msec_value(conf->timeout, prev->timeout, 10000);
+    ngx_conf_merge_size_value(conf->bufsize, prev->bufsize, 1024);
 
     return NGX_CONF_OK;
 }
@@ -219,9 +228,11 @@ ngx_rtmp_netcall_create(ngx_rtmp_session_t *s, ngx_rtmp_netcall_init_t *ci)
     }
 
     cs->timeout = cacf->timeout;
+    cs->bufsize = cacf->bufsize;
     cs->url = ci->url;
     cs->session = s;
     cs->filter = ci->filter;
+    cs->sink = ci->sink;
     cs->handle = ci->handle;
     if (cs->handle == NULL) {
         cs->detached = 1;
@@ -281,6 +292,7 @@ ngx_rtmp_netcall_close(ngx_connection_t *cc)
     ngx_pool_t                         *pool;
     ngx_rtmp_session_t                 *s;
     ngx_rtmp_netcall_ctx_t             *ctx;
+    ngx_buf_t                          *b;
 
     cs = cc->data;
 
@@ -294,16 +306,22 @@ ngx_rtmp_netcall_close(ngx_connection_t *cc)
         s = cs->session;
         ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_netcall_module);
 
+        if (cs->in && cs->sink) {
+            cs->sink(cs->session, cs->in);
+
+            b = cs->in->buf;
+            b->pos = b->last = b->start;
+
+        }
+
         for(css = &ctx->cs; *css; css = &((*css)->next)) {
             if (*css == cs) {
                 *css = cs->next;
                 break;
             }
         }
 
-        if (cs->handle &&
-                cs->handle(s, cs->arg, cs->in) != NGX_OK)
-        {
+        if (cs->handle && cs->handle(s, cs->arg, cs->in) != NGX_OK) {
             ngx_rtmp_finalize_session(s);
         }
     }
@@ -352,29 +370,43 @@ ngx_rtmp_netcall_recv(ngx_event_t *rev)
 
     for ( ;; ) {
 
-        if (cs->inlast == NULL
-                || cs->inlast->buf->last == cs->inlast->buf->end)
+        if (cs->inlast == NULL ||
+            cs->inlast->buf->last == cs->inlast->buf->end)
         {
-            cl = ngx_alloc_chain_link(cc->pool);
-            if (cl == NULL) {
-                ngx_rtmp_netcall_close(cc);
-                return;
-            }
-            cl->next = NULL;
+            if (cs->in && cs->sink) {
+                if (!cs->detached) {
+                    if (cs->sink(cs->session, cs->in) != NGX_OK) {
+                        ngx_rtmp_netcall_close(cc);
+                        return;
+                    }
+                }
 
-            cl->buf = ngx_create_temp_buf(cc->pool, 1024);
-            if (cl->buf == NULL) {
-                ngx_rtmp_netcall_close(cc);
-                return;
-            }
+                b = cs->in->buf;
+                b->pos = b->last = b->start;
 
-            if (cs->in == NULL) {
-                cs->in = cl;
             } else {
-                cs->inlast->next = cl;
-            }
+                cl = ngx_alloc_chain_link(cc->pool);
+                if (cl == NULL) {
+                    ngx_rtmp_netcall_close(cc);
+                    return;
+                }
+
+                cl->next = NULL;
+
+                cl->buf = ngx_create_temp_buf(cc->pool, cs->bufsize);
+                if (cl->buf == NULL) {
+                    ngx_rtmp_netcall_close(cc);
+                    return;
+                }
 
-            cs->inlast = cl;
+                if (cs->in == NULL) {
+                    cs->in = cl;
+                } else {
+                    cs->inlast->next = cl;
+                }
+
+                cs->inlast = cl;
+            }
         }
 
         b = cs->inlast->buf;
@@ -459,8 +491,9 @@ ngx_rtmp_netcall_send(ngx_event_t *wev)
 
 
 ngx_chain_t *
-ngx_rtmp_netcall_http_format_header(ngx_url_t *url, ngx_pool_t *pool, 
-        size_t content_length, ngx_str_t *content_type)
+ngx_rtmp_netcall_http_format_header(ngx_str_t *uri, ngx_str_t *host,
+                                   ngx_pool_t *pool, size_t content_length,
+                                   ngx_str_t *content_type)
 {
     ngx_chain_t                    *cl;
     ngx_buf_t                      *b;
@@ -480,8 +513,8 @@ ngx_rtmp_netcall_http_format_header(ngx_url_t *url, ngx_pool_t *pool,
     }
 
     b = ngx_create_temp_buf(pool, sizeof(rq_tmpl) 
-            + url->uri.len
-            + url->host.len
+            + uri->len
+            + host->len
             + content_type->len
             + 5);
 
@@ -492,7 +525,7 @@ ngx_rtmp_netcall_http_format_header(ngx_url_t *url, ngx_pool_t *pool,
     cl->buf = b;
 
     b->last = ngx_snprintf(b->last, b->end - b->last, rq_tmpl,
-            &url->uri, &url->host, content_type, content_length);
+            uri, host, content_type, content_length);
 
     return cl;
 }

diff --git a/ngx_rtmp_netcall_module.h b/ngx_rtmp_netcall_module.h
@@ -15,6 +15,8 @@
 typedef ngx_chain_t * (*ngx_rtmp_netcall_create_pt)(ngx_rtmp_session_t *s,
         void *arg, ngx_pool_t *pool);
 typedef ngx_int_t (*ngx_rtmp_netcall_filter_pt)(ngx_chain_t *in);
+typedef ngx_int_t (*ngx_rtmp_netcall_sink_pt)(ngx_rtmp_session_t *s,
+        ngx_chain_t *in);
 typedef ngx_int_t (*ngx_rtmp_netcall_handle_pt)(ngx_rtmp_session_t *s, 
         void *arg, ngx_chain_t *in);
 
@@ -32,6 +34,7 @@ typedef struct {
     ngx_url_t                      *url;
     ngx_rtmp_netcall_create_pt      create;
     ngx_rtmp_netcall_filter_pt      filter;
+    ngx_rtmp_netcall_sink_pt        sink;
     ngx_rtmp_netcall_handle_pt      handle;
     void                           *arg;
     size_t                          argsize;
@@ -41,14 +44,13 @@ typedef struct {
 ngx_int_t ngx_rtmp_netcall_create(ngx_rtmp_session_t *s, 
         ngx_rtmp_netcall_init_t *ci);
 
-extern ngx_str_t    ngx_rtmp_netcall_content_type_urlencoded;
-
 
 /* HTTP handling */
 ngx_chain_t * ngx_rtmp_netcall_http_format_session(ngx_rtmp_session_t *s, 
         ngx_pool_t *pool);
-ngx_chain_t * ngx_rtmp_netcall_http_format_header(ngx_url_t *url, 
-        ngx_pool_t *pool, size_t content_length, ngx_str_t *content_type);
+ngx_chain_t * ngx_rtmp_netcall_http_format_header(ngx_str_t *uri,
+        ngx_str_t *host, ngx_pool_t *pool, size_t content_length,
+        ngx_str_t *content_type);
 ngx_chain_t * ngx_rtmp_netcall_http_skip_header(ngx_chain_t *in);
 
 

diff --git a/ngx_rtmp_notify_module.c b/ngx_rtmp_notify_module.c
@@ -28,6 +28,10 @@ static ngx_int_t ngx_rtmp_notify_done(ngx_rtmp_session_t *s, char *cbname,
        ngx_url_t *url);
 
 
+ngx_str_t   ngx_rtmp_notify_urlencoded = 
+            ngx_string("application/x-www-form-urlencoded");
+
+
 #define NGX_RTMP_NOTIFY_PUBLISHING              0x01
 #define NGX_RTMP_NOTIFY_PLAYING                 0x02
 
@@ -190,6 +194,7 @@ ngx_rtmp_notify_publish_create(ngx_rtmp_session_t *s, void *arg,
     ngx_chain_t                    *hl, *cl, *pl;
     ngx_buf_t                      *b;
     ngx_str_t                      *addr_text;
+    ngx_url_t                      *url;
     size_t                          name_len, type_len, args_len;
 
     nacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_notify_module);
@@ -244,9 +249,10 @@ ngx_rtmp_notify_publish_create(ngx_rtmp_session_t *s, void *arg,
     }
 
     /* HTTP header */
-    hl = ngx_rtmp_netcall_http_format_header(nacf->url[NGX_RTMP_NOTIFY_PUBLISH],
+    url = nacf->url[NGX_RTMP_NOTIFY_PUBLISH];
+    hl = ngx_rtmp_netcall_http_format_header(&url->uri, &url->host,
             pool, cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
-            &ngx_rtmp_netcall_content_type_urlencoded);
+            &ngx_rtmp_notify_urlencoded);
 
     if (hl == NULL) {
         return NULL;
@@ -270,6 +276,7 @@ ngx_rtmp_notify_play_create(ngx_rtmp_session_t *s, void *arg,
     ngx_chain_t                    *hl, *cl, *pl;
     ngx_buf_t                      *b;
     ngx_str_t                      *addr_text;
+    ngx_url_t                      *url;
     size_t                          name_len, args_len;
 
     nacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_notify_module);
@@ -324,9 +331,10 @@ ngx_rtmp_notify_play_create(ngx_rtmp_session_t *s, void *arg,
     }
 
     /* HTTP header */
-    hl = ngx_rtmp_netcall_http_format_header(nacf->url[NGX_RTMP_NOTIFY_PLAY],
+    url = nacf->url[NGX_RTMP_NOTIFY_PLAY];
+    hl = ngx_rtmp_netcall_http_format_header(&url->uri, &url->host,
             pool, cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
-            &ngx_rtmp_netcall_content_type_urlencoded);
+            &ngx_rtmp_notify_urlencoded);
 
     if (hl == NULL) {
         return NULL;
@@ -401,9 +409,9 @@ ngx_rtmp_notify_done_create(ngx_rtmp_session_t *s, void *arg,
     }
 
     /* HTTP header */
-    hl = ngx_rtmp_netcall_http_format_header(ds->url, pool,
-            cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
-            &ngx_rtmp_netcall_content_type_urlencoded);
+    hl = ngx_rtmp_netcall_http_format_header(&ds->url->uri, &ds->url->host,
+            pool, cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
+            &ngx_rtmp_notify_urlencoded);
 
     if (hl == NULL) {
         return NULL;
@@ -428,6 +436,7 @@ ngx_rtmp_notify_record_done_create(ngx_rtmp_session_t *s, void *arg,
     ngx_chain_t                    *hl, *cl, *pl;
     ngx_buf_t                      *b;
     ngx_str_t                      *addr_text;
+    ngx_url_t                      *url;
     size_t                          name_len, args_len;
 
     nacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_notify_module);
@@ -489,10 +498,10 @@ ngx_rtmp_notify_record_done_create(ngx_rtmp_session_t *s, void *arg,
     }
 
     /* HTTP header */
-    hl = ngx_rtmp_netcall_http_format_header(
-            nacf->url[NGX_RTMP_NOTIFY_RECORD_DONE],
+    url = nacf->url[NGX_RTMP_NOTIFY_RECORD_DONE];
+    hl = ngx_rtmp_netcall_http_format_header(&url->uri, &url->host,
             pool, cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
-            &ngx_rtmp_netcall_content_type_urlencoded);
+            &ngx_rtmp_notify_urlencoded);
 
     if (hl == NULL) {
         return NULL;