2 files changed, 29 insertions, 15 deletions

diff --git a/src/main.zig b/src/main.zig
index bcb2d98..fb6be1c 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -3,6 +3,7 @@ const zits = @import("zits");
 const clap = @import("clap");
 
 const MessageType = zits.MessageParser.MessageType;
+const parseNextMessage = zits.MessageParser.parseNextMessage;
 
 const SubCommands = enum {
     help,
@@ -173,17 +174,12 @@ fn handleConnection(io: std.Io, stream: std.Io.net.Stream, info: ServerInfo) voi
 fn processClient(in: *std.Io.Reader, out: *std.Io.Writer, info: ServerInfo) !void {
     try writeInfo(out, info);
 
-    const initial_message_type = MessageType.parse((in.takeDelimiter(' ') catch return error.InvalidMessageType) orelse "") orelse return error.InvalidMessageType;
-    if (initial_message_type != .connect) return error.InvalidMessageType;
-
     // 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);
 
-    std.debug.print("buffered:{s}\n", .{in.buffered()});
-
     // var client_state = try std.json.parseFromSliceLeaky(ClientState, json_parse_alloc, in.buffered(), .{});
     // in.toss(in.buffered().len);
 
@@ -193,23 +189,32 @@ fn processClient(in: *std.Io.Reader, out: *std.Io.Writer, info: ServerInfo) !voi
     std.debug.print("client_state: {}\n", .{client_state});
 
     while (true) {
-        // Rebase the next message to the start of the buffer
-        // in.rebase(in.buffer.len);
-        const next_message_type = MessageType.parse((in.takeDelimiter(' ') catch return error.InvalidMessageType) orelse "") orelse return error.InvalidMessageType;
+        const next_message_type = parseNextMessage(json_parse_alloc, in) orelse return;
 
         switch (next_message_type) {
-            .connect => {
-                json_parse_alloc_fb = std.heap.FixedBufferAllocator.init(&json_parse_buf);
+            .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});
             },
+            .ping => writePong(out) catch |err| {
+                std.debug.panic("failed to pong: {any}\n", .{err});
+            },
             else => |msg| std.debug.print("received {}\n", .{msg}),
         }
     }
 }
 
+fn writePong(out: *std.Io.Writer) !void {
+    std.debug.print("in writePong\n", .{});
+    _ = try out.write("PONG");
+    _ = try out.write("\r\n");
+    try out.flush();
+}
+
 fn writeInfo(out: *std.Io.Writer, info: ServerInfo) !void {
     _ = try out.write("INFO ");
     try std.json.Stringify.value(info, .{}, out);
diff --git a/src/server/message_parser.zig b/src/server/message_parser.zig
index 917ea70..404a4e3 100644
--- a/src/server/message_parser.zig
+++ b/src/server/message_parser.zig
@@ -123,15 +123,24 @@ fn parsePub(in: *std.Io.Reader) !Message.Pub {
 // try returning error in debug mode, only null in release?
 pub fn parseNextMessage(alloc: std.mem.Allocator, in: *std.Io.Reader) ?Message {
     const message_type: MessageType = blk: {
-        const word: []const u8 = (in.takeDelimiter(' ') catch return null) orelse return null;
-        std.debug.print("word: {s}\n", .{word});
-        break :blk MessageType.parse(word) orelse return null;
+        var word: ["CONNECT".len]u8 = undefined;
+        var len: usize = 0;
+        for (&word, 0..) |*b, i| {
+            const byte = in.takeByte() catch return null;
+            if (std.ascii.isUpper(byte)) {
+                b.* = byte;
+                len = i + 1;
+            } else break;
+        }
+        std.debug.print("word: '{s}'\n", .{word[0..len]});
+        break :blk MessageType.parse(word[0..len]) orelse return null;
     };
-    // defer in.toss(2); // CRLF
+    std.debug.print("buffered: '{s}'\n", .{in.buffered()});
+    defer in.toss(2); // CRLF
     return switch (message_type) {
         .connect => .{ .connect = parseJsonMessage(Message.Connect, alloc, in) catch return null },
         .@"pub" => .{ .@"pub" = parsePub(in) catch |err| std.debug.panic("{}", .{err}) },
-        .ping => .{ .ping = {} },
+        .ping => .ping,
         else => null,
     };
 }