tests: cover parent stack callback dispatch

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

tests/internal/network.c

@@ -7,6 +7,8 @@
 #include <fluent-bit/flb_network.h>
 #include <fluent-bit/flb_socket.h>
 #include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_coro.h>
+#include <fluent-bit/flb_downstream.h>
 
 #include <time.h>
 #include "flb_tests_internal.h"
@@ -19,6 +21,43 @@
 #define TEST_EV_CLIENT        MK_EVENT_NOTIFICATION
 #define TEST_EV_SERVER        MK_EVENT_CUSTOM
 
+struct parent_callback_context {
+    struct flb_connection connection;
+    struct flb_coro *parent_coro;
+    int callback_on_parent;
+    int callback_result;
+    int coroutine_done;
+};
+
+static int parent_callback(void *data)
+{
+    struct parent_callback_context *context;
+
+    context = data;
+    context->callback_on_parent = flb_coro_get() == context->parent_coro;
+
+    return 73;
+}
+
+static void parent_callback_coro(void)
+{
+    struct flb_coro *coro;
+    struct parent_callback_context *context;
+
+    coro = flb_coro_get();
+    context = coro->data;
+
+    context->callback_result = flb_downstream_conn_event_call_parent(
+                                   &context->connection,
+                                   parent_callback,
+                                   context);
+    context->coroutine_done = FLB_TRUE;
+
+    while (FLB_TRUE) {
+        flb_coro_yield(coro, FLB_FALSE);
+    }
+}
+
 static int socket_check_ok(flb_sockfd_t fd)
 {
     int ret;
@@ -201,10 +240,55 @@ void test_accept_empty_nonblocking_listener()
     flb_socket_close(fd_server);
 }
 
+void test_downstream_event_callback_runs_on_parent_stack()
+{
+    size_t stack_size;
+    struct flb_coro *coro;
+    struct parent_callback_context context;
+
+    memset(&context, 0, sizeof(context));
+
+    flb_coro_thread_init();
+
+    coro = flb_coro_create(&context);
+    if (!TEST_CHECK(coro != NULL)) {
+        return;
+    }
+
+    coro->caller = co_active();
+    coro->callee = co_create(test_env_config->coro_stack_size,
+                             parent_callback_coro,
+                             &stack_size);
+    if (!TEST_CHECK(coro->callee != NULL)) {
+        flb_coro_destroy(coro);
+        return;
+    }
+
+#ifdef FLB_HAVE_VALGRIND
+    coro->valgrind_stack_id = VALGRIND_STACK_REGISTER(
+                                  coro->callee,
+                                  ((char *) coro->callee) + stack_size);
+#endif
+
+    context.parent_coro = flb_coro_get();
+    context.connection.event_coroutine = coro;
+
+    flb_downstream_conn_event_resume(&context.connection);
+
+    TEST_CHECK(context.callback_on_parent == FLB_TRUE);
+    TEST_CHECK(context.callback_result == 73);
+    TEST_CHECK(context.coroutine_done == FLB_TRUE);
+
+    context.connection.event_coroutine = NULL;
+    flb_coro_destroy(coro);
+}
+
 TEST_LIST = {
     { "ipv4_client_server", test_ipv4_client_server},
     { "ipv6_client_server", test_ipv6_client_server},
     { "ipv6_bracketed_listen", test_ipv6_bracketed_listen},
     { "accept_empty_nonblocking_listener", test_accept_empty_nonblocking_listener},
+    { "downstream_event_callback_runs_on_parent_stack",
+      test_downstream_event_callback_runs_on_parent_stack },
     { 0 }
 };