Parse connect properly

1 files changed, 5 insertions, 6 deletions

diff --git a/src/server/message_parser.zig b/src/server/message_parser.zig
index c8359d5..369f326 100644
--- a/src/server/message_parser.zig
+++ b/src/server/message_parser.zig
@@ -95,10 +95,9 @@ const Message = union(MessageType) {
 };
 
 fn parseJsonMessage(T: type, alloc: std.mem.Allocator, in: *std.Io.Reader) !T {
-    var json_reader: std.json.Reader = .init(alloc, in);
-    defer json_reader.deinit();
+    const slice = try in.takeDelimiterInclusive('}');
 
-    return std.json.parseFromTokenSourceLeaky(T, alloc, &json_reader, .{});
+    return std.json.parseFromSliceLeaky(T, alloc, slice, .{});
 }
 
 fn parsePub(in: *std.Io.Reader) !Message.Pub {
@@ -109,7 +108,7 @@ fn parsePub(in: *std.Io.Reader) !Message.Pub {
         reply_to = next;
         break :blk try std.fmt.parseUnsigned(usize, (try in.takeDelimiter(' ')) orelse return error.EndOfStream, 10);
     };
-    // in.toss(2); // CRLF
+    in.toss(2); // CRLF
     const payload = try in.take(bytes);
 
     return .{
@@ -127,7 +126,7 @@ pub fn parseNextMessage(alloc: std.mem.Allocator, in: *std.Io.Reader) ?Message {
         std.debug.print("word: {s}\n", .{word});
         break :blk MessageType.parse(word) orelse return null;
     };
-    // defer in.toss(2); // CRLF
+    defer in.toss(2); // CRLF
     return switch (message_type) {
         .connect => .{ .connect = parseJsonMessage(Message.Connect, alloc, in) catch return null },
         .@"pub" => .{ .@"pub" = parsePub(in) catch return null },
@@ -138,7 +137,7 @@ pub fn parseNextMessage(alloc: std.mem.Allocator, in: *std.Io.Reader) ?Message {
 
 test parseNextMessage {
     const input =
-        \\CONNECT {"verbose":false,"pedantic":false,"tls_required":false,"name":"NATS CLI Version v0.2.4","lang":"go","version":"1.43.0","protocol":1,"echo":true,"headers":true,"no_responders":true}
+        \\CONNECT {"verbose":false,"pedantic":false,"tls_required":false,"name":"NATS CLI Version v0.2.4","lang":"go","version":"1.43.0","protocol":1,"echo":true,"headers":true,"no_responders":true}\r\n
     ;
     var reader: std.Io.Reader = .fixed(input);
     var arena: std.heap.ArenaAllocator = .init(std.testing.allocator);