inline-method

server.c
View Options

			#include <compat.h> // IWYU pragma: keep

			#include <stdbool.h>
			#include <stdint.h>
			#include <stdio.h>
			#include <stdlib.h>
			#include <string.h>
			#include <strings.h>
			#include <signal.h>
			#include <uv.h>

			#include "ant.h"
			#include "internal.h"
			#include "ptr.h"

			#include "gc/modules.h"
			#include "net/listener.h"
			#include "streams/readable.h"

			#include "http/http1_parser.h"
			#include "http/http1_writer.h"

			#include "modules/assert.h"
			#include "modules/abort.h"
			#include "modules/blob.h"
			#include "modules/buffer.h"
			#include "modules/headers.h"
			#include "modules/request.h"
			#include "modules/response.h"
			#include "modules/server.h"
			#include "modules/timer.h"
			#include "modules/domexception.h"

			typedef struct server_runtime_s server_runtime_t;
			typedef struct server_request_s server_request_t;
			typedef struct server_conn_state_s server_conn_state_t;

			static server_runtime_t *g_server = NULL;

			typedef struct stop_waiter_s {
			ant_value_t promise;
			struct stop_waiter_s *next;
			} stop_waiter_t;

			typedef enum {
			SERVER_WRITE_NONE = 0,
			SERVER_WRITE_CLOSE_CLIENT,
			SERVER_WRITE_STREAM_READ,
			SERVER_WRITE_KEEP_ALIVE,
			} server_write_action_t;

			typedef struct {
			server_request_t *request;
			ant_conn_t *conn;
			server_write_action_t action;
			} server_write_req_t;

			struct server_request_s {
			server_runtime_t *server;
			server_conn_state_t *conn_state;
			ant_conn_t *conn;

			ant_value_t request_obj;
			ant_value_t response_obj;
			ant_value_t response_promise;
			ant_value_t response_reader;
			ant_value_t response_read_promise;

			struct server_request_s *next;
			size_t consumed_len;

			int refs;
			bool keep_alive;
			bool response_started;
			};

			struct server_conn_state_s {
			server_runtime_t *server;
			ant_conn_t *conn;

			ant_http1_conn_parser_t parser;
			uv_timer_t drain_timer;
			server_request_t request;
			server_request_t *active_req;

			bool drain_timer_closed;
			bool drain_scheduled;
			};

			struct server_runtime_s {
			ant_t *js;

			ant_value_t export_obj;
			ant_value_t fetch_fn;
			ant_value_t server_ctx;
			uv_loop_t *loop;

			ant_listener_t listener;
			uv_signal_t sigint_handle;
			uv_signal_t sigterm_handle;
			stop_waiter_t *stop_waiters;
			server_request_t *requests;

			char *hostname;
			char *unix_path;

			uint64_t request_timeout_ms;
			uint64_t idle_timeout_ms;

			int port;
			bool sigint_closed;
			bool sigterm_closed;
			bool stopping;
			bool force_stop;
			};

			static inline void server_request_retain(server_request_t *req) {
			if (req) req->refs++;
			}

			static inline server_request_t server_current_request(ant_t js) {
			return (server_request_t *)js_get_native_ptr(js->current_func);
			}

			static inline server_runtime_t server_current_runtime(ant_t js) {
			return (server_runtime_t *)js_get_native_ptr(js->current_func);
			}

			static ant_value_t server_mkreqfun(
			ant_t *js,
			ant_value_t (fn)(ant_t , ant_value_t *, int),
			server_request_t *req
			) {
			ant_value_t func = js_heavy_mkfun(js, fn, js_mkundef());
			js_set_native_ptr(func, req);
			return func;
			}

			static ant_value_t server_mkruntimefun(
			ant_t *js,
			ant_value_t (fn)(ant_t , ant_value_t *, int),
			server_runtime_t *server
			) {
			ant_value_t func = js_heavy_mkfun(js, fn, js_mkundef());
			js_set_native_ptr(func, server);
			return func;
			}

			static ant_value_t server_exception_reason(ant_t *js, ant_value_t value) {
			if (!is_err(value)) return value;
			if (!js->thrown_exists) return value;

			value = js->thrown_value;
			js->thrown_exists = false;
			js->thrown_value = js_mkundef();
			js->thrown_stack = js_mkundef();
			return value;
			}

			static void server_remove_request(server_runtime_t server, server_request_t req) {
			server_request_t **it = NULL;
			if (!server \|\| !req) return;

			for (it = &server->requests; it; it = &(it)->next) {
			if (*it == req) {
			*it = req->next;
			return;
			}}
			}

			static void server_request_reset(server_request_t *req) {
			server_runtime_t *server = NULL;
			server_conn_state_t *conn_state = NULL;

			if (!req) return;
			server = req->server;
			conn_state = req->conn_state;

			*req = (server_request_t){
			.server = server,
			.conn_state = conn_state,
			.conn = conn_state ? conn_state->conn : NULL,
			.request_obj = js_mkundef(),
			.response_obj = js_mkundef(),
			.response_promise = js_mkundef(),
			.response_reader = js_mkundef(),
			.response_read_promise = js_mkundef(),
			};
			}

			static void server_conn_state_maybe_free(server_conn_state_t *cs) {
			if (!cs) return;
			if (cs->conn) return;
			if (cs->active_req) return;
			if (cs->request.refs > 0) return;
			if (!cs->drain_timer_closed) return;
			free(cs);
			}

			static void server_request_release(server_request_t *req) {
			if (!req) return;
			if (--req->refs > 0) return;

			server_remove_request(req->server, req);
			server_request_reset(req);
			server_conn_state_maybe_free(req->conn_state);
			}

			static server_request_t server_find_request(server_runtime_t server, ant_value_t request_obj) {
			server_request_t *req = NULL;
			if (!server \|\| !is_object_type(request_obj)) return NULL;

			for (req = server->requests; req; req = req->next) {
			if (req->request_obj == request_obj) return req;
			}
			return NULL;
			}
			static void stop_waiters_resolve(server_runtime_t *server) {
			stop_waiter_t *waiter = server->stop_waiters;
			server->stop_waiters = NULL;

			while (waiter) {
			stop_waiter_t *next = waiter->next;
			js_resolve_promise(server->js, waiter->promise, js_mkundef());
			free(waiter);
			waiter = next;
			}
			}

			static void server_maybe_finish_stop(server_runtime_t *server) {
			if (!server \|\| !server->stopping) return;
			if (ant_listener_has_connections(&server->listener)) return;
			if (!ant_listener_is_closed(&server->listener) \|\| !server->sigint_closed \|\| !server->sigterm_closed) return;

			stop_waiters_resolve(server);
			}

			static void server_signal_close_cb(uv_handle_t *handle) {
			server_runtime_t server = (server_runtime_t )handle->data;
			if (!server) return;

			if (handle == (uv_handle_t *)&server->sigint_handle) server->sigint_closed = true;
			if (handle == (uv_handle_t *)&server->sigterm_handle) server->sigterm_closed = true;
			server_maybe_finish_stop(server);
			}

			static void server_begin_stop(server_runtime_t *server, bool force) {
			if (!server) return;
			if (force) server->force_stop = true;

			server->stopping = true;
			ant_listener_stop(&server->listener, server->force_stop);

			if (!uv_is_closing((uv_handle_t *)&server->sigint_handle))
			uv_close((uv_handle_t *)&server->sigint_handle, server_signal_close_cb);
			else server->sigint_closed = true;

			if (!uv_is_closing((uv_handle_t *)&server->sigterm_handle))
			uv_close((uv_handle_t *)&server->sigterm_handle, server_signal_close_cb);
			else server->sigterm_closed = true;

			server_maybe_finish_stop(server);
			}

			static void server_signal_cb(uv_signal_t *handle, int signum) {
			server_runtime_t server = (server_runtime_t )handle->data;
			server_begin_stop(server, false);
			}

			static ant_value_t server_headers_from_parsed(ant_t js, const ant_http1_parsed_request_t parsed) {
			ant_value_t headers = headers_create_empty(js);
			const ant_http_header_t *hdr = NULL;

			if (is_err(headers)) return headers;
			for (hdr = parsed->headers; hdr; hdr = hdr->next) {
			ant_value_t step = headers_append_literal(js, headers, hdr->name, hdr->value);
			if (is_err(step)) return step;
			}
			return headers;
			}

			static ant_value_t server_call_fetch(server_runtime_t *server, ant_value_t request_obj) {
			ant_t *js = server->js;
			ant_value_t args[2] = { request_obj, server->server_ctx };
			ant_value_t saved_this = js->this_val;
			ant_value_t result = js_mkundef();

			js->this_val = server->export_obj;
			if (vtype(server->fetch_fn) == T_CFUNC) result = js_as_cfunc(server->fetch_fn)(js, args, 2);
			else result = sv_vm_call(js->vm, js, server->fetch_fn, server->export_obj, args, 2, NULL, false);
			js->this_val = saved_this;

			return result;
			}

			static ant_value_t server_abort_signal_task(ant_t js, ant_value_t args, int nargs) {
			ant_value_t payload = js_get_slot(js->current_func, SLOT_DATA);
			ant_value_t signal = 0; ant_value_t reason = 0;
			if (vtype(payload) != T_OBJ) return js_mkundef();

			signal = js_get(js, payload, "signal");
			reason = js_get(js, payload, "reason");

			if (abort_signal_is_signal(signal) && !abort_signal_is_aborted(signal))
			signal_do_abort(js, signal, reason);

			return js_mkundef();
			}

			static void server_queue_abort_signal(ant_t *js, ant_value_t signal, ant_value_t reason) {
			ant_value_t callback = 0;
			ant_value_t payload = 0;

			if (!abort_signal_is_signal(signal) \|\| abort_signal_is_aborted(signal)) return;
			payload = js_mkobj(js);
			if (is_err(payload)) return;

			js_set(js, payload, "signal", signal);
			js_set(js, payload, "reason", reason);
			callback = js_heavy_mkfun(js, server_abort_signal_task, payload);

			queue_microtask(js, callback);
			}

			static void server_abort_request(server_request_t req, const char message) {
			ant_t *js = NULL;

			ant_value_t signal = 0; ant_value_t reason = 0; ant_value_t abort_reason = 0;
			if (!req \|\| !req->server \|\| !is_object_type(req->request_obj)) return;

			js = req->server->js;
			abort_reason = js_get_slot(req->request_obj, SLOT_REQUEST_ABORT_REASON);
			if (vtype(abort_reason) != T_UNDEF) return;

			reason = make_dom_exception(js,
			message ? message : "The request was aborted", "AbortError"
			);

			js_set_slot_wb(js, req->request_obj, SLOT_REQUEST_ABORT_REASON, reason);
			signal = js_get_slot(req->request_obj, SLOT_REQUEST_SIGNAL);

			if (!abort_signal_is_signal(signal) \|\| abort_signal_is_aborted(signal)) return;
			server_queue_abort_signal(js, signal, reason);
			}

			static bool server_response_chunk(server_request_t req, ant_value_t value, const uint8_t out, size_t len) {
			blob_data_t *blob = NULL;

			if (vtype(value) == T_STR) {
			out = (const uint8_t )js_getstr(req->server->js, value, len);
			if (!out) len = 0;
			return true;
			}

			if (is_object_type(value) && blob_is_blob(req->server->js, value)) {
			blob = blob_get_data(value);
			*out = blob ? blob->data : NULL;
			*len = blob ? blob->size : 0;
			return true;
			}

			return buffer_source_get_bytes(req->server->js, value, out, len);
			}

			static void server_cancel_response_body(server_request_t req, const char message) {
			ant_t *js = NULL;

			ant_value_t stream = 0;
			ant_value_t reason = 0;
			ant_value_t result = 0;

			if (!req \|\| !req->server \|\| !is_object_type(req->response_obj)) return;

			js = req->server->js;
			stream = js_get_slot(req->response_obj, SLOT_RESPONSE_BODY_STREAM);
			if (!rs_is_stream(stream)) return;

			reason = make_dom_exception(
			js, message ? message : "The response body was canceled",
			"AbortError"
			);

			result = readable_stream_cancel(js, stream, reason);
			if (vtype(result) == T_PROMISE) promise_mark_handled(result);
			}

			static void server_start_stream_read(server_request_t *req);
			static void server_on_read(ant_conn_t conn, ssize_t nread, void user_data);
			static void server_write_cb(ant_conn_t conn, int status, void user_data);

			static void server_on_deferred_drain(uv_timer_t *handle) {
			server_conn_state_t cs = handle ? (server_conn_state_t )handle->data : NULL;

			if (!cs) return;
			cs->drain_scheduled = false;

			if (!cs->conn \|\| cs->active_req) return;
			if (ant_conn_is_closing(cs->conn)) return;
			if (ant_conn_buffer_len(cs->conn) == 0) return;

			server_on_read(cs->conn, (ssize_t)ant_conn_buffer_len(cs->conn), cs->server);
			}

			static void server_on_drain_timer_close(uv_handle_t *handle) {
			server_conn_state_t cs = handle ? (server_conn_state_t )handle->data : NULL;

			if (!cs) return;
			cs->drain_timer_closed = true;
			cs->drain_scheduled = false;
			server_conn_state_maybe_free(cs);
			}

			static void server_schedule_drain(server_conn_state_t *cs) {
			if (!cs \|\| !cs->conn) return;
			if (cs->drain_scheduled) return;

			cs->drain_scheduled = true;
			if (uv_timer_start(&cs->drain_timer, server_on_deferred_drain, 0, 0) != 0)
			cs->drain_scheduled = false;
			}

			static bool server_queue_write(ant_conn_t conn, server_request_t req, char *data, size_t len, server_write_action_t action) {
			server_write_req_t wr = calloc(1, sizeof(wr));
			int rc = 0;

			if (!conn \|\| ant_conn_is_closing(conn)) {
			free(data);
			return false;
			}

			if (!wr) {
			free(data);
			return false;
			}

			wr->request = req;
			wr->conn = conn;
			wr->action = action;
			if (req) server_request_retain(req);

			rc = ant_conn_write(conn, data, len, server_write_cb, wr);
			if (rc != 0) {
			if (req) server_request_release(req);
			free(wr);
			ant_conn_close(conn);
			return false;
			}

			return true;
			}

			static bool server_queue_final_chunk(server_request_t *req, server_write_action_t action) {
			ant_http1_buffer_t buf;
			char *out = NULL;
			size_t out_len = 0;

			if (!req \|\| !req->conn \|\| ant_conn_is_closing(req->conn)) return false;

			ant_http1_buffer_init(&buf);
			if (!ant_http1_write_final_chunk(&buf)) {
			ant_http1_buffer_free(&buf);
			ant_conn_close(req->conn);
			return false;
			}

			out = ant_http1_buffer_take(&buf, &out_len);
			return server_queue_write(req->conn, req, out, out_len, action);
			}

			static void server_send_basic_response(
			ant_conn_t conn, int status, const char status_text,
			const char content_type, const uint8_t body, size_t body_len
			) {
			ant_http1_buffer_t buf;
			char *out = NULL;
			size_t out_len = 0;

			ant_http1_buffer_init(&buf);
			if (!ant_http1_write_basic_response(&buf, status, status_text, content_type, body, body_len, false)) {
			ant_http1_buffer_free(&buf);
			ant_conn_close(conn);
			return;
			}

			out = ant_http1_buffer_take(&buf, &out_len);
			server_queue_write(conn, NULL, out, out_len, SERVER_WRITE_CLOSE_CLIENT);
			}

			static inline void server_send_text_response(
			ant_conn_t *conn,
			int status,
			const char *status_text,
			const char *body
			) {
			if (!body) body = "";
			server_send_basic_response(
			conn, status,
			status_text,
			"text/plain;charset=UTF-8",
			(const uint8_t *)body,
			strlen(body)
			);
			}

			static inline void server_send_internal_error(ant_conn_t conn, const char body) {
			server_send_text_response(
			conn, 500,
			"Internal Server Error",
			body ? body : "Internal Server Error"
			);
			}

			static void server_finish_with_response(server_request_t *req, ant_value_t response_obj) {
			response_data_t *resp = response_get_data(response_obj);
			ant_value_t headers = response_get_headers(response_obj);

			ant_value_t stream = js_get_slot(response_obj, SLOT_RESPONSE_BODY_STREAM);
			bool body_is_stream = resp && resp->body_is_stream && rs_is_stream(stream);
			bool head_only = false;

			const char *status_text = NULL;
			ant_http1_buffer_t buf;

			char *out = NULL;
			size_t out_len = 0;

			if (!req->conn \|\| ant_conn_is_closing(req->conn)) return;
			if (!resp) {
			server_send_internal_error(req->conn, "Invalid Response");
			return;
			}

			req->response_obj = response_obj;
			req->response_started = true;
			head_only = strcasecmp(request_get_data(req->request_obj)->method, "HEAD") == 0;
			status_text = (resp->status_text && resp->status_text[0]) ? resp->status_text : ant_http1_default_status_text(resp->status);

			ant_http1_buffer_init(&buf);
			if (!ant_http1_write_response_head(&buf, resp->status, status_text, headers, body_is_stream, resp->body_size, req->keep_alive)) {
			ant_http1_buffer_free(&buf);
			ant_conn_close(req->conn);
			return;
			}

			if (!body_is_stream && !head_only && resp->body_size > 0)
			ant_http1_buffer_append(&buf, resp->body_data, resp->body_size);

			if (buf.failed) {
			ant_http1_buffer_free(&buf);
			ant_conn_close(req->conn);
			return;
			}

			out = ant_http1_buffer_take(&buf, &out_len);
			ant_conn_set_timeout_ms(req->conn, req->server->idle_timeout_ms);

			if (body_is_stream && head_only) server_cancel_response_body(
			req, "The response body was canceled for a HEAD request"
			);

			server_queue_write(
			req->conn,
			req, out, out_len,
			body_is_stream && !head_only
			? SERVER_WRITE_STREAM_READ
			: (req->keep_alive ? SERVER_WRITE_KEEP_ALIVE : SERVER_WRITE_CLOSE_CLIENT)
			);
			}

			static ant_value_t server_on_response_reject(ant_t js, ant_value_t args, int nargs) {
			server_request_t *req = server_current_request(js);
			ant_value_t reason = (nargs > 0) ? args[0] : js_mkundef();
			const char *msg = NULL;

			if (!req) return js_mkundef();
			req->response_promise = js_mkundef();

			if (req->conn && !ant_conn_is_closing(req->conn)) {
			msg = js_str(js, reason);
			server_send_internal_error(req->conn, msg);
			}

			server_request_release(req);
			return js_mkundef();
			}

			static ant_value_t server_on_response_fulfill(ant_t js, ant_value_t args, int nargs) {
			server_request_t *req = server_current_request(js);
			ant_value_t value = (nargs > 0) ? args[0] : js_mkundef();

			if (!req) return js_mkundef();
			req->response_promise = js_mkundef();

			if (!response_get_data(value)) {
			if (req->conn && !ant_conn_is_closing(req->conn))
			server_send_internal_error(req->conn, "fetch handler must return a Response");
			} else if (req->conn && !ant_conn_is_closing(req->conn)) {
			server_finish_with_response(req, value);
			}

			server_request_release(req);
			return js_mkundef();
			}

			static void server_handle_fetch_result(server_request_t *req, ant_value_t result) {
			ant_t *js = req->server->js;

			if (is_err(result)) {
			ant_value_t reason = server_exception_reason(js, result);
			const char *msg = js_str(js, reason);
			server_send_internal_error(req->conn, msg);
			return;
			}

			if (vtype(result) == T_PROMISE) {
			ant_value_t fulfill = server_mkreqfun(js, server_on_response_fulfill, req);
			ant_value_t reject = server_mkreqfun(js, server_on_response_reject, req);
			ant_value_t then_result = 0;

			req->response_promise = result;
			server_request_retain(req);
			then_result = js_promise_then(js, result, fulfill, reject);
			promise_mark_handled(then_result);
			return;
			}

			if (!response_get_data(result)) {
			server_send_internal_error(req->conn, "fetch handler must return a Response");
			return;
			}

			server_finish_with_response(req, result);
			}

			static ant_value_t server_stream_read_reject(ant_t js, ant_value_t args, int nargs) {
			server_request_t *req = server_current_request(js);
			if (!req) return js_mkundef();

			req->response_read_promise = js_mkundef();
			server_queue_final_chunk(req, SERVER_WRITE_CLOSE_CLIENT);
			server_request_release(req);

			return js_mkundef();
			}

			static ant_value_t server_stream_read_fulfill(ant_t js, ant_value_t args, int nargs) {
			server_request_t *req = server_current_request(js);
			ant_value_t result = (nargs > 0) ? args[0] : js_mkundef();

			ant_value_t done = 0;
			ant_value_t value = 0;

			const uint8_t *chunk = NULL;
			size_t chunk_len = 0;

			ant_http1_buffer_t buf;
			char *out = NULL;
			size_t out_len = 0;

			if (!req) return js_mkundef();
			req->response_read_promise = js_mkundef();

			if (!req->conn \|\| ant_conn_is_closing(req->conn)) {
			server_request_release(req);
			return js_mkundef();
			}

			done = js_get(js, result, "done");
			if (done == js_true) {
			ant_conn_set_timeout_ms(req->conn, req->server->idle_timeout_ms);
			server_queue_final_chunk(
			req, req->keep_alive
			? SERVER_WRITE_KEEP_ALIVE
			: SERVER_WRITE_CLOSE_CLIENT
			);

			server_request_release(req);
			return js_mkundef();
			}

			value = js_get(js, result, "value");
			if (!server_response_chunk(req, value, &chunk, &chunk_len)) {
			fprintf(stderr, "Response body stream chunk must be a string, Blob, ArrayBuffer, DataView, or TypedArray\n");
			server_cancel_response_body(req, "Invalid response body chunk");
			server_queue_final_chunk(req, SERVER_WRITE_CLOSE_CLIENT);
			server_request_release(req);
			return js_mkundef();
			}

			ant_http1_buffer_init(&buf);
			if (!ant_http1_write_chunk(&buf, chunk, chunk_len)) {
			ant_http1_buffer_free(&buf);
			ant_conn_close(req->conn);
			server_request_release(req);
			return js_mkundef();
			}

			out = ant_http1_buffer_take(&buf, &out_len);
			ant_conn_set_timeout_ms(req->conn, req->server->idle_timeout_ms);

			server_queue_write(req->conn, req, out, out_len, SERVER_WRITE_STREAM_READ);
			server_request_release(req);

			return js_mkundef();
			}

			static void server_start_stream_read(server_request_t *req) {
			ant_t *js = req->server->js;

			ant_value_t next_p = 0;
			ant_value_t fulfill = 0;
			ant_value_t reject = 0;
			ant_value_t then_result = 0;

			if (!req->conn \|\| ant_conn_is_closing(req->conn)) return;
			if (!is_object_type(req->response_reader)) return;

			next_p = rs_default_reader_read(js, req->response_reader);
			req->response_read_promise = next_p;
			fulfill = server_mkreqfun(js, server_stream_read_fulfill, req);
			reject = server_mkreqfun(js, server_stream_read_reject, req);

			server_request_retain(req);
			then_result = js_promise_then(js, next_p, fulfill, reject);
			promise_mark_handled(then_result);
			}

			static bool server_request_ensure_reader(server_request_t *req) {
			ant_t *js;
			ant_value_t reader_args[1];
			ant_value_t saved;

			if (!req \|\| !is_object_type(req->response_obj)) return false;
			if (vtype(req->response_reader) != T_UNDEF) return true;

			js = req->server->js;
			reader_args[0] = js_get_slot(req->response_obj, SLOT_RESPONSE_BODY_STREAM);

			saved = js->new_target;
			js->new_target = g_reader_proto;
			req->response_reader = js_rs_reader_ctor(js, reader_args, 1);
			js->new_target = saved;

			return !is_err(req->response_reader);
			}


			static void server_write_cb(ant_conn_t conn, int status, void user_data) {
			server_write_req_t wr = (server_write_req_t )user_data;
			server_request_t *req = wr->request;
			server_conn_state_t cs = conn ? (server_conn_state_t )ant_conn_get_user_data(conn) : NULL;

			if (status < 0 && conn && !ant_conn_is_closing(conn))
			ant_conn_close(conn);

			if (status == 0 && conn && !ant_conn_is_closing(conn)) {
			switch (wr->action) {
			case SERVER_WRITE_STREAM_READ:
			if (!server_request_ensure_reader(req)) {
			ant_conn_close(conn);
			break;
			}
			server_start_stream_read(req);
			break;

			case SERVER_WRITE_KEEP_ALIVE:
			if (cs && req) {
			ant_conn_consume(conn, req->consumed_len);
			ant_http1_conn_parser_reset(&cs->parser);
			cs->active_req = NULL;
			ant_conn_set_timeout_ms(conn, cs->server->idle_timeout_ms);
			ant_conn_resume_read(conn);
			server_request_release(req);
			if (ant_conn_buffer_len(conn) > 0) server_schedule_drain(cs);
			}
			break;

			case SERVER_WRITE_CLOSE_CLIENT:
			ant_conn_close(conn);
			break;

			default: break;
			}}

			if (req) server_request_release(req);
			free(wr);
			}

			static ant_value_t server_request_ip(ant_t js, ant_value_t args, int nargs) {
			server_runtime_t *server = server_current_runtime(js);
			server_request_t *req = NULL;
			ant_value_t out = 0;

			if (!server \|\| nargs < 1) return js_mknull();
			req = server_find_request(server, args[0]);
			if (!req \|\| !req->conn \|\| !ant_conn_has_remote_addr(req->conn)) return js_mknull();

			out = js_mkobj(js);
			js_set(js, out, "address", js_mkstr(js, ant_conn_remote_addr(req->conn), strlen(ant_conn_remote_addr(req->conn))));
			js_set(js, out, "port", js_mknum(ant_conn_remote_port(req->conn)));
			return out;
			}

			static ant_value_t server_timeout(ant_t js, ant_value_t args, int nargs) {
			server_runtime_t *server = server_current_runtime(js);
			server_request_t *req = NULL;
			int timeout = 0;

			if (!server \|\| nargs < 2) return js_mkundef();
			req = server_find_request(server, args[0]);
			if (!req \|\| !req->conn) return js_mkundef();

			timeout = (int)js_getnum(args[1]);
			ant_conn_set_timeout_ms(req->conn, (uint64_t)timeout * 1000ULL);

			return js_mkundef();
			}

			static ant_value_t server_stop(ant_t js, ant_value_t args, int nargs) {
			server_runtime_t *server = server_current_runtime(js);
			stop_waiter_t *waiter = NULL;
			ant_value_t promise = js_mkpromise(js);
			bool force = (nargs > 0 && js_truthy(js, args[0]));

			if (!server) {
			js_resolve_promise(js, promise, js_mkundef());
			return promise;
			}

			waiter = calloc(1, sizeof(*waiter));
			if (!waiter) {
			js_reject_promise(js, promise, js_mkerr(js, "out of memory"));
			return promise;
			}

			waiter->promise = promise;
			waiter->next = server->stop_waiters;
			server->stop_waiters = waiter;
			server_begin_stop(server, force);
			return promise;
			}

			static void server_process_client_request(
			ant_conn_t *conn,
			ant_http1_parsed_request_t *parsed,
			size_t consumed_len
			) {
			server_conn_state_t cs = (server_conn_state_t )ant_conn_get_user_data(conn);
			server_runtime_t *server = cs ? cs->server : NULL;
			server_request_t *req = NULL;

			ant_t *js = NULL;
			ant_value_t headers = 0;
			ant_value_t request_obj = 0;
			ant_value_t result = 0;
			bool keep_alive = false;

			if (!server \|\| !cs) {
			ant_http1_free_parsed_request(parsed);
			ant_conn_close(conn);
			return;
			}

			js = server->js;
			req = &cs->request;
			keep_alive = parsed->keep_alive;
			headers = server_headers_from_parsed(js, parsed);

			if (is_err(headers)) {
			ant_http1_free_parsed_request(parsed);
			server_send_internal_error(conn, NULL);
			return;
			}

			request_obj = request_create_server(
			js,
			parsed->method,
			parsed->target,
			parsed->absolute_target,
			parsed->host,
			server->hostname,
			server->port,
			headers,
			parsed->body,
			parsed->body_len,
			parsed->content_type
			);
			ant_http1_free_parsed_request(parsed);

			if (is_err(request_obj)) {
			server_send_internal_error(conn, NULL);
			return;
			}

			server_request_reset(req);
			req->server = server;
			req->conn_state = cs;
			req->conn = conn;
			req->request_obj = request_obj;
			req->consumed_len = consumed_len;
			req->keep_alive = keep_alive;
			req->refs = 1;
			req->next = server->requests;
			server->requests = req;
			cs->active_req = req;
			ant_conn_pause_read(conn);
			ant_conn_set_timeout_ms(conn, server->request_timeout_ms);

			result = server_call_fetch(server, request_obj);
			server_handle_fetch_result(req, result);
			}

			static void server_on_read(ant_conn_t conn, ssize_t nread, void user_data) {
			server_conn_state_t cs = (server_conn_state_t )ant_conn_get_user_data(conn);
			ant_http1_parsed_request_t parsed = {0};
			ant_http1_parse_result_t parse_result = ANT_HTTP1_PARSE_INCOMPLETE;
			size_t consumed = 0;

			if (!conn \|\| !cs) return;
			if (cs->active_req) return;
			if (ant_conn_buffer_len(conn) == 0) return;

			ant_conn_set_timeout_ms(conn, cs->server->request_timeout_ms);
			parse_result = ant_http1_conn_parser_execute(
			&cs->parser,
			ant_conn_buffer(conn),
			ant_conn_buffer_len(conn),
			&parsed,
			&consumed
			);

			if (parse_result == ANT_HTTP1_PARSE_ERROR) {
			ant_http1_free_parsed_request(&parsed);
			server_send_text_response(conn, 400, "Bad Request", "Bad Request");
			return;
			}

			if (parse_result != ANT_HTTP1_PARSE_OK) return;
			server_process_client_request(conn, &parsed, consumed);
			}

			static void server_on_end(ant_conn_t conn, void user_data) {
			if (conn) ant_conn_close(conn);
			}

			static void server_on_conn_close(ant_conn_t conn, void user_data) {
			server_runtime_t server = (server_runtime_t )user_data;
			server_conn_state_t cs = (server_conn_state_t )ant_conn_get_user_data(conn);

			if (cs) {
			ant_conn_set_user_data(conn, NULL);
			cs->conn = NULL;
			ant_http1_conn_parser_free(&cs->parser);
			if (!uv_is_closing((uv_handle_t *)&cs->drain_timer))
			uv_close((uv_handle_t *)&cs->drain_timer, server_on_drain_timer_close);
			if (cs->active_req) {
			server_abort_request(cs->active_req, "Client disconnected");
			server_cancel_response_body(cs->active_req, "Client disconnected");
			cs->active_req->conn = NULL;
			cs->active_req = NULL;
			server_request_release(&cs->request);
			cs = NULL;
			}
			if (cs) server_conn_state_maybe_free(cs);
			}

			server_maybe_finish_stop(server);
			}

			static void server_on_listener_close(ant_listener_t listener, void user_data) {
			server_maybe_finish_stop((server_runtime_t *)user_data);
			}

			static void server_on_accept(ant_listener_t listener, ant_conn_t conn, void *user_data) {
			server_runtime_t server = (server_runtime_t )user_data;
			server_conn_state_t *cs = NULL;

			(void)listener;
			if (!conn \|\| !server) return;

			cs = calloc(1, sizeof(*cs));
			if (!cs) {
			ant_conn_close(conn);
			return;
			}

			cs->server = server;
			cs->conn = conn;
			cs->drain_timer_closed = true;
			cs->request.server = server;
			cs->request.conn_state = cs;
			server_request_reset(&cs->request);

			if (uv_timer_init(server->loop, &cs->drain_timer) != 0) {
			free(cs);
			ant_conn_close(conn);
			return;
			}

			cs->drain_timer.data = cs;
			cs->drain_timer_closed = false;
			ant_http1_conn_parser_init(&cs->parser);
			ant_conn_set_user_data(conn, cs);
			ant_conn_set_no_delay(conn, true);
			}

			static bool server_export_has_fetch_handler(ant_t js, ant_value_t default_export, bool looks_like_config) {
			ant_value_t fetch = 0;

			if (looks_like_config) *looks_like_config = false;
			if (!is_object_type(default_export)) return false;

			fetch = js_get(js, default_export, "fetch");
			if (is_callable(fetch)) return true;

			if (looks_like_config) {
			ant_value_t port = js_get(js, default_export, "port");
			ant_value_t hostname = js_get(js, default_export, "hostname");
			ant_value_t idle_timeout = js_get(js, default_export, "idleTimeout");
			ant_value_t request_timeout = js_get(js, default_export, "requestTimeout");
			ant_value_t unix_path = js_get(js, default_export, "unix");
			ant_value_t tls = js_get(js, default_export, "tls");

			*looks_like_config =
			vtype(fetch) != T_UNDEF \|\|
			vtype(port) != T_UNDEF \|\|
			vtype(hostname) != T_UNDEF \|\|
			vtype(idle_timeout) != T_UNDEF \|\|
			vtype(request_timeout) != T_UNDEF \|\|
			vtype(unix_path) != T_UNDEF \|\|
			vtype(tls) != T_UNDEF;
			}

			return false;
			}

			int server_maybe_start_from_export(ant_t *js, ant_value_t default_export) {
			bool looks_like_server = false;
			ant_value_t server_result = 0;
			const char *error = NULL;

			if (!server_export_has_fetch_handler(js, default_export, &looks_like_server)) {
			if (!looks_like_server) return EXIT_SUCCESS;
			error = "Module does not export a fetch handler";
			goto fail;
			}

			server_result = server_start_from_export(js, default_export);
			if (is_err(server_result)) {
			error = js_str(js, server_result);
			goto fail;
			}

			return EXIT_SUCCESS;

			fail:
			fprintf(stderr, "%s\n", error);
			return EXIT_FAILURE;
			}

			ant_value_t server_start_from_export(ant_t *js, ant_value_t default_export) {
			server_runtime_t *server = NULL;

			ant_value_t port_v = 0;
			ant_value_t hostname_v = 0;
			ant_value_t idle_timeout_v = 0;
			ant_value_t request_timeout_v = 0;
			ant_value_t unix_v = 0;
			ant_value_t tls_v = 0;

			ant_listener_callbacks_t callbacks = {0};
			int rc = 0;

			if (g_server) return js_mkerr(js, "server is already running");
			if (!is_object_type(default_export)) return js_mkerr_typed(js, JS_ERR_TYPE, "Module does not export a fetch handler");

			server = malloc(sizeof(*server));
			if (!server) return js_mkerr(js, "out of memory");

			*server = (server_runtime_t){
			.js = js,
			.export_obj = default_export,
			.fetch_fn = js_get(js, default_export, "fetch"),
			.hostname = strdup("0.0.0.0"),
			.unix_path = NULL,
			.port = 3000,
			.request_timeout_ms = 30000,
			.idle_timeout_ms = 30000,
			.loop = uv_default_loop(),
			};

			if (!server->hostname) {
			free(server);
			return js_mkerr(js, "out of memory");
			}

			unix_v = js_get(js, default_export, "unix");
			tls_v = js_get(js, default_export, "tls");

			if (vtype(unix_v) != T_UNDEF && vtype(unix_v) != T_NULL) {
			if (vtype(unix_v) != T_STR) {
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_TYPE, "server unix must be a string");
			}

			server->unix_path = strdup(js_getstr(js, unix_v, NULL));
			if (!server->unix_path) {
			free(server->hostname);
			free(server);
			return js_mkerr(js, "out of memory");
			}
			}

			if (vtype(tls_v) != T_UNDEF && vtype(tls_v) != T_NULL) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_TYPE, "tls server config is not implemented yet");
			}

			port_v = js_get(js, default_export, "port");
			hostname_v = js_get(js, default_export, "hostname");
			idle_timeout_v = js_get(js, default_export, "idleTimeout");
			request_timeout_v = js_get(js, default_export, "requestTimeout");

			if (vtype(port_v) != T_UNDEF && vtype(port_v) != T_NULL) {
			if (vtype(port_v) != T_NUM) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_TYPE, "server port must be a number");
			}
			server->port = (int)js_getnum(port_v);
			if (server->port < 0 \|\| server->port > 65535) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_RANGE, "server port must be between 0 and 65535");
			}
			}

			if (vtype(hostname_v) != T_UNDEF && vtype(hostname_v) != T_NULL) {
			char *next_hostname = NULL;
			if (vtype(hostname_v) != T_STR) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_TYPE, "server hostname must be a string");
			}
			next_hostname = strdup(js_getstr(js, hostname_v, NULL));
			if (!next_hostname) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr(js, "out of memory");
			}
			free(server->hostname);
			server->hostname = next_hostname;
			}

			if (vtype(idle_timeout_v) != T_UNDEF && vtype(idle_timeout_v) != T_NULL) {
			double timeout = 0;
			if (vtype(idle_timeout_v) != T_NUM) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_TYPE, "server idleTimeout must be a number");
			}
			timeout = js_getnum(idle_timeout_v);
			if (timeout < 0) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_RANGE, "server idleTimeout must be >= 0");
			}
			server->idle_timeout_ms = (uint64_t)(timeout * 1000.0);
			}

			if (vtype(request_timeout_v) != T_UNDEF && vtype(request_timeout_v) != T_NULL) {
			double timeout = 0;
			if (vtype(request_timeout_v) != T_NUM) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_TYPE, "server requestTimeout must be a number");
			}

			timeout = js_getnum(request_timeout_v);
			if (timeout < 0) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_RANGE, "server requestTimeout must be >= 0");
			}

			server->request_timeout_ms = (uint64_t)(timeout * 1000.0);
			}

			uv_signal_init(server->loop, &server->sigint_handle);
			uv_signal_init(server->loop, &server->sigterm_handle);
			server->sigint_handle.data = server;
			server->sigterm_handle.data = server;

			callbacks.on_accept = server_on_accept;
			callbacks.on_read = server_on_read;
			callbacks.on_end = server_on_end;
			callbacks.on_conn_close = server_on_conn_close;
			callbacks.on_listener_close = server_on_listener_close;

			if (server->unix_path) {
			rc = ant_listener_listen_pipe(
			&server->listener, server->loop,
			server->unix_path,
			128, server->idle_timeout_ms, &callbacks, server
			);
			} else {
			rc = ant_listener_listen_tcp(
			&server->listener, server->loop,
			server->hostname, server->port,
			128, server->idle_timeout_ms, &callbacks, server
			);
			}

			if (rc != 0) {
			free(server->unix_path);
			free(server->hostname);
			free(server);
			return js_mkerr_typed(js, JS_ERR_TYPE, "%s", uv_strerror(rc));
			}

			server->port = ant_listener_port(&server->listener);
			uv_signal_start(&server->sigint_handle, server_signal_cb, SIGINT);
			uv_signal_start(&server->sigterm_handle, server_signal_cb, SIGTERM);

			server->server_ctx = js_mkobj(js);
			js_set(js, server->server_ctx, "requestIP", server_mkruntimefun(js, server_request_ip, server));
			js_set(js, server->server_ctx, "timeout", server_mkruntimefun(js, server_timeout, server));
			js_set(js, server->server_ctx, "stop", server_mkruntimefun(js, server_stop, server));

			g_server = server;
			return js_mkundef();
			}

			void gc_mark_server(ant_t *js, gc_mark_fn mark) {
			server_request_t *req = NULL;
			stop_waiter_t *waiter = NULL;

			if (!g_server) return;
			mark(js, g_server->export_obj);
			mark(js, g_server->fetch_fn);
			mark(js, g_server->server_ctx);

			for (req = g_server->requests; req; req = req->next) {
			mark(js, req->request_obj);
			mark(js, req->response_obj);
			mark(js, req->response_promise);
			mark(js, req->response_reader);
			mark(js, req->response_read_promise);
			}

			for (waiter = g_server->stop_waiters; waiter; waiter = waiter->next)
			mark(js, waiter->promise);
			}

Owner Packages

No Owners

ant/src/modules/server.c3fa05edd2fe3mir/inline-method

server.cView Options

Owner Packages

ant/src/modules/server.c
3fa05edd2fe3mir/inline-method

server.c
View Options