diff options
Diffstat (limited to 'src/main.zig')
| -rw-r--r-- | src/main.zig | 116 |
1 files changed, 44 insertions, 72 deletions
diff --git a/src/main.zig b/src/main.zig index fb6be1c..e24bbcb 100644 --- a/src/main.zig +++ b/src/main.zig @@ -2,8 +2,7 @@ const std = @import("std"); const zits = @import("zits"); const clap = @import("clap"); -const MessageType = zits.MessageParser.MessageType; -const parseNextMessage = zits.MessageParser.parseNextMessage; +const Message = zits.MessageParser.Message; const SubCommands = enum { help, @@ -27,7 +26,7 @@ const main_params = clap.parseParamsComptime( const MainArgs = clap.ResultEx(clap.Help, &main_params, main_parsers); pub fn main() !void { - var dba = std.heap.DebugAllocator(.{}){}; + var dba: std.heap.DebugAllocator(.{}) = .init; defer _ = dba.deinit(); const gpa = dba.allocator(); @@ -109,30 +108,27 @@ fn serverMain(gpa: std.mem.Allocator, iter: *std.process.ArgIterator, main_args: .max_payload = 1048576, }; - // const info: ServerInfo = .{ - // .server_id = "foo", - // .server_name = "bar", - // .version = "6.9.0", - // .max_payload = 6969, - // }; - - var server = try std.Io.net.IpAddress.listen(.{ - .ip4 = .{ - .bytes = .{ 0, 0, 0, 0 }, - .port = info.port, + var server = try std.Io.net.IpAddress.listen( + .{ + .ip4 = .{ + .bytes = .{ 0, 0, 0, 0 }, + .port = info.port, + }, }, - }, io, .{}); + io, + .{}, + ); defer server.deinit(io); var group: std.Io.Group = .init; defer group.wait(io); for (0..5) |_| { const stream = try server.accept(io); - group.async(io, handleConnection, .{ io, stream, info }); + group.async(io, handleConnection, .{ gpa, io, stream, info }); } } -fn handleConnection(io: std.Io, stream: std.Io.net.Stream, info: ServerInfo) void { +fn handleConnection(allocator: std.mem.Allocator, io: std.Io, stream: std.Io.net.Stream, info: ServerInfo) void { defer stream.close(io); var w_buffer: [1024]u8 = undefined; var writer = stream.writer(io, &w_buffer); @@ -142,76 +138,52 @@ fn handleConnection(io: std.Io, stream: std.Io.net.Stream, info: ServerInfo) voi var reader = stream.reader(io, &r_buffer); const in = &reader.interface; - processClient(in, out, info) catch |err| { + processClient(allocator, in, out, info) catch |err| { std.debug.panic("Error processing client: {}\n", .{err}); }; - - // var stdout_buffer: [1024]u8 = undefined; - // const stdout_file = std.fs.File.stdout(); - // var stdout_file_writer = stdout_file.writer(&stdout_buffer); - // const stdout_writer = &stdout_file_writer.interface; - - // var timeout = io.async(std.Io.sleep, .{ io, .fromSeconds(10), .real }); - // defer timeout.cancel(io) catch {}; - - // var user_res = io.async(std.Io.Reader.streamRemaining, .{ in, stdout_writer }); - // defer _ = user_res.cancel(io) catch {}; - - // switch (io.select(.{ - // .timeout = &timeout, - // .data = &user_res, - // }) catch unreachable) { - // .timeout => std.debug.print("timeout\n", .{}), - // .data => |_| { - // stdout_writer.flush() catch |err| { - // std.debug.print("Could not flush stdout: {}\n", .{err}); - // }; - // // std.debug.print("received data {any}\n", .{d}); - // }, - // } } -fn processClient(in: *std.Io.Reader, out: *std.Io.Writer, info: ServerInfo) !void { +fn processClient(gpa: std.mem.Allocator, in: *std.Io.Reader, out: *std.Io.Writer, info: ServerInfo) !void { try writeInfo(out, info); - // move this inside client_state declaration - var json_parse_buf: [4096]u8 = undefined; - var json_parse_alloc_fb: std.heap.FixedBufferAllocator = std.heap.FixedBufferAllocator.init(&json_parse_buf); - var json_parse_alloc = json_parse_alloc_fb.allocator(); - var json_reader: std.json.Reader = .init(json_parse_alloc, in); - - // var client_state = try std.json.parseFromSliceLeaky(ClientState, json_parse_alloc, in.buffered(), .{}); - // in.toss(in.buffered().len); - - // var client_state = try std.json.parseFromTokenSourceLeaky(ClientState, json_parse_alloc, &json_reader, .{}); - - const client_state = 0; - std.debug.print("client_state: {}\n", .{client_state}); + var client_state_arena: std.heap.ArenaAllocator = .init(gpa); + defer client_state_arena.deinit(); + const client_state = (try Message.next(client_state_arena.allocator(), in)).connect; + _ = client_state; + var message_parsing_arena: std.heap.ArenaAllocator = .init(gpa); + defer message_parsing_arena.deinit(); + const message_parsing_allocator = message_parsing_arena.allocator(); while (true) { - const next_message_type = parseNextMessage(json_parse_alloc, in) orelse return; - - switch (next_message_type) { + defer _ = message_parsing_arena.reset(.retain_capacity); + const next_message = Message.next(message_parsing_allocator, in) catch |err| { + switch (err) { + error.EndOfStream => { + break; + }, + else => { + return err; + }, + } + }; + switch (next_message) { .connect => |connect| { - std.debug.print("connect: {s}\n", .{connect.name orelse "\"\""}); - json_parse_alloc_fb = .init(&json_parse_buf); - json_parse_alloc = json_parse_alloc_fb.allocator(); - json_reader = .init(json_parse_alloc, in); - // client_state = try std.json.parseFromTokenSourceLeaky(ClientState, json_parse_alloc, &json_reader, .{}); - std.debug.print("client_state: {any}\n", .{client_state}); + std.debug.panic("Connection message after already connected: {any}\n", .{connect}); }, - .ping => writePong(out) catch |err| { - std.debug.panic("failed to pong: {any}\n", .{err}); - }, - else => |msg| std.debug.print("received {}\n", .{msg}), + .ping => try writePong(out), + .@"pub" => try writeOk(out), + else => |msg| std.debug.panic("Message type not implemented: {any}\n", .{msg}), } } } +fn writeOk(out: *std.Io.Writer) !void { + _ = try out.write("+OK\r\n"); + try out.flush(); +} + fn writePong(out: *std.Io.Writer) !void { - std.debug.print("in writePong\n", .{}); - _ = try out.write("PONG"); - _ = try out.write("\r\n"); + _ = try out.write("PONG\r\n"); try out.flush(); } |