feat: mostly migrated to frames

Author: mookums

examples/basic/main.zig

@@ -13,6 +13,23 @@ const Server = http.Server;
 const Router = http.Router;
 const Context = http.Context;
 const Route = http.Route;
+const Middleware = http.Middleware;
+
+const Next = http.Next;
+const Response = http.Response;
+const Respond = http.Respond;
+
+const CompressMiddleware = http.CompressMiddleware;
+const RateLimitMiddleware = http.RateLimitMiddleware;
+const ThreadSafeRateLimit = http.ThreadSafeRateLimit;
+
+fn base_handler(_: Context, _: void) !Respond {
+    return .{
+        .status = .OK,
+        .mime = http.Mime.TEXT,
+        .body = "Hello, world!",
+    };
+}
 
 pub fn main() !void {
     const host: []const u8 = "0.0.0.0";
@@ -26,14 +43,20 @@ pub fn main() !void {
     // will spawn our runtimes.
     var t = try Tardy.init(allocator, .{
         .threading = .auto,
-        .size_tasks_initial = 1024,
-        .size_aio_reap_max = 1024,
+        .size_tasks_initial = 128,
+        .size_aio_reap_max = 128,
     });
     defer t.deinit();
 
-    //var router = try Router.init(allocator, &.{}, .{});
-    //defer router.deinit(allocator);
-    //router.print_route_tree();
+    var limiter = ThreadSafeRateLimit.init(allocator);
+    defer limiter.deinit();
+
+    var router = try Router.init(allocator, &.{
+        //RateLimitMiddleware(std.time.ms_per_s, &limiter),
+        //CompressMiddleware(.{ .gzip = .{} }),
+        Route.init("/").get({}, base_handler).layer(),
+    }, .{});
+    defer router.deinit(allocator);
 
     // create socket for tardy
     var socket = try Socket.init(.{ .tcp = .{ .host = host, .port = port } });
@@ -46,23 +69,20 @@ pub fn main() !void {
         socket: *Socket,
     };
 
-    const params: EntryParams = .{ .router = undefined, .socket = &socket };
+    const params: EntryParams = .{ .router = &router, .socket = &socket };
 
     // This provides the entry function into the Tardy runtime. This will run
     // exactly once inside of each runtime (each thread gets a single runtime).
     try t.entry(
         &params,
         struct {
             fn entry(rt: *Runtime, p: *const EntryParams) !void {
-                var server = Server.init(rt.allocator, .{});
-                try server.serve(rt, p.socket);
+                var server = Server.init(rt.allocator, .{
+                    .stack_size = 1024 * 1024 * 4,
+                    .socket_buffer_bytes = 1024 * 4,
+                });
+                try server.serve(rt, p.router, p.socket);
             }
         }.entry,
-        {},
-        struct {
-            fn exit(rt: *Runtime, _: void) !void {
-                _ = rt;
-            }
-        }.exit,
     );
 }

src/core/lib.zig

@@ -1,2 +1,6 @@
 pub const Job = @import("job.zig").Job;
 pub const Pseudoslice = @import("pseudoslice.zig").Pseudoslice;
+
+pub fn Pair(comptime A: type, comptime B: type) type {
+    return struct { A, B };
+}

src/http/context.zig

@@ -18,165 +18,15 @@ const MiddlewareWithData = @import("router/middleware.zig").MiddlewareWithData;
 const Next = @import("router/middleware.zig").Next;
 
 const Runtime = @import("tardy").Runtime;
-const TaskFn = @import("tardy").TaskFn;
-
 const Server = @import("server.zig").Server;
 
-const raw_respond = @import("server.zig").raw_respond;
-
 // Context is dependent on the server that gets created.
 pub const Context = struct {
     const Self = @This();
     allocator: std.mem.Allocator,
     runtime: *Runtime,
     /// The Request that triggered this handler.
     request: *const Request,
-    /// The Response that will be returned.
-    response: *Response,
-    captures: []Capture,
-    queries: *QueryMap,
-    provision: *Provision,
-    next: Next,
-    triggered: bool = false,
-
-    pub fn to_sse(self: *Self, then: TaskFn(bool, *SSE)) !void {
-        const sse = try self.allocator.create(SSE);
-        sse.* = .{
-            .context = self,
-            .runtime = self.runtime,
-            .allocator = self.allocator,
-        };
-
-        try self.respond_headers_only(
-            .{
-                .status = .OK,
-                .mime = Mime.generate(
-                    "text/event-stream",
-                    "sse",
-                    "Server-Sent Events",
-                ),
-            },
-            null,
-            sse,
-            then,
-        );
-    }
-
-    pub fn close(self: *Self) !void {
-        self.provision.job = .close;
-        try self.runtime.net.close(
-            self.provision,
-            Server.close_task,
-            self.provision.socket,
-        );
-    }
-
-    pub fn send_then(
-        self: *Self,
-        data: []const u8,
-        ctx: anytype,
-        then: TaskFn(bool, @TypeOf(ctx)),
-    ) !void {
-        const pslice = Pseudoslice.init(data, "", self.provision.buffer);
-
-        const first_chunk = try Server.prepare_send(
-            self.runtime,
-            self.provision,
-            .{
-                .other = .{
-                    .func = then,
-                    .ctx = ctx,
-                },
-            },
-            pslice,
-        );
-
-        try self.runtime.net.send(
-            self.provision,
-            Server.send_then_other_task,
-            self.provision.socket,
-            first_chunk,
-        );
-    }
-
-    pub fn send_then_recv(self: *Self, data: []const u8) !void {
-        const pslice = Pseudoslice.init(data, "", self.provision.buffer);
-
-        const first_chunk = try Server.prepare_send(
-            self.runtime,
-            self.provision,
-            .recv,
-            pslice,
-        );
-
-        try self.runtime.net.send(
-            self.provision,
-            Server.send_then_recv_task,
-            self.provision.socket,
-            first_chunk,
-        );
-    }
-
-    // This will respond with the headers only.
-    // You will be in charge of sending the body.
-    pub fn respond_headers_only(
-        self: *Self,
-        options: ResponseSetOptions,
-        content_length: ?usize,
-        ctx: anytype,
-        then: TaskFn(bool, @TypeOf(ctx)),
-    ) !void {
-        assert(!self.triggered);
-        self.triggered = true;
-
-        // the body should not be set.
-        assert(options.body == null);
-        self.response.set(options);
-
-        const headers = try self.provision.response.headers_into_buffer(
-            self.provision.buffer,
-            content_length,
-        );
-
-        try self.send_then(headers, ctx, then);
-    }
-
-    pub fn respond_without_middleware(self: *Self) !void {
-        const body = self.response.body orelse "";
-        const headers = try self.provision.response.headers_into_buffer(
-            self.provision.buffer,
-            @intCast(body.len),
-        );
-        const pslice = Pseudoslice.init(headers, body, self.provision.buffer);
-
-        const first_chunk = try Server.prepare_send(
-            self.runtime,
-            self.provision,
-            .recv,
-            pslice,
-        );
-
-        try self.runtime.net.send(
-            self.provision,
-            Server.send_then_recv_task,
-            self.provision.socket,
-            first_chunk,
-        );
-    }
-
-    /// This is your standard response.
-    pub fn respond(self: *Self, options: ResponseSetOptions) !void {
-        assert(!self.triggered);
-        self.triggered = true;
-        self.response.set(options);
-
-        // If we have a post chain, iterate through it.
-        if (self.next.post_chain.len > 0) {
-            self.next.stage = .post;
-            try self.next.run();
-            return;
-        }
-
-        try self.respond_without_middleware();
-    }
+    /// Address for this Request/Response.
+    address: std.net.Address,
 };

src/http/lib.zig

@@ -2,20 +2,26 @@ pub const Status = @import("status.zig").Status;
 pub const Method = @import("method.zig").Method;
 pub const Request = @import("request.zig").Request;
 pub const Response = @import("response.zig").Response;
+pub const Respond = @import("response.zig").Respond;
 pub const Mime = @import("mime.zig").Mime;
 pub const Encoding = @import("encoding.zig").Encoding;
 pub const Date = @import("date.zig").Date;
 pub const Headers = @import("../core/case_string_map.zig").CaseStringMap([]const u8);
 
+pub const Context = @import("context.zig").Context;
+
 pub const Router = @import("router.zig").Router;
 pub const Route = @import("router/route.zig").Route;
-pub const Layer = @import("router/layer.zig").Layer;
+pub const SSE = @import("sse.zig").SSE;
 
-pub const Context = @import("context.zig").Context;
+pub const Layer = @import("router/middleware.zig").Layer;
 pub const Middleware = @import("router/middleware.zig").Middleware;
 pub const MiddlewareFn = @import("router/middleware.zig").MiddlewareFn;
 pub const Next = @import("router/middleware.zig").Next;
-pub const SSE = @import("sse.zig").SSE;
+
+pub const ThreadSafeRateLimit = @import("middlewares/lib.zig").ThreadSafeRateLimit;
+pub const RateLimitMiddleware = @import("middlewares/lib.zig").RateLimitMiddleware;
+pub const CompressMiddleware = @import("middlewares/lib.zig").CompressMiddleware;
 
 pub const FsDir = @import("router/fs_dir.zig").FsDir;
 

src/http/middlewares/lib.zig

@@ -0,0 +1,113 @@
+const std = @import("std");
+
+const Mime = @import("../mime.zig").Mime;
+const Respond = @import("../response.zig").Respond;
+
+const Middleware = @import("../router/middleware.zig").Middleware;
+const Next = @import("../router/middleware.zig").Next;
+const Layer = @import("../router/middleware.zig").Layer;
+const TypedMiddlewareFn = @import("../router/middleware.zig").TypedMiddlewareFn;
+
+pub const ThreadSafeRateLimit = struct {
+    map: std.AutoHashMap(u128, i64),
+    mutex: std.Thread.Mutex = .{},
+
+    pub fn init(allocator: std.mem.Allocator) ThreadSafeRateLimit {
+        return .{
+            .map = std.AutoHashMap(u128, i64).init(allocator),
+            .mutex = .{},
+        };
+    }
+
+    pub fn deinit(self: *ThreadSafeRateLimit) void {
+        self.map.deinit();
+    }
+};
+
+fn get_ip(addr: std.net.Address) u128 {
+    return switch (addr.any.family) {
+        std.posix.AF.INET => @intCast(addr.in.sa.addr),
+        std.posix.AF.INET6 => std.mem.bytesAsValue(u128, &addr.in6.sa.addr[0]).*,
+        else => unreachable,
+    };
+}
+
+// TODO: Do a proper rate-limiter. This is super crude.
+pub fn RateLimitMiddleware(comptime ms_per_request: usize, limiter: *ThreadSafeRateLimit) Layer {
+    const func: TypedMiddlewareFn(*ThreadSafeRateLimit) = struct {
+        fn rate_limit_mw(next: *Next, limit: *ThreadSafeRateLimit) !Respond {
+            const ip = get_ip(next.context.address);
+            const time = std.time.milliTimestamp();
+
+            limit.mutex.lock();
+            const addr_entry = try limit.map.getOrPut(ip);
+            const last_time = addr_entry.value_ptr.*;
+            addr_entry.value_ptr.* = time;
+            limit.mutex.unlock();
+
+            if (addr_entry.found_existing) {
+                if (time - last_time >= ms_per_request) return try next.run();
+                return Respond{
+                    .status = .@"Too Many Requests",
+                    .mime = Mime.TEXT,
+                    .body = "",
+                };
+            } else return try next.run();
+        }
+    }.rate_limit_mw;
+
+    return Middleware.init(limiter, func).layer();
+}
+
+fn rate_limit_mw(next: *Next, limiter: *ThreadSafeRateLimit) !Respond {
+    const ip = get_ip(next.context.address);
+    const time = std.time.milliTimestamp();
+
+    limiter.mutex.lock();
+    const addr_entry = try limiter.map.getOrPut(ip);
+    const last_time = addr_entry.value_ptr.*;
+    addr_entry.value_ptr.* = time;
+    limiter.mutex.unlock();
+
+    if (addr_entry.found_existing) {
+        if (time - last_time >= std.time.ms_per_s) return try next.run();
+        return Respond{
+            .status = .@"Too Many Requests",
+            .mime = Mime.TEXT,
+            .body = "",
+        };
+    } else return try next.run();
+}
+
+// TODO: Consider using a C dependency here.
+// Might be nice to get some fast general compression such as zstd :)
+const Compression = union(enum) {
+    gzip: std.compress.gzip.Options,
+};
+
+pub fn CompressMiddleware(comptime compression: Compression) Layer {
+    const func: TypedMiddlewareFn(void) = switch (compression) {
+        .gzip => |inner| struct {
+            fn gzip_mw(next: *Next, _: void) !Respond {
+                var respond = try next.run();
+
+                var compressed = std.ArrayList(u8).init(next.context.allocator);
+
+                var body_stream = std.io.fixedBufferStream(respond.body);
+                try std.compress.gzip.compress(body_stream.reader(), compressed.writer(), inner);
+
+                // TODO: consider having the headers be a part of the provision?
+                // might be nice to reuse them as things go on??
+                var header_list = std.ArrayList([2][]const u8).init(next.context.allocator);
+                try header_list.appendSlice(respond.headers);
+                try header_list.append(.{ "Content-Encoding", "gzip" });
+
+                respond.body = try compressed.toOwnedSlice();
+                respond.headers = try header_list.toOwnedSlice();
+                return respond;
+            }
+        }.gzip_mw,
+    };
+
+    return Middleware.init({}, func).layer();
+}