[njs] Stream: improved async callback support for s.send().

[Dmitry Volyntsev]

details:   https://hg.nginx.org/njs/rev/7a4f1f8a2cae
branches:  
changeset: 1962:7a4f1f8a2cae
user:      Dmitry Volyntsev <xeioex at nginx.com>
date:      Mon Sep 26 17:49:39 2022 -0700
description:
Stream: improved async callback support for s.send().

Previously, the "from_upstream" flag (introduced in b33aae5e8dc6) was
ignored for s.send() calls invoked directly from a body filter.
This makes the s.send() behaviour context dependent.

The fix is to always take "from_upstream" flag into account when it is
provided.

This fixed #552 issue on Github.

diffstat:

 nginx/ngx_js.h               |   4 ++++
 nginx/ngx_stream_js_module.c |  32 +++++++++++++++++---------------
 2 files changed, 21 insertions(+), 15 deletions(-)

diffs (88 lines):

diff -r ef0e05668f39 -r 7a4f1f8a2cae nginx/ngx_js.h
--- a/nginx/ngx_js.h	Thu Sep 22 19:05:36 2022 -0700
+++ b/nginx/ngx_js.h	Mon Sep 26 17:49:39 2022 -0700
@@ -22,6 +22,10 @@
 #define NGX_JS_BOOLEAN      8
 #define NGX_JS_NUMBER       16
 
+#define NGX_JS_BOOL_FALSE   0
+#define NGX_JS_BOOL_TRUE    1
+#define NGX_JS_BOOL_UNSET   2
+
 #define ngx_js_buffer_type(btype) ((btype) & ~NGX_JS_DEPRECATED)
 
 
diff -r ef0e05668f39 -r 7a4f1f8a2cae nginx/ngx_stream_js_module.c
--- a/nginx/ngx_stream_js_module.c	Thu Sep 22 19:05:36 2022 -0700
+++ b/nginx/ngx_stream_js_module.c	Mon Sep 26 17:49:39 2022 -0700
@@ -101,7 +101,7 @@ static njs_int_t ngx_stream_js_ext_on(nj
 static njs_int_t ngx_stream_js_ext_off(njs_vm_t *vm, njs_value_t *args,
      njs_uint_t nargs, njs_index_t unused);
 static njs_int_t ngx_stream_js_ext_send(njs_vm_t *vm, njs_value_t *args,
-     njs_uint_t nargs, njs_index_t unused);
+     njs_uint_t nargs, njs_index_t from_upstream);
 static njs_int_t ngx_stream_js_ext_set_return_value(njs_vm_t *vm,
     njs_value_t *args, njs_uint_t nargs, njs_index_t unused);
 
@@ -442,6 +442,7 @@ static njs_external_t  ngx_stream_js_ext
         .enumerable = 1,
         .u.method = {
             .native = ngx_stream_js_ext_send,
+            .magic8 = NGX_JS_BOOL_UNSET,
         }
     },
 
@@ -1286,7 +1287,7 @@ ngx_stream_js_ext_off(njs_vm_t *vm, njs_
 
 static njs_int_t
 ngx_stream_js_ext_send(njs_vm_t *vm, njs_value_t *args, njs_uint_t nargs,
-    njs_index_t unused)
+    njs_index_t from_upstream)
 {
     unsigned               last_buf, flush;
     njs_str_t              buffer;
@@ -1349,6 +1350,17 @@ ngx_stream_js_ext_send(njs_vm_t *vm, njs
         if (value != NULL) {
             last_buf = njs_value_bool(value);
         }
+
+        if (from_upstream == NGX_JS_BOOL_UNSET) {
+            value = njs_vm_object_prop(vm, flags, &from_key, &lvalue);
+            if (value != NULL) {
+                from_upstream = njs_value_bool(value);
+            }
+
+            if (value == NULL && ctx->buf == NULL) {
+                goto exception;
+            }
+        }
     }
 
     cl = ngx_chain_get_free_buf(c->pool, &ctx->free);
@@ -1371,23 +1383,13 @@ ngx_stream_js_ext_send(njs_vm_t *vm, njs
     b->pos = b->start;
     b->last = b->end;
 
-    if (ctx->buf != NULL) {
+    if (from_upstream == NGX_JS_BOOL_UNSET) {
         *ctx->last_out = cl;
         ctx->last_out = &cl->next;
 
     } else {
-        if (!njs_value_is_object(flags)) {
-            goto exception;
-        }
-
-        value = njs_vm_object_prop(vm, flags, &from_key, &lvalue);
-        if (value == NULL) {
-            goto exception;
-        }
-
-        if (ngx_stream_js_next_filter(s, ctx, cl, njs_value_bool(value))
-            == NGX_ERROR)
-        {
+
+        if (ngx_stream_js_next_filter(s, ctx, cl, from_upstream) == NGX_ERROR) {
             njs_vm_error(vm, "ngx_stream_js_next_filter() failed");
             return NJS_ERROR;
         }