Implement the web UI server with Zig I/O

Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.

src/ui_server.zigmodified+324-43

Changes to src/ui_server.zig

		⋯13 linesShow the 13 lines above this hunk@@ -14,47 +14,237 @@ pub const UIServerOptions = struct {
14	14	verbose: bool = false,
15	15	};
16	16
17		-/// UI Server for web-based test visualization
18		-/// Note: std.net was removed in Zig 0.16. Server functionality is stubbed out
19		-/// until the new Io-based networking API stabilizes.
	17	+/// UI Server for web-based test visualization.
20	18	pub const UIServer = struct {
21	19	allocator: std.mem.Allocator,
22	20	options: UIServerOptions,
23	21	mutex: compat.Mutex = .{},
24	22	history: ?*test_history.TestHistory = null,
	23	+ threaded_io: std.Io.Threaded,
	24	+ listener: ?std.Io.net.Server = null,
	25	+ thread: ?std.Thread = null,
	26	+ running: std.atomic.Value(bool) = .init(false),
	27	+ active_client: ?std.Io.net.Stream = null,
	28	+ clients: std.ArrayList(std.Io.net.Stream) = .empty,
25	29
26	30	const Self = @This();
	31	+ const index_html = @embedFile("ui/index.html");
	32	+
	33	+ pub const Error = error{
	34	+ AlreadyStarted,
	35	+ InvalidRequest,
	36	+ ServerStopped,
	37	+ };
27	38
28	39	pub fn init(allocator: std.mem.Allocator, options: UIServerOptions) Self {
29	40	return .{
30	41	.allocator = allocator,
31	42	.options = options,
	43	+ .threaded_io = .init(allocator, .{ .environ = .empty }),
32	44	};
33	45	}
34	46
35	47	pub fn deinit(self: *Self) void {
36		- _ = self;
	48	+ self.stop();
	49	+ self.clients.deinit(self.allocator);
	50	+ self.threaded_io.deinit();
37	51	}
38	52
39		- /// Start the UI server (stubbed - std.net removed in Zig 0.16)
	53	+ /// Bind the configured address and start accepting HTTP clients.
40	54	pub fn start(self: *Self) !void {
	55	+ if (self.running.load(.acquire) or self.listener != null) return Error.AlreadyStarted;
	56	+
	57	+ const io = self.threaded_io.io();
	58	+ const address = if (std.ascii.eqlIgnoreCase(self.options.host, "localhost"))
	59	+ std.Io.net.IpAddress{ .ip4 = .loopback(self.options.port) }
	60	+ else
	61	+ try std.Io.net.IpAddress.parse(self.options.host, self.options.port);
	62	+
	63	+ self.listener = try address.listen(io, .{ .reuse_address = true });
	64	+ self.options.port = self.listener.?.socket.address.getPort();
	65	+ self.running.store(true, .release);
	66	+ self.thread = std.Thread.spawn(.{}, serve, .{self}) catch |err| {
	67	+ self.running.store(false, .release);
	68	+ self.listener.?.deinit(io);
	69	+ self.listener = null;
	70	+ return err;
	71	+ };
	72	+
41	73	if (self.options.verbose) {
42		- std.debug.print("UI Server: networking stubbed out (std.net removed in Zig 0.16)\n", .{});
	74	+ std.debug.print("UI Server: listening on http://{s}:{d}\n", .{ self.options.host, self.options.port });
	75	+ }
	76	+ }
	77	+
	78	+ /// Stop accepting connections and close every active event stream.
	79	+ pub fn stop(self: *Self) void {
	80	+ const was_running = self.running.swap(false, .acq_rel);
	81	+ if (was_running) {
	82	+ if (self.listener) |server| {
	83	+ const listening_stream = std.Io.net.Stream{ .socket = server.socket };
	84	+ listening_stream.shutdown(self.threaded_io.io(), .both) catch {};
	85	+ server.socket.close(self.threaded_io.io());
	86	+ }
	87	+ }
	88	+
	89	+ // A client may have connected without finishing its request headers.
	90	+ // Shutting down that in-flight socket ensures shutdown cannot hang
	91	+ // waiting for the request parser; the accept thread remains its owner.
	92	+ self.mutex.lock();
	93	+ if (self.active_client) |client| client.shutdown(self.threaded_io.io(), .both) catch {};
	94	+ self.mutex.unlock();
	95	+
	96	+ if (self.thread) |thread| {
	97	+ thread.join();
	98	+ self.thread = null;
43	99	}
	100	+
	101	+ self.listener = null;
	102	+
	103	+ self.mutex.lock();
	104	+ defer self.mutex.unlock();
	105	+ for (self.clients.items) |client| client.close(self.threaded_io.io());
	106	+ self.clients.clearRetainingCapacity();
44	107	}
45	108
46		- /// Accept a client connection (stubbed)
	109	+ /// Return the actual bound port. This differs from the requested port when
	110	+ /// port zero asks the operating system for an ephemeral port.
	111	+ pub fn port(self: *const Self) u16 {
	112	+ return self.options.port;
	113	+ }
	114	+
	115	+ fn serve(self: *Self) void {
	116	+ while (self.running.load(.acquire)) {
	117	+ self.acceptClient() catch |err| {
	118	+ if (!self.running.load(.acquire)) return;
	119	+ if (self.options.verbose) {
	120	+ std.debug.print("UI Server error: {s}\n", .{@errorName(err)});
	121	+ }
	122	+ compat.sleep(10 * std.time.ns_per_ms);
	123	+ };
	124	+ }
	125	+ }
	126	+
	127	+ /// Accept and serve one HTTP client connection.
47	128	pub fn acceptClient(self: *Self) !void {
48		- _ = self;
49		- // Stubbed: no networking available
50		- compat.sleep(100 * std.time.ns_per_ms);
	129	+ if (!self.running.load(.acquire)) return Error.ServerStopped;
	130	+ const stream = if (self.listener) |*listener|
	131	+ try listener.accept(self.threaded_io.io())
	132	+ else
	133	+ return Error.ServerStopped;
	134	+ if (!self.running.load(.acquire)) {
	135	+ stream.close(self.threaded_io.io());
	136	+ return Error.ServerStopped;
	137	+ }
	138	+
	139	+ self.mutex.lock();
	140	+ self.active_client = stream;
	141	+ self.mutex.unlock();
	142	+
	143	+ const disposition = self.handleClient(stream) catch |err| {
	144	+ self.finishClient(false) catch {};
	145	+ return err;
	146	+ };
	147	+ try self.finishClient(disposition == .event_stream);
51	148	}
52	149
53		- /// Broadcast an event to all connected clients (stubbed)
	150	+ const ClientDisposition = enum { close, event_stream };
	151	+
	152	+ fn handleClient(self: *Self, stream: std.Io.net.Stream) !ClientDisposition {
	153	+ var read_buffer: [4096]u8 = undefined;
	154	+ var stream_reader = stream.reader(self.threaded_io.io(), &read_buffer);
	155	+ const request_line_raw = (try stream_reader.interface.takeDelimiter('\n')) orelse return Error.InvalidRequest;
	156	+ const request_line = std.mem.trimEnd(u8, request_line_raw, "\r");
	157	+ var parts = std.mem.splitScalar(u8, request_line, ' ');
	158	+ const method = parts.next() orelse return Error.InvalidRequest;
	159	+ const target = parts.next() orelse return Error.InvalidRequest;
	160	+ _ = parts.next() orelse return Error.InvalidRequest;
	161	+
	162	+ const is_get = std.mem.eql(u8, method, "GET");
	163	+ const is_events = std.mem.eql(u8, target, "/events");
	164	+ const is_index = std.mem.eql(u8, target, "/") or std.mem.eql(u8, target, "/index.html");
	165	+
	166	+ while (true) {
	167	+ const header_raw = (try stream_reader.interface.takeDelimiter('\n')) orelse return Error.InvalidRequest;
	168	+ if (std.mem.trimEnd(u8, header_raw, "\r").len == 0) break;
	169	+ }
	170	+
	171	+ if (!is_get) {
	172	+ try self.respond(stream, "405 Method Not Allowed", "text/plain; charset=utf-8", "Method Not Allowed\n");
	173	+ } else if (is_events) {
	174	+ try self.openEventStream(stream);
	175	+ return .event_stream;
	176	+ } else if (is_index) {
	177	+ try self.respond(stream, "200 OK", "text/html; charset=utf-8", index_html);
	178	+ } else {
	179	+ try self.respond(stream, "404 Not Found", "text/plain; charset=utf-8", "Not Found\n");
	180	+ }
	181	+ return .close;
	182	+ }
	183	+
	184	+ fn finishClient(self: *Self, keep_open: bool) !void {
	185	+ self.mutex.lock();
	186	+ defer self.mutex.unlock();
	187	+ const client = self.active_client orelse return;
	188	+ self.active_client = null;
	189	+
	190	+ if (keep_open and self.running.load(.acquire)) {
	191	+ self.clients.append(self.allocator, client) catch |err| {
	192	+ client.close(self.threaded_io.io());
	193	+ return err;
	194	+ };
	195	+ } else {
	196	+ client.close(self.threaded_io.io());
	197	+ }
	198	+ }
	199	+
	200	+ fn respond(self: *Self, stream: std.Io.net.Stream, status: []const u8, content_type: []const u8, body: []const u8) !void {
	201	+ var write_buffer: [2048]u8 = undefined;
	202	+ var stream_writer = stream.writer(self.threaded_io.io(), &write_buffer);
	203	+ try stream_writer.interface.print(
	204	+ "HTTP/1.1 {s}\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n",
	205	+ .{ status, content_type, body.len },
	206	+ );
	207	+ try stream_writer.interface.writeAll(body);
	208	+ try stream_writer.interface.flush();
	209	+ }
	210	+
	211	+ fn openEventStream(self: *Self, stream: std.Io.net.Stream) !void {
	212	+ var write_buffer: [512]u8 = undefined;
	213	+ var stream_writer = stream.writer(self.threaded_io.io(), &write_buffer);
	214	+ try stream_writer.interface.writeAll(
	215	+ "HTTP/1.1 200 OK\r\n" ++
	216	+ "Content-Type: text/event-stream\r\n" ++
	217	+ "Cache-Control: no-cache\r\n" ++
	218	+ "Connection: keep-alive\r\n" ++
	219	+ "X-Accel-Buffering: no\r\n\r\n" ++
	220	+ "event: connected\ndata: {}\n\n",
	221	+ );
	222	+ try stream_writer.interface.flush();
	223	+ }
	224	+
	225	+ fn sendEvent(self: *Self, stream: std.Io.net.Stream, event: []const u8, data: []const u8) !void {
	226	+ var write_buffer: [1024]u8 = undefined;
	227	+ var stream_writer = stream.writer(self.threaded_io.io(), &write_buffer);
	228	+ try stream_writer.interface.print("event: {s}\ndata: {s}\n\n", .{ event, data });
	229	+ try stream_writer.interface.flush();
	230	+ }
	231	+
	232	+ /// Broadcast an event to all connected clients. A disconnected browser is
	233	+ /// removed without disrupting the test run or the remaining clients.
54	234	pub fn broadcast(self: *Self, event: []const u8, data: []const u8) !void {
55		- _ = self;
56		- _ = event;
57		- _ = data;
	235	+ self.mutex.lock();
	236	+ defer self.mutex.unlock();
	237	+
	238	+ var index: usize = 0;
	239	+ while (index < self.clients.items.len) {
	240	+ const client = self.clients.items[index];
	241	+ self.sendEvent(client, event, data) catch {
	242	+ client.close(self.threaded_io.io());
	243	+ _ = self.clients.swapRemove(index);
	244	+ continue;
	245	+ };
	246	+ index += 1;
	247	+ }
58	248	}
59	249	};
60	250
		⋯28 linesShow the 28 lines above this hunk@@ -89,69 +279,62 @@ pub const UIReporter = struct {
89	279	self.buffer.deinit(self.reporter.allocator);
90	280	}
91	281
92		- fn onRunStart(reporter: *reporter_mod.Reporter, total: usize) !void {
93		- const self: *Self = @fieldParentPtr("reporter", reporter);
94		-
	282	+ fn encode(self: *Self, value: anytype) !void {
95	283	self.buffer.clearRetainingCapacity();
96		- try self.buffer.print(reporter.allocator, "", .{total});
	284	+ var writer: std.Io.Writer.Allocating = .fromArrayList(self.reporter.allocator, &self.buffer);
	285	+ defer self.buffer = writer.toArrayList();
	286	+ try std.json.Stringify.value(value, .{}, &writer.writer);
	287	+ }
97	288
	289	+ fn onRunStart(reporter: *reporter_mod.Reporter, total: usize) !void {
	290	+ const self: *Self = @fieldParentPtr("reporter", reporter);
	291	+ try self.encode(.{ .total = total });
98	292	try self.server.broadcast("run_start", self.buffer.items);
99	293	}
100	294
101	295	fn onRunEnd(reporter: *reporter_mod.Reporter, results: *reporter_mod.TestResults) !void {
102	296	const self: *Self = @fieldParentPtr("reporter", reporter);
103	297
104		- self.buffer.clearRetainingCapacity();
105		- try self.buffer.print(reporter.allocator, "", .{
106		- results.total,
107		- results.passed,
108		- results.failed,
109		- results.skipped,
	298	+ try self.encode(.{
	299	+ .total = results.total,
	300	+ .passed = results.passed,
	301	+ .flaky = results.flaky,
	302	+ .failed = results.failed,
	303	+ .skipped = results.skipped,
110	304	});
111		-
112	305	try self.server.broadcast("run_end", self.buffer.items);
113	306	}
114	307
115	308	fn onSuiteStart(reporter: *reporter_mod.Reporter, suite_name: []const u8) !void {
116	309	const self: *Self = @fieldParentPtr("reporter", reporter);
117	310
118		- self.buffer.clearRetainingCapacity();
119		- try self.buffer.print(reporter.allocator, "", .{suite_name});
120		-
	311	+ try self.encode(.{ .name = suite_name });
121	312	try self.server.broadcast("suite_start", self.buffer.items);
122	313	}
123	314
124	315	fn onSuiteEnd(reporter: *reporter_mod.Reporter, suite_name: []const u8) !void {
125	316	const self: *Self = @fieldParentPtr("reporter", reporter);
126	317
127		- self.buffer.clearRetainingCapacity();
128		- try self.buffer.print(reporter.allocator, "", .{suite_name});
129		-
	318	+ try self.encode(.{ .name = suite_name });
130	319	try self.server.broadcast("suite_end", self.buffer.items);
131	320	}
132	321
133	322	fn onTestStart(reporter: *reporter_mod.Reporter, test_name: []const u8) !void {
134	323	const self: *Self = @fieldParentPtr("reporter", reporter);
135	324
136		- self.buffer.clearRetainingCapacity();
137		- try self.buffer.print(reporter.allocator, "", .{test_name});
138		-
	325	+ try self.encode(.{ .name = test_name });
139	326	try self.server.broadcast("test_start", self.buffer.items);
140	327	}
141	328
142	329	fn onTestEnd(reporter: *reporter_mod.Reporter, test_case: *const suite.TestCase) !void {
143	330	const self: *Self = @fieldParentPtr("reporter", reporter);
144	331
145		- self.buffer.clearRetainingCapacity();
146		-
147		- const error_msg = test_case.error_message orelse "";
148		- try self.buffer.print(reporter.allocator, "", .{
149		- test_case.name,
150		- @tagName(test_case.status),
151		- test_case.execution_time_ns,
152		- error_msg,
	332	+ try self.encode(.{
	333	+ .name = test_case.name,
	334	+ .status = @tagName(test_case.status),
	335	+ .execution_time_ns = test_case.execution_time_ns,
	336	+ .error_message = test_case.error_message orelse "",
153	337	});
154		-
155	338	try self.server.broadcast("test_end", self.buffer.items);
156	339	}
157	340	};
		⋯95 linesShow the 95 lines above this hunk@@ -253,4 +436,102 @@ test "UIReporter initialization" {
253	436
254	437	try std.testing.expectEqual(false, ui_reporter.reporter.use_colors);
255	438	try std.testing.expectEqual(@as(usize, 0), ui_reporter.buffer.items.len);
	439	+
	440	+ try ui_reporter.reporter.onSuiteStart("quoted \"suite\"\nname");
	441	+ try std.testing.expectEqualStrings("{\"name\":\"quoted \\\"suite\\\"\\nname\"}", ui_reporter.buffer.items);
	442	+}
	443	+
	444	+fn writeTestRequest(io: std.Io, stream: std.Io.net.Stream, target: []const u8) !void {
	445	+ var write_buffer: [512]u8 = undefined;
	446	+ var writer = stream.writer(io, &write_buffer);
	447	+ try writer.interface.print("GET {s} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n", .{target});
	448	+ try writer.interface.flush();
	449	+}
	450	+
	451	+fn expectLine(reader: *std.Io.Reader, expected: []const u8) !void {
	452	+ const raw = (try reader.takeDelimiter('\n')) orelse return error.EndOfStream;
	453	+ try std.testing.expectEqualStrings(expected, std.mem.trimEnd(u8, raw, "\r"));
	454	+}
	455	+
	456	+test "UI server serves the embedded page on an ephemeral port" {
	457	+ var server = UIServer.init(std.testing.allocator, .{ .port = 0 });
	458	+ defer server.deinit();
	459	+ try server.start();
	460	+ try std.testing.expect(server.port() != 0);
	461	+
	462	+ var client_io: std.Io.Threaded = .init(std.testing.allocator, .{ .environ = .empty });
	463	+ defer client_io.deinit();
	464	+ const io = client_io.io();
	465	+ const address = std.Io.net.IpAddress{ .ip4 = .loopback(server.port()) };
	466	+ const stream = try address.connect(io, .{ .mode = .stream });
	467	+ defer stream.close(io);
	468	+ try writeTestRequest(io, stream, "/");
	469	+
	470	+ var read_buffer: [4096]u8 = undefined;
	471	+ var stream_reader = stream.reader(io, &read_buffer);
	472	+ const response = try stream_reader.interface.allocRemaining(std.testing.allocator, .limited(UIServer.index_html.len + 1024));
	473	+ defer std.testing.allocator.free(response);
	474	+ try std.testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 200 OK\r\n"));
	475	+ try std.testing.expect(std.mem.indexOf(u8, response, "Zig Test Framework - Live Results") != null);
	476	+}
	477	+
	478	+test "UI server streams broadcast events over SSE" {
	479	+ var server = UIServer.init(std.testing.allocator, .{ .port = 0 });
	480	+ defer server.deinit();
	481	+ try server.start();
	482	+
	483	+ var client_io: std.Io.Threaded = .init(std.testing.allocator, .{ .environ = .empty });
	484	+ defer client_io.deinit();
	485	+ const io = client_io.io();
	486	+ const address = std.Io.net.IpAddress{ .ip4 = .loopback(server.port()) };
	487	+ const stream = try address.connect(io, .{ .mode = .stream });
	488	+ defer stream.close(io);
	489	+ try writeTestRequest(io, stream, "/events");
	490	+
	491	+ var read_buffer: [4096]u8 = undefined;
	492	+ var stream_reader = stream.reader(io, &read_buffer);
	493	+ try expectLine(&stream_reader.interface, "HTTP/1.1 200 OK");
	494	+ while (true) {
	495	+ const raw = (try stream_reader.interface.takeDelimiter('\n')) orelse return error.EndOfStream;
	496	+ if (std.mem.trimEnd(u8, raw, "\r").len == 0) break;
	497	+ }
	498	+ try expectLine(&stream_reader.interface, "event: connected");
	499	+ try expectLine(&stream_reader.interface, "data: {}");
	500	+ try expectLine(&stream_reader.interface, "");
	501	+
	502	+ try server.broadcast("run_start", "{\"total\":2}");
	503	+ try expectLine(&stream_reader.interface, "event: run_start");
	504	+ try expectLine(&stream_reader.interface, "data: {\"total\":2}");
	505	+ try expectLine(&stream_reader.interface, "");
	506	+}
	507	+
	508	+test "UI server shutdown cancels an incomplete HTTP request" {
	509	+ var server = UIServer.init(std.testing.allocator, .{ .port = 0 });
	510	+ defer server.deinit();
	511	+ try server.start();
	512	+
	513	+ var client_io: std.Io.Threaded = .init(std.testing.allocator, .{ .environ = .empty });
	514	+ defer client_io.deinit();
	515	+ const io = client_io.io();
	516	+ const address = std.Io.net.IpAddress{ .ip4 = .loopback(server.port()) };
	517	+ const stream = try address.connect(io, .{ .mode = .stream });
	518	+ defer stream.close(io);
	519	+
	520	+ var write_buffer: [128]u8 = undefined;
	521	+ var writer = stream.writer(io, &write_buffer);
	522	+ try writer.interface.writeAll("GET / HTTP/1.1\r\n");
	523	+ try writer.interface.flush();
	524	+
	525	+ var request_is_active = false;
	526	+ for (0..100) |_| {
	527	+ server.mutex.lock();
	528	+ request_is_active = server.active_client != null;
	529	+ server.mutex.unlock();
	530	+ if (request_is_active) break;
	531	+ compat.sleep(std.time.ns_per_ms);
	532	+ }
	533	+ try std.testing.expect(request_is_active);
	534	+
	535	+ server.stop();
	536	+ try std.testing.expect(!server.running.load(.acquire));
256	537	}