added forcing detached netcalls when handler is NULL

ngx_rtmp_netcall_module.c

@@ -36,6 +36,7 @@ typedef struct ngx_rtmp_netcall_session_s {
     struct ngx_rtmp_netcall_session_s          *next;
     void                                       *arg;
     ngx_rtmp_netcall_handle_pt                  handle;
+    ngx_rtmp_netcall_filter_pt                  filter;
     ngx_chain_t                                *in;
     ngx_chain_t                                *inlast;
     ngx_chain_t                                *out;
@@ -208,13 +209,17 @@ ngx_rtmp_netcall_create(ngx_rtmp_session_t *s, ngx_rtmp_netcall_init_t *ci)
 
     cacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_netcall_module);
     if (cacf == NULL) {
-        return NGX_ERROR;
+        goto error;
     }
 
     cs->timeout = cacf->timeout;
     cs->url = ci->url;
     cs->session = s;
+    cs->filter = ci->filter;
     cs->handle = ci->handle;
+    if (cs->handle == NULL) {
+        cs->detached = 1;
+    }
 
     pc->log = s->connection->log;
     pc->get = ngx_rtmp_netcall_get_peer;
@@ -245,8 +250,10 @@ ngx_rtmp_netcall_create(ngx_rtmp_session_t *s, ngx_rtmp_netcall_init_t *ci)
     cc->write->handler = ngx_rtmp_netcall_send;
     cc->read->handler = ngx_rtmp_netcall_recv;
 
-    cs->next = ctx->cs;
+    if (!cs->detached) {
-    ctx->cs = cs;
+        cs->next = ctx->cs;
+        ctx->cs = cs;
+    }
 
     ngx_rtmp_netcall_send(cc->write);
 
@@ -318,6 +325,7 @@ ngx_rtmp_netcall_recv(ngx_event_t *rev)
 {
     ngx_rtmp_netcall_session_t         *cs;
     ngx_connection_t                   *cc;
+    ngx_chain_t                        *cl;
     ngx_int_t                           n;
     ngx_buf_t                          *b;
 
@@ -329,8 +337,6 @@ ngx_rtmp_netcall_recv(ngx_event_t *rev)
     }
 
     if (rev->timedout) {
-        ngx_log_error(NGX_LOG_INFO, cc->log, NGX_ETIMEDOUT, 
-                "netcall: client recv timed out");
         cc->timedout = 1;
         ngx_rtmp_netcall_close(cc);
         return;
@@ -345,22 +351,26 @@ ngx_rtmp_netcall_recv(ngx_event_t *rev)
         if (cs->inlast == NULL
                 || cs->inlast->buf->last == cs->inlast->buf->end)
         {
-            cs->inlast = ngx_alloc_chain_link(cc->pool);
+            cl = ngx_alloc_chain_link(cc->pool);
-            if (cs->inlast == NULL) {
+            if (cl == NULL) {
                 ngx_rtmp_netcall_close(cc);
                 return;
             }
+            cl->next = NULL;
 
-            cs->inlast->next = NULL;
+            cl->buf = ngx_create_temp_buf(cc->pool, 1024);
-            cs->inlast->buf = ngx_create_temp_buf(cc->pool, 1024);
+            if (cl->buf == NULL) {
-            if (cs->inlast->buf == NULL) {
                 ngx_rtmp_netcall_close(cc);
                 return;
             }
 
             if (cs->in == NULL) {
-                cs->in = cs->inlast;
+                cs->in = cl;
+            } else {
+                cs->inlast->next = cl;
             }
+
+            cs->inlast = cl;
         }
 
         b = cs->inlast->buf;
@@ -373,8 +383,15 @@ ngx_rtmp_netcall_recv(ngx_event_t *rev)
         }
 
         if (n == NGX_AGAIN) {
-            ngx_add_timer(cc->read, cs->timeout);
+            if (cs->filter && cs->in
-            if (ngx_handle_write_event(cc->read, 0) != NGX_OK) {
+                && cs->filter(cs->in) != NGX_AGAIN)
+            {
+                ngx_rtmp_netcall_close(cc);
+                return;
+            }
+
+            ngx_add_timer(rev, cs->timeout);
+            if (ngx_handle_read_event(rev, 0) != NGX_OK) {
                 ngx_rtmp_netcall_close(cc);
             }
             return;
@@ -422,8 +439,8 @@ ngx_rtmp_netcall_send(ngx_event_t *wev)
 
     /* more data to send? */
     if (cl) {
-        ngx_add_timer(cc->write, cs->timeout);
+        ngx_add_timer(wev, cs->timeout);
-        if (ngx_handle_write_event(cc->write, 0) != NGX_OK) {
+        if (ngx_handle_write_event(wev, 0) != NGX_OK) {
             ngx_rtmp_netcall_close(cc);
         }
         return;
@@ -602,7 +619,7 @@ ngx_rtmp_netcall_memcache_set(ngx_rtmp_session_t *s, ngx_pool_t *pool,
     cl->next = NULL;
     cl->buf = b;
 
-    b->last = ngx_snprintf(b->last, b->end - b->last,
+    b->last = ngx_sprintf(b->pos,
             "set %V %ui %ui %ui\r\n%V\r\n",
             key, flags, sec, (ngx_uint_t)value->len, value);
 

ngx_rtmp_netcall_module.h

@@ -14,13 +14,24 @@
 
 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_handle_pt)(ngx_rtmp_session_t *s, 
         void *arg, ngx_chain_t *in);
 
 
+/* If handle is NULL then netcall is created detached
+ * which means it's completely independent of RTMP
+ * session and its result is never visible to anyone.
+ *
+ * WARNING: It's not recommended to create non-detached
+ * netcalls from disconect handlers. Netcall disconnect
+ * handler which detaches active netcalls is executed
+ * BEFORE your handler. It leads to a crash
+ * after netcall connection is closed */
 typedef struct {
     ngx_url_t                      *url;
     ngx_rtmp_netcall_create_pt      create;
+    ngx_rtmp_netcall_filter_pt      filter;
     ngx_rtmp_netcall_handle_pt      handle;
     void                           *arg;
     size_t                          argsize;

ngx_rtmp_notify_module.c

@@ -308,6 +308,7 @@ ngx_rtmp_notify_publish(ngx_rtmp_session_t *s, ngx_rtmp_publish_t *v)
         return next_publish(s, v);
     }
 
+    ngx_memzero(&ci, sizeof(ci));
     ci.url = nacf->publish_url;
     ci.create = ngx_rtmp_notify_publish_create;
     ci.handle = ngx_rtmp_notify_publish_handle;
@@ -329,6 +330,7 @@ ngx_rtmp_notify_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
         return next_play(s, v);
     }
 
+    ngx_memzero(&ci, sizeof(ci));
     ci.url = nacf->play_url;
     ci.create = ngx_rtmp_notify_play_create;
     ci.handle = ngx_rtmp_notify_play_handle;