module Server { import Std::Io::{Print, PrintLine, PrintInt}; import Std::Net::{Net_Create, Net_SetReuse, Net_Bind, Net_Listen, Net_Accept, Net_Send, Net_Recv, Net_Close, Net_LastError}; import Std::String::{String_Len, String_StartsWith}; import Std::Channel::{Channel, Channel_New, Channel_Send, Channel_Recv}; import Std::Array::{Array_Drop}; import Config::{ServerConfig}; import Http::{HttpRequest, HttpResponse, Http_StatusText, Http_NewResponse, Request_WantsKeepAlive, HeaderEntry}; import Errors::{ParseResult}; import Parser::{ParseRequest}; import Router::{Router, Router_Dispatch}; extern func bux_strlen(s: String) -> uint; extern func bux_sb_new(initial_cap: uint) -> *void; extern func bux_sb_append(sb: *void, s: String); extern func bux_sb_append_int(sb: *void, n: int64); extern func bux_sb_build(sb: *void) -> String; extern func bux_sb_free(sb: *void); /// Cap requests per TCP connection (safety + fair scheduling). const MAX_KEEPALIVE_REQUESTS: int = 1000; pub struct ConnectionTask { fd: int; } pub func BuildResponse(resp: HttpResponse, keepAlive: bool) -> String { let sb: *void = bux_sb_new(4096); bux_sb_append(sb, "HTTP/1.1 "); bux_sb_append_int(sb, resp.statusCode as int64); bux_sb_append(sb, " "); bux_sb_append(sb, Http_StatusText(resp.statusCode)); bux_sb_append(sb, "\r\n"); bux_sb_append(sb, "Server: Nexus/0.3.0 (Bux)\r\n"); if bux_strlen(resp.extraHeaders) > 0 { bux_sb_append(sb, resp.extraHeaders); } if bux_strlen(resp.contentType) > 0 { bux_sb_append(sb, "Content-Type: "); bux_sb_append(sb, resp.contentType); bux_sb_append(sb, "\r\n"); } let bodyLen: uint = bux_strlen(resp.body); bux_sb_append(sb, "Content-Length: "); bux_sb_append_int(sb, bodyLen as int64); bux_sb_append(sb, "\r\n"); if keepAlive { bux_sb_append(sb, "Connection: keep-alive\r\n"); bux_sb_append(sb, "Keep-Alive: timeout=5, max=1000\r\n"); } else { bux_sb_append(sb, "Connection: close\r\n"); } bux_sb_append(sb, "\r\n"); if bodyLen > 0 { bux_sb_append(sb, resp.body); } let result: String = bux_sb_build(sb); bux_sb_free(sb); return result; } /// Serve one TCP client: zero or more HTTP requests (HTTP/1.1 keep-alive). pub func HandleConnection(fd: int, router: Router) { var reqCount: int = 0; while reqCount < MAX_KEEPALIVE_REQUESTS { let raw: String = Net_Recv(fd, 8192); if String_Len(raw) == 0 { return; } // HTTP/2 preface detection — one-shot response, then close if String_StartsWith(raw, "PRI * HTTP/2.0") { let resp: HttpResponse = Http_NewResponse(200, "text/plain; charset=utf-8", "HTTP/2 detected — full support planned for future release.\r\n"); Net_Send(fd, BuildResponse(resp, false)); return; } let parsed: ParseResult = ParseRequest(raw); var keepAlive: bool = false; if parsed.tag == ParseResult_Ok { var req: HttpRequest = parsed.data.Ok_0; keepAlive = Request_WantsKeepAlive(&req); // Last request on the connection quota must close if reqCount + 1 >= MAX_KEEPALIVE_REQUESTS { keepAlive = false; } let resp: HttpResponse = Router_Dispatch(router, req); Net_Send(fd, BuildResponse(resp, keepAlive)); // Free header buffer (moved into req at parse; no Drop on nested fields) Array_Drop(&req.headers); } else { let resp: HttpResponse = Http_NewResponse(400, "text/plain; charset=utf-8", "Bad Request"); Net_Send(fd, BuildResponse(resp, false)); return; } reqCount = reqCount + 1; if !keepAlive { return; } } } pub struct WorkerCtx { taskQueue: *Channel; router: Router; } pub func Worker(ctx: *WorkerCtx) { while true { let task: ConnectionTask = Channel_Recv(ctx.taskQueue); HandleConnection(task.fd, ctx.router); Net_Close(task.fd); } } pub struct AcceptorCtx { serverFd: int; taskQueue: *Channel; } pub func Acceptor(ctx: *AcceptorCtx) { while true { let fd: int = Net_Accept(ctx.serverFd); if fd >= 0 { let task: ConnectionTask = ConnectionTask { fd: fd }; Channel_Send(ctx.taskQueue, task); } } } pub func RunServer(config: ServerConfig, router: Router) -> int { PrintLine("================================================"); PrintLine(" Nexus HTTP Server v0.3.0"); PrintLine(" HTTP/1.1 keep-alive + thread-pool"); PrintLine(" Built with Bux"); PrintLine("================================================"); PrintLine(""); let serverFd: int = Net_Create(); if serverFd < 0 { PrintLine("FATAL: socket() failed"); return 1; } if !Net_SetReuse(serverFd) { PrintLine("WARN: SO_REUSEADDR failed"); } if !Net_Bind(serverFd, config.bindAddr, config.port) { Print("FATAL: bind failed: "); PrintLine(Net_LastError()); Net_Close(serverFd); return 1; } if !Net_Listen(serverFd, config.backlog) { PrintLine("FATAL: listen() failed"); Net_Close(serverFd); return 1; } Print("Listening on http://"); Print(config.bindAddr); Print(":"); PrintInt(config.port); PrintLine(""); PrintInt(config.workerCount); PrintLine(" worker threads | keep-alive: on | static: ./public/"); PrintLine("Endpoints: / /api/health /api/info /ws"); PrintLine("Press Ctrl+C to stop."); PrintLine(""); let taskQueue: Channel = Channel_New(config.backlog as int64); let workerCtx: WorkerCtx = WorkerCtx { taskQueue: &taskQueue, router: router }; let acceptorCtx: AcceptorCtx = AcceptorCtx { serverFd: serverFd, taskQueue: &taskQueue }; // Spawn workers (main thread will also become one) var i: int = 0; while i < config.workerCount - 1 { spawn Worker(&workerCtx); i = i + 1; } // Spawn acceptor spawn Acceptor(&acceptorCtx); // Main thread works too Worker(&workerCtx); return 0; } }