lib/http/src/router/runtime.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const http = @import("../root.zig");
3 const Connection = http.Connection;
4 const Request = http.Request;
5 const Response = http.Response;
6 const ParseError = http.ParseError;
7 const WebSocket = http.WebSocket;
8 const WebSocketCloseCode = http.WebSocketCloseCode;
9
10 const log = std.log.scoped(.http_router);
11
12 const RouterRouteKey = struct {
13 method: Request.Method,
14 path: []const u8,
15 };
16
17 const RouterRouteKeyContext = struct {
18 pub fn hash(_: RouterRouteKeyContext, key: RouterRouteKey) u64 {
19 var h = std.hash.Wyhash.init(0);
20 h.update(std.mem.asBytes(&key.method));
21 h.update(key.path);
22 return h.final();
23 }
24
25 pub fn eql(_: RouterRouteKeyContext, a: RouterRouteKey, b: RouterRouteKey) bool {
26 return a.method == b.method and std.mem.eql(u8, a.path, b.path);
27 }
28 };
29
30 fn RouterRouteEntryType(comptime Handler: type, comptime WsHandler: type) type {
31 return struct {
32 handler: Handler,
33 is_websocket: bool,
34 ws_handler: ?WsHandler,
35 websocket_limits: ?WebSocket.Limits,
36 };
37 }
38
39 pub fn Router(comptime Context: type) type {
40 return struct {
41 allocator: std.mem.Allocator,
42 routes: RouteMap,
43 not_found_handler: ?Handler,
44 context: Context,
45
46 const Self = @This();
47
48 pub const Handler = *const fn (Context, *Request, *Response) anyerror!void;
49 pub const WsHandler = *const fn (Context, *WebSocket) anyerror!void;
50
51 const RouteEntry: type = RouterRouteEntryType(Handler, WsHandler);
52 const RouteMap: type = std.HashMap(RouterRouteKey, RouteEntry, RouterRouteKeyContext, 80);
53
54 const WebSocketSession = struct {
55 allocator: std.mem.Allocator,
56 websocket: WebSocket,
57 context: Context,
58 handler: WsHandler,
59
60 fn drive(self: *WebSocketSession) !void {
61 self.handler(self.context, &self.websocket) catch |err| switch (err) {
62 error.WouldBlock => return,
63 else => {
64 self.websocket.close(WebSocketCloseCode.internal_error, "Handler error") catch |close_err| {
65 log.warn("failed to send websocket close after handler error: {s}", .{@errorName(close_err)});
66 };
67 self.websocket.connection.markClosing();
68 return err;
69 },
70 };
71 if (self.websocket.state == .closed) self.websocket.connection.markClosing();
72 }
73
74 fn destroy(self: *WebSocketSession) void {
75 self.websocket.deinit(self.allocator);
76 self.allocator.destroy(self);
77 }
78 };
79
80 pub fn init(allocator: std.mem.Allocator, context: Context) Self {
81 return .{
82 .allocator = allocator,
83 .routes = RouteMap.init(allocator),
84 .not_found_handler = null,
85 .context = context,
86 };
87 }
88
89 pub fn deinit(self: *Self) void {
90 self.routes.deinit();
91 }
92
93 pub fn route(self: *Self, method: Request.Method, path: []const u8, handler: Handler) !void {
94 try self.routes.put(.{ .method = method, .path = path }, .{
95 .handler = handler,
96 .is_websocket = false,
97 .ws_handler = null,
98 .websocket_limits = null,
99 });
100 }
101
102 pub fn get(self: *Self, path: []const u8, handler: Handler) !void {
103 try self.route(.GET, path, handler);
104 }
105
106 pub fn head(self: *Self, path: []const u8, handler: Handler) !void {
107 try self.route(.HEAD, path, handler);
108 }
109
110 pub fn post(self: *Self, path: []const u8, handler: Handler) !void {
111 try self.route(.POST, path, handler);
112 }
113
114 pub fn put(self: *Self, path: []const u8, handler: Handler) !void {
115 try self.route(.PUT, path, handler);
116 }
117
118 pub fn delete(self: *Self, path: []const u8, handler: Handler) !void {
119 try self.route(.DELETE, path, handler);
120 }
121
122 pub fn websocket(
123 self: *Self,
124 path: []const u8,
125 limits: WebSocket.Limits,
126 upgrade_handler: Handler,
127 ws_handler: WsHandler,
128 ) !void {
129 try self.routes.put(.{ .method = .GET, .path = path }, .{
130 .handler = upgrade_handler,
131 .is_websocket = true,
132 .ws_handler = ws_handler,
133 .websocket_limits = limits,
134 });
135 }
136
137 pub fn notFound(self: *Self, handler: Handler) void {
138 self.not_found_handler = handler;
139 }
140
141 pub fn handleRequest(self: *Self, conn: *Connection) !bool {
142 switch (conn.currentState()) {
143 .http => {},
144 .websocket => {
145 conn.driveProtocol() catch |err| {
146 conn.markClosing();
147 return err;
148 };
149 if (conn.currentState() == .websocket) conn.waitForInput();
150 return false;
151 },
152 .closing, .closed => return false,
153 }
154
155 const parse_result = while (true) {
156 if (conn.hasBufferedInput()) {
157 if (Request.parse(
158 conn.requestScratch(),
159 conn.bufferedInput(),
160 )) |result| {
161 break result;
162 } else |err| {
163 if (err != ParseError.IncompleteRequest) {
164 try sendBadRequest(conn);
165 conn.markClosing();
166 return false;
167 }
168 }
169 }
170 _ = conn.bufferInput() catch |err| switch (err) {
171 error.ConnectionClosed => return false,
172 error.WouldBlock => {
173 conn.waitForRead();
174 return false;
175 },
176 error.InputCapacityExceeded => {
177 try sendBadRequest(conn);
178 conn.markClosing();
179 return false;
180 },
181 else => return err,
182 };
183 };
184 defer conn.consumeBufferedInput(parse_result.consumed);
185
186 var req = parse_result.request;
187
188 var res = Response.init(conn.responseScratch());
189 defer res.deinit();
190
191 const key = RouterRouteKey{ .method = req.method, .path = req.pathOnly() };
192 if (self.routes.get(key)) |entry| {
193 if (entry.is_websocket and req.isWebSocketUpgrade()) {
194 return self.handleWebSocketUpgrade(conn, &req, entry);
195 } else {
196 entry.handler(self.context, &req, &res) catch |err| {
197 try sendInternalError(conn, req.method);
198 conn.markClosing();
199 return err;
200 };
201 }
202 } else if (self.not_found_handler) |handler| {
203 handler(self.context, &req, &res) catch |err| {
204 try sendInternalError(conn, req.method);
205 conn.markClosing();
206 return err;
207 };
208 } else {
209 res.status = 404;
210 res.status_text = "Not Found";
211 res.body = "Not Found";
212 }
213
214 try writeResponse(conn, try res.serializeForMethod(req.method));
215
216 const connection_header = req.headers.get("Connection") orelse "";
217 const keep_alive = connectionWantsKeepAlive(req.version, connection_header);
218 if (keep_alive) {
219 conn.waitForInput();
220 } else {
221 conn.markClosing();
222 }
223 return keep_alive;
224 }
225
226 fn handleWebSocketUpgrade(self: *Self, conn: *Connection, req: *Request, entry: RouteEntry) !bool {
227 const handshake = WebSocket.validateClientHandshake(req) catch {
228 try sendBadRequest(conn);
229 conn.markClosing();
230 return false;
231 };
232
233 var res = Response.init(conn.responseScratch());
234 defer res.deinit();
235
236 entry.handler(self.context, req, &res) catch |err| {
237 try sendInternalError(conn, req.method);
238 conn.markClosing();
239 return err;
240 };
241
242 if (res.status != 200 and res.status != 0) {
243 try writeResponse(conn, try res.serializeForMethod(req.method));
244 conn.markClosing();
245 return false;
246 }
247
248 const session = try self.allocator.create(WebSocketSession);
249 errdefer self.allocator.destroy(session);
250 session.* = .{
251 .allocator = self.allocator,
252 .websocket = try WebSocket.init(
253 self.allocator,
254 entry.websocket_limits.?,
255 conn,
256 ),
257 .context = self.context,
258 .handler = entry.ws_handler.?,
259 };
260 errdefer session.websocket.deinit(self.allocator);
261
262 const accept_key = handshake.accept();
263
264 var ws_res = try Response.switchingProtocols(conn.responseScratch(), &accept_key);
265 defer ws_res.deinit();
266 try writeResponse(conn, try ws_res.serialize());
267
268 session.websocket.activate();
269 conn.installProtocol(session, WebSocketSession.drive, WebSocketSession.destroy);
270
271 return false;
272 }
273
274 fn sendBadRequest(conn: *Connection) !void {
275 var res = Response.init(conn.responseScratch());
276 defer res.deinit();
277 res.status = 400;
278 res.status_text = "Bad Request";
279 res.body = "Bad Request";
280
281 try writeResponse(conn, try res.serialize());
282 }
283
284 fn sendInternalError(conn: *Connection, method: Request.Method) !void {
285 var res = Response.init(conn.responseScratch());
286 defer res.deinit();
287 res.status = 500;
288 res.status_text = "Internal Server Error";
289 res.body = "Internal Server Error";
290
291 try writeResponse(conn, try res.serializeForMethod(method));
292 }
293 };
294 }
295
296 fn connectionWantsKeepAlive(version: Request.Version, connection_header: []const u8) bool {
297 if (version == .http_1_0) {
298 return containsHeaderToken(connection_header, "keep-alive");
299 }
300 return !containsHeaderToken(connection_header, "close");
301 }
302
303 fn containsHeaderToken(value: []const u8, token: []const u8) bool {
304 var iter = std.mem.splitScalar(u8, value, ',');
305 while (iter.next()) |part| {
306 const trimmed = std.mem.trim(u8, part, " \t");
307 if (std.ascii.eqlIgnoreCase(trimmed, token)) return true;
308 }
309 return false;
310 }
311
312 pub fn upgradeWebsocket(
313 allocator: std.mem.Allocator,
314 limits: WebSocket.Limits,
315 conn: *Connection,
316 req: *const Request,
317 ) !WebSocket {
318 const handshake = try WebSocket.validateClientHandshake(req);
319 const accept_key = handshake.accept();
320 var websocket = try WebSocket.init(allocator, limits, conn);
321 errdefer websocket.deinit(allocator);
322
323 var response = try Response.switchingProtocols(conn.responseScratch(), &accept_key);
324 defer response.deinit();
325 try writeResponse(conn, try response.serialize());
326
327 conn.markWebSocket();
328 websocket.activate();
329 return websocket;
330 }
331
332 fn writeResponse(conn: *Connection, serialized: http.ResponseSerialized) !void {
333 try conn.write(serialized.head);
334 if (serialized.body) |body| try conn.write(body);
335 }