oclp publish and oclp play need to wait respone to decide continue or disconnect

Author: AlexWoo

http/ngx_http_flv_live_module.c

@@ -276,12 +276,6 @@ ngx_http_flv_live_write_handler(ngx_http_request_t *r)
         ngx_del_timer(wev);
     }
 
-    rc = ngx_http_flv_live_send_header(r);
-    if (rc == NGX_ERROR || rc > NGX_OK) {
-        ngx_http_finalize_request(r, rc);
-        return;
-    }
-
     ctx = ngx_http_get_module_ctx(r, ngx_http_flv_live_module);
     s = ctx->session;
 
@@ -290,6 +284,14 @@ ngx_http_flv_live_write_handler(ngx_http_request_t *r)
         return;
     }
 
+    if (s->out_chain) {
+        rc = ngx_http_flv_live_send_header(r);
+        if (rc == NGX_ERROR || rc > NGX_OK) {
+            ngx_http_finalize_request(r, rc);
+            return;
+        }
+    }
+
     while (s->out_chain) {
         present = r->connection->sent;
 

ngx_rtmp_cmd_module.c

@@ -223,8 +223,6 @@ ngx_rtmp_publish_filter(ngx_rtmp_session_t *s, ngx_rtmp_publish_t *v)
 
     s->published = 1;
 
-    ngx_rtmp_oclp_pnotify_start(s, 1);
-
     if (!s->relay) { /* relay pull */
         ngx_rtmp_cmd_stream_init(s, v->name, v->args, 1);
     }
@@ -244,8 +242,6 @@ ngx_rtmp_play_filter(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
 
     s->played = 1;
 
-    ngx_rtmp_oclp_pnotify_start(s, 0);
-
     if (!s->relay) { /* relay push */
         ngx_rtmp_cmd_stream_init(s, v->name, v->args, 0);
     }
@@ -277,8 +273,6 @@ ngx_rtmp_close_stream_filter(ngx_rtmp_session_t *s,
         ngx_live_delete_stream(&s->serverid, &s->stream);
     }
 
-    ngx_rtmp_oclp_pnotify_done(s);
-
     return ngx_rtmp_close_stream(s, v);
 }
 

ngx_rtmp_init.c

@@ -220,6 +220,9 @@ ngx_rtmp_close_session_handler(ngx_event_t *e)
 
     s = e->data;
     c = s->connection;
+    if (c) {
+        c->destroyed = 1;
+    }
 
     ngx_log_error(NGX_LOG_INFO, s->log, 0, "async close session");
 
@@ -336,9 +339,6 @@ ngx_rtmp_finalize_session(ngx_rtmp_session_t *s)
         return;
     }
 
-    if (c) {
-        c->destroyed = 1;
-    }
     e = &s->close;
     e->data = s;
     e->handler = ngx_rtmp_close_session_handler;

ngx_rtmp_oclp_module.c

@@ -8,12 +8,18 @@
 #include <ngx_config.h>
 #include <ngx_core.h>
 #include "ngx_rtmp.h"
+#include "ngx_rtmp_cmd_module.h"
 #include "ngx_live_relay.h"
 #include "ngx_dynamic_resolver.h"
 #include "ngx_toolkit_misc.h"
 #include "ngx_netcall.h"
 
 
+static ngx_rtmp_publish_pt          next_publish;
+static ngx_rtmp_play_pt             next_play;
+static ngx_rtmp_close_stream_pt     next_close_stream;
+
+
 static ngx_live_push_pt             next_push;
 static ngx_live_pull_pt             next_pull;
 static ngx_live_push_close_pt       next_push_close;
@@ -137,6 +143,9 @@ typedef struct {
     ngx_rtmp_oclp_event_t      *event;
     ngx_uint_t                  type;
     ngx_live_relay_t           *relay;
+
+    ngx_rtmp_publish_t          publish_v;
+    ngx_rtmp_play_t             play_v;
 } ngx_rtmp_oclp_ctx_t;
 
 
@@ -815,85 +824,97 @@ static void
 ngx_rtmp_oclp_pnotify_start_handle(ngx_netcall_ctx_t *nctx, ngx_int_t code)
 {
     ngx_rtmp_session_t         *s;
+    ngx_rtmp_oclp_ctx_t        *octx;
 
     s = nctx->data;
 
-    if (code != NGX_HTTP_OK) {
+    octx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_oclp_module);
+
+    if (code < NGX_HTTP_OK || code > NGX_HTTP_SPECIAL_RESPONSE) {
         ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0,
                 "oclp %s start notify error: %i",
                ngx_rtmp_oclp_app_type[nctx->type], code);
 
         if (code != -1) {
-            if (nctx->type == NGX_RTMP_OCLP_PUBLISH) {
-                ngx_rtmp_send_status(s, "NetStream.Publish.Forbidden", "status",
-                        "Publish stream Forbidden");
-            } else {
-                s->status = 403;
-                ngx_rtmp_send_status(s, "NetStream.Play.Forbidden", "status",
-                        "Play stream Forbidden");
-            }
-            ngx_rtmp_finalize_session(s);
+            goto error;
         }
 
-        return;
+        goto next;
+    }
+
+next:
+    if (octx->type == NGX_RTMP_OCLP_PUBLISH) {
+        if (next_publish(s, &octx->publish_v) != NGX_OK) {
+            goto error;
+        }
+    } else {
+        if (next_play(s, &octx->play_v) != NGX_OK) {
+            goto error;
+        }
     }
 
     ngx_rtmp_oclp_common_update_create(s, nctx);
+
+    return;
+
+error:
+    if (octx->type == NGX_RTMP_OCLP_PUBLISH) {
+        ngx_rtmp_send_status(s, "NetStream.Publish.Forbidden", "status",
+                "Publish stream Forbidden");
+    } else {
+        s->status = 403;
+        ngx_rtmp_send_status(s, "NetStream.Play.Forbidden", "status",
+                "Play stream Forbidden");
+    }
+    ngx_rtmp_finalize_session(s);
 }
 
-void
-ngx_rtmp_oclp_pnotify_start(ngx_rtmp_session_t *s, unsigned publishing)
+static ngx_int_t
+ngx_rtmp_oclp_pnotify_start(ngx_rtmp_session_t *s, ngx_uint_t type)
 {
     ngx_rtmp_oclp_app_conf_t   *oacf;
     ngx_rtmp_oclp_event_t      *event;
     ngx_rtmp_oclp_ctx_t        *ctx;
     ngx_netcall_ctx_t          *nctx;
-    ngx_uint_t                  type;
-
-    if (s->relay || s->interprocess) {
-        return;
-    }
-
-    type = publishing? NGX_RTMP_OCLP_PUBLISH: NGX_RTMP_OCLP_PLAY;
 
     oacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_oclp_module);
 
     if (oacf->events[type].nelts == 0) {
-        return;
+        return NGX_DECLINED;
     }
 
     ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_oclp_module);
     if (ctx == NULL) {
         ctx = ngx_pcalloc(s->pool, sizeof(ngx_rtmp_oclp_ctx_t));
         if (ctx == NULL) {
-            return;
+            ngx_log_error(NGX_LOG_ERR, s->log, 0, "palloc oclp ctx failed");
+            return NGX_ERROR;
         }
         ngx_rtmp_set_ctx(s, ctx, ngx_rtmp_oclp_module);
     }
 
     event = oacf->events[type].elts;
 
-    if (oacf->events[type].nelts &&
-        (event->stage & NGX_RTMP_OCLP_START) == NGX_RTMP_OCLP_START)
-    {
-        nctx = ngx_netcall_create_ctx(type, &event->groupid,
-                event->stage, event->timeout, event->update, 0);
+    nctx = ngx_netcall_create_ctx(type, &event->groupid,
+            event->stage, event->timeout, event->update, 0);
 
-        ngx_rtmp_oclp_common_url(&nctx->url, s, event, nctx,
-                                 NGX_RTMP_OCLP_START);
-        nctx->handler = ngx_rtmp_oclp_pnotify_start_handle;
-        nctx->data = s;
+    ngx_rtmp_oclp_common_url(&nctx->url, s, event, nctx,
+            NGX_RTMP_OCLP_START);
+    nctx->handler = ngx_rtmp_oclp_pnotify_start_handle;
+    nctx->data = s;
 
-        ctx->nctx = nctx;
+    ctx->nctx = nctx;
+    ctx->type = type;
 
-        ngx_log_error(NGX_LOG_INFO, s->log, 0, "oclp %s start create %V",
-                ngx_rtmp_oclp_app_type[nctx->type], &nctx->url);
+    ngx_log_error(NGX_LOG_INFO, s->log, 0, "oclp %s start create %V",
+            ngx_rtmp_oclp_app_type[nctx->type], &nctx->url);
 
-        ngx_netcall_create(nctx, s->log);
-    }
+    ngx_netcall_create(nctx, s->log);
+
+    return NGX_OK;
 }
 
-void
+static void
 ngx_rtmp_oclp_pnotify_done(ngx_rtmp_session_t *s)
 {
     ngx_rtmp_oclp_ctx_t        *ctx;
@@ -1459,6 +1480,73 @@ ngx_rtmp_oclp_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h, ngx_chain_t *in)
     return NGX_OK;
 }
 
+static ngx_int_t
+ngx_rtmp_oclp_publish(ngx_rtmp_session_t *s, ngx_rtmp_publish_t *v)
+{
+    ngx_rtmp_oclp_ctx_t        *octx;
+
+    if (s->relay || s->interprocess) {
+        goto next;
+    }
+
+    switch (ngx_rtmp_oclp_pnotify_start(s, NGX_RTMP_OCLP_PUBLISH)) {
+    case NGX_OK:
+        break;
+    case NGX_DECLINED:
+        goto next;
+    default:
+        return NGX_ERROR;
+    }
+
+    octx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_oclp_module);
+    ngx_memcpy(&octx->publish_v, v, sizeof(ngx_rtmp_publish_t));
+
+    return NGX_OK;
+
+next:
+    return next_publish(s, v);
+}
+
+static ngx_int_t
+ngx_rtmp_oclp_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
+{
+    ngx_rtmp_oclp_ctx_t        *octx;
+
+    if (s->relay || s->interprocess) {
+        goto next;
+    }
+
+    switch (ngx_rtmp_oclp_pnotify_start(s, NGX_RTMP_OCLP_PLAY)) {
+    case NGX_OK:
+        break;
+    case NGX_DECLINED:
+        goto next;
+    default:
+        return NGX_ERROR;
+    }
+
+    octx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_oclp_module);
+    ngx_memcpy(&octx->play_v, v, sizeof(ngx_rtmp_play_t));
+
+    return NGX_OK;
+
+next:
+    return next_play(s, v);
+}
+
+static ngx_int_t
+ngx_rtmp_oclp_close_stream(ngx_rtmp_session_t *s, ngx_rtmp_close_stream_t *v)
+{
+    if (s->relay || s->interprocess) {
+        goto next;
+    }
+
+    ngx_rtmp_oclp_pnotify_done(s);
+
+next:
+    return next_close_stream(s, v);
+}
+
 static ngx_int_t
 ngx_rtmp_oclp_postconfiguration(ngx_conf_t *cf)
 {
@@ -1485,5 +1573,14 @@ ngx_rtmp_oclp_postconfiguration(ngx_conf_t *cf)
     next_pull_close = ngx_live_pull_close;
     ngx_live_pull_close = ngx_rtmp_oclp_pull_close;
 
+    next_publish = ngx_rtmp_publish;
+    ngx_rtmp_publish = ngx_rtmp_oclp_publish;
+
+    next_play = ngx_rtmp_play;
+    ngx_rtmp_play = ngx_rtmp_oclp_play;
+
+    next_close_stream = ngx_rtmp_close_stream;
+    ngx_rtmp_close_stream = ngx_rtmp_oclp_close_stream;
+
     return NGX_OK;
 }

ngx_rtmp_oclp_module.h

@@ -12,8 +12,6 @@
 #include "ngx_netcall.h"
 
 
-void ngx_rtmp_oclp_pnotify_done(ngx_rtmp_session_t *s);
-void ngx_rtmp_oclp_pnotify_start(ngx_rtmp_session_t *s, unsigned publishing);
 void ngx_rtmp_oclp_stream_start(ngx_rtmp_session_t *s);
 void ngx_rtmp_oclp_stream_done(ngx_rtmp_session_t *s);