From d6d177aede7f314b9120b46c038493da51763815 Mon Sep 17 00:00:00 2001 From: Robby Zambito Date: Tue, 18 Nov 2025 13:21:34 -0500 Subject: --- src/main.zig | 167 ++++++++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 138 insertions(+), 29 deletions(-) (limited to 'src') diff --git a/src/main.zig b/src/main.zig index 7d0bfaf..6ebd11e 100644 --- a/src/main.zig +++ b/src/main.zig @@ -97,13 +97,6 @@ fn serverMain(gpa: std.mem.Allocator, iter: *std.process.ArgIterator, main_args: defer threaded.deinit(); const io = threaded.io(); - // const info: ServerInfo = .{ - // .server_id = "foo", - // .server_name = "bar", - // .version = "6.9.0", - // .max_payload = 6969, - // }; - const info: ServerInfo = .{ .server_id = "NBEK5DBBB4ZO5LTBGPXACZSB2QUTODC6GGN5NLOSPIGSRFWJID4XU52C", .server_name = "bar", @@ -113,6 +106,13 @@ 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 }, @@ -134,40 +134,149 @@ fn handleConnection(io: std.Io, stream: std.Io.net.Stream, info: ServerInfo) voi var w_buffer: [1024]u8 = undefined; var writer = stream.writer(io, &w_buffer); const out = &writer.interface; - writeInfo(out, info) catch |err| { - std.debug.print("failed to write to client: {}\n", .{err}); - }; var r_buffer: [1024]u8 = undefined; var reader = stream.reader(io, &r_buffer); const in = &reader.interface; - 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; + processClient(in, out, info) catch |err| { + std.debug.print("Error processing client: {}\n", .{err}); + }; - var timeout = io.async(std.Io.sleep, .{ io, .fromSeconds(1), .real }); - defer timeout.cancel(io) catch {}; + // 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 user_res = io.async(std.Io.Reader.streamRemaining, .{ in, stdout_writer }); - defer _ = user_res.cancel(io) catch {}; + // var timeout = io.async(std.Io.sleep, .{ io, .fromSeconds(10), .real }); + // defer timeout.cancel(io) catch {}; - switch (io.select(.{ - .timeout = &timeout, - .data = &user_res, - }) catch unreachable) { - .timeout => std.debug.print("timeout\n", .{}), - .data => |d| { - std.debug.print("received data {any}\n", .{d}); - }, + // 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 { + try writeInfo(out, info); + + const ClientState = struct { + verbose: bool = false, + pedantic: bool = false, + tls_required: bool = false, + auth_token: ?[]const u8 = null, + user: ?[]const u8 = null, + pass: ?[]const u8 = null, + name: ?[]const u8 = null, + lang: []const u8, + version: []const u8, + protocol: u32, + echo: ?bool = null, + sig: ?[]const u8 = null, + jwt: ?[]const u8 = null, + no_responders: ?bool = null, + headers: ?bool = null, + nkey: ?[]const u8 = null, + }; + + const MessageType = enum { + info, + connect, + @"pub", + hpub, + sub, + unsub, + msg, + hmsg, + ping, + pong, + @"+ok", + @"-err", + + // fn parse(input: []u8) !MessageType { + // // if (std.mem.eql(u8, "INFO", input)) return .info; + // if (std.mem.eql(u8, "CONNECT", input)) return .connect; + // if (std.mem.eql(u8, "PUB", input)) return .@"pub"; + // if (std.mem.eql(u8, "HPUB", input)) return .hpub; + // if (std.mem.eql(u8, "SUB", input)) return .sub; + // if (std.mem.eql(u8, "UNSUB", input)) return .unsub; + // // if (std.mem.eql(u8, "MSG", input)) return .msg; + // // if (std.mem.eql(u8, "HMSG", input)) return .hmsg; + // if (std.mem.eql(u8, "PING", input)) return .ping; + // if (std.mem.eql(u8, "PONG", input)) return .pong; + // // if (std.mem.eql(u8, "@"+OK"", input)) return .@"+ok"; + // // if (std.mem.eql(u8, "@"-ERR"", input)) return .@"-err"; + // return error.InvalidMessageType; + // } + + const client_types = std.StaticStringMap(@This()).initComptime( + .{ + // {"INFO", .info}, + .{ "CONNECT", .connect }, + .{ "PUB", .@"pub" }, + .{ "HPUB", .hpub }, + .{ "SUB", .sub }, + .{ "UNSUB", .unsub }, + // {"MSG", .msg}, + // {"HMSG", .hmsg}, + .{ "PING", .ping }, + .{ "PONG", .pong }, + // {"+OK", .@"+ok"}, + // {"-ERR", .@"-err"}, + }, + ); + fn parse(input: []u8) !@This() { + return client_types.get(input) orelse return error.InvalidMessageType; + } + }; + + const initial_message_type = try MessageType.parse((in.takeDelimiter(' ') catch return error.InvalidMessageType) orelse return error.InvalidMessageType); + if (initial_message_type != .connect) return error.InvalidMessageType; + + // move this inside client_state declaration + var json_parse_buf: [1024]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); + + std.debug.print("buffered:{s}\n", .{in.buffered()}); + + var client_state = try std.json.parseFromTokenSourceLeaky(ClientState, json_parse_alloc, &json_reader, .{}); + + std.debug.print("client_state: {any}\n", .{client_state}); + + while (true) { + // Rebase the next message to the start of the buffer + // in.rebase(in.buffer.len); + const next_message_type = try MessageType.parse((in.takeDelimiter(' ') catch return error.InvalidMessageType) orelse return error.InvalidMessageType); + + switch (next_message_type) { + .connect => { + json_parse_alloc_fb = std.heap.FixedBufferAllocator.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}); + }, + else => |msg| std.debug.print("received {}\n", .{msg}), + } } } fn writeInfo(out: *std.Io.Writer, info: ServerInfo) !void { - defer out.flush() catch {}; - _ = try out.write("INFO "); try std.json.Stringify.value(info, .{}, out); - _ = try out.write(&.{ 0x0D, 0x0A }); + _ = try out.write("\r\n"); + try out.flush(); } -- cgit