Add chunk_handler

Makefile

@@ -59,7 +59,7 @@ bin/res/%.txt: res/%.*
 bin/sesimos: bin/server.o bin/logger.o bin/cache_handler.o bin/async.o bin/workers.o \
              bin/worker/request_handler.o bin/worker/tcp_acceptor.o \
              bin/worker/fastcgi_handler.o bin/worker/local_handler.o bin/worker/proxy_handler.o \
-             bin/worker/ws_frame_handler.o \
+             bin/worker/ws_frame_handler.o bin/worker/chunk_handler.o \
              bin/lib/http_static.o bin/res/default.o bin/res/proxy.o bin/res/style.o \
              bin/res/icon_error.o bin/res/icon_info.o bin/res/icon_success.o bin/res/icon_warning.o \
              bin/res/globe.o \
@@ -92,6 +92,8 @@ bin/worker/proxy_handler.o: src/worker/func.h
 
 bin/worker/ws_frame_handler.o: src/worker/func.h
 
+bin/worker/chunk_handler.o: src/worker/func.h
+
 bin/lib/compress.o: src/lib/compress.h
 
 bin/lib/config.o: src/lib/config.h src/lib/utils.h src/lib/uri.h src/logger.h

src/worker/chunk_handler.c

@@ -0,0 +1,50 @@
+/**
+ * sesimos - secure, simple, modern web server
+ * @brief FastCGI frame handler
+ * @file src/worker/fcti_frame_handler.c
+ * @author Lorenz Stechauner
+ * @date 2023-01-22
+ */
+
+#include "func.h"
+#include "../logger.h"
+#include "../workers.h"
+
+#include <errno.h>
+
+void chunk_handler_func(chunk_ctx_t *ctx) {
+    logger_set_prefix("[%*s]%s", ADDRSTRLEN, ctx->client->socket.s_addr, ctx->client->log_prefix);
+
+    char buf[CHUNK_SIZE];
+    long sent = sock_splice_chunked(&ctx->client->socket, ctx->socket, buf, sizeof(buf), ctx->flags | SOCK_SINGLE_CHUNK);
+    if (sent < 0) {
+        // error
+        error("Unable to splice chunk");
+        errno = 0;
+        ctx->err_cb(ctx->client);
+    } else if (sent == 0) {
+        // last chunk
+        ctx->client->chunks_transferred = 1;
+        ctx->next_cb(ctx);
+    } else {
+        // next chunk
+        handle_chunk(ctx);
+        return;
+    }
+
+    free(ctx);
+}
+
+int handle_chunks(client_ctx_t *ctx, sock *socket, int flags, void (*next_cb)(chunk_ctx_t *), void (*err_cb)(chunk_ctx_t *)) {
+    chunk_ctx_t *a = malloc(sizeof(chunk_ctx_t));
+
+    a->client = ctx;
+    a->socket = socket;
+    a->flags = flags;
+    a->next_cb = (void (*)(void *)) next_cb;
+    a->err_cb  = (void (*)(void *)) err_cb;
+
+    handle_chunk(a);
+
+    return 0;
+}

src/worker/func.h

@@ -19,7 +19,7 @@
 typedef struct {
     sock socket;
     int req_num;
-    unsigned char s_keep_alive:1, c_keep_alive:1, use_fastcgi:4, use_proxy:2, ws_close:2;
+    unsigned char s_keep_alive:1, c_keep_alive:1, use_fastcgi:4, use_proxy:2, ws_close:2, chunks_transferred:1;
     char cc[3], host[256];
     char req_host[256], err_msg[256];
     char log_prefix[128];
@@ -43,6 +43,14 @@ typedef struct {
     void *other;
 } ws_ctx_t;
 
+typedef struct {
+    client_ctx_t *client;
+    sock *socket;
+    int flags;
+    void (*next_cb)(void *);
+    void (*err_cb)(void *);
+} chunk_ctx_t;
+
 void tcp_acceptor_func(client_ctx_t *ctx);
 
 void request_handler_func(client_ctx_t *ctx);
@@ -55,6 +63,8 @@ void proxy_handler_func(client_ctx_t *ctx);
 
 void ws_frame_handler_func(ws_ctx_t *ctx);
 
+void chunk_handler_func(chunk_ctx_t *ctx);
+
 int respond(client_ctx_t *ctx);
 
 void request_complete(client_ctx_t *ctx);

src/workers.c

@@ -13,15 +13,16 @@
 #include "async.h"
 
 static mpmc_t tcp_acceptor_ctx, request_handler_ctx, local_handler_ctx, fastcgi_handler_ctx, proxy_handler_ctx,
-              ws_frame_handler_ctx;
+              ws_frame_handler_ctx, chunk_handler_ctx;
 
 int workers_init(void) {
-    mpmc_init(&tcp_acceptor_ctx,      8, 64, (void (*)(void *)) tcp_acceptor_func,     "tcp");
+    mpmc_init(&tcp_acceptor_ctx,          8, 64, (void (*)(void *)) tcp_acceptor_func,          "tcp");
-    mpmc_init(&request_handler_ctx,  16, 64, (void (*)(void *)) request_handler_func,  "req");
+    mpmc_init(&request_handler_ctx,       8, 64, (void (*)(void *)) request_handler_func,       "req");
-    mpmc_init(&local_handler_ctx,    16, 64, (void (*)(void *)) local_handler_func,    "local");
+    mpmc_init(&local_handler_ctx,         8, 64, (void (*)(void *)) local_handler_func,         "local");
-    mpmc_init(&fastcgi_handler_ctx,  16, 64, (void (*)(void *)) fastcgi_handler_func,  "fcgi");
+    mpmc_init(&fastcgi_handler_ctx,       8, 64, (void (*)(void *)) fastcgi_handler_func,       "fcgi");
-    mpmc_init(&proxy_handler_ctx,    16, 64, (void (*)(void *)) proxy_handler_func,    "proxy");
+    mpmc_init(&proxy_handler_ctx,         8, 64, (void (*)(void *)) proxy_handler_func,         "proxy");
-    mpmc_init(&ws_frame_handler_ctx, 16, 64, (void (*)(void *)) ws_frame_handler_func, "ws");
+    mpmc_init(&ws_frame_handler_ctx,      8, 64, (void (*)(void *)) ws_frame_handler_func,      "ws");
+    mpmc_init(&chunk_handler_ctx,         8, 64, (void (*)(void *)) chunk_handler_func,         "chunk");
     return -1;
 }
 
@@ -32,6 +33,7 @@ void workers_stop(void) {
     mpmc_stop(&proxy_handler_ctx);
     mpmc_stop(&request_handler_ctx);
     mpmc_stop(&ws_frame_handler_ctx);
+    mpmc_stop(&chunk_handler_ctx);
 }
 
 void workers_destroy(void) {
@@ -41,6 +43,7 @@ void workers_destroy(void) {
     mpmc_destroy(&proxy_handler_ctx);
     mpmc_destroy(&request_handler_ctx);
     mpmc_destroy(&ws_frame_handler_ctx);
+    mpmc_destroy(&chunk_handler_ctx);
 }
 
 int tcp_accept(client_ctx_t *ctx) {
@@ -85,3 +88,14 @@ int ws_handle_frame(ws_ctx_t *ctx) {
                  (void (*)(void *)) ws_close,
                  (void (*)(void *)) ws_close);
 }
+
+static int handle_chunk_cb(chunk_ctx_t *ctx) {
+    return mpmc_queue(&chunk_handler_ctx, ctx);
+}
+
+int handle_chunk(chunk_ctx_t *ctx) {
+    return async(ctx->socket, ASYNC_WAIT_READ, 0, ctx,
+                 (void (*)(void *)) handle_chunk_cb,
+                 (void (*)(void *)) ctx->err_cb,
+                 (void (*)(void *)) ctx->err_cb);
+}

src/workers.h

@@ -29,4 +29,6 @@ int proxy_handle(client_ctx_t *ctx);
 
 int ws_handle_frame(ws_ctx_t *ctx);
 
+int handle_chunk(chunk_ctx_t *ctx);
+
 #endif //SESIMOS_WORKERS_H