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;
+    processClient(in, out, info) catch |err| {
-    const stdout_file = std.fs.File.stdout();
+        std.debug.print("Error processing client: {}\n", .{err});
-    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(1), .real });
+    // var stdout_buffer: [1024]u8 = undefined;
-    defer timeout.cancel(io) catch {};
+    // 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 });
+    // var timeout = io.async(std.Io.sleep, .{ io, .fromSeconds(10), .real });
-    defer _ = user_res.cancel(io) catch {};
+    // defer timeout.cancel(io) catch {};
 
-    switch (io.select(.{
+    // var user_res = io.async(std.Io.Reader.streamRemaining, .{ in, stdout_writer });
-        .timeout = &timeout,
+    // defer _ = user_res.cancel(io) catch {};
-        .data = &user_res,
+
-    }) catch unreachable) {
+    // switch (io.select(.{
-        .timeout => std.debug.print("timeout\n", .{}),
+    //     .timeout = &timeout,
-        .data => |d| {
+    //     .data = &user_res,
-            std.debug.print("received data {any}\n", .{d});
+    // }) 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();
 }