Made progress, but not perfect.

the message isn't crossing container boundaries, but it works in the
test!

1 files changed, 13 insertions, 2 deletions

diff --git a/src/server/message_parser.zig b/src/server/message_parser.zig
index e791d14..5669d2a 100644
--- a/src/server/message_parser.zig
+++ b/src/server/message_parser.zig
@@ -13,6 +13,7 @@ pub const MessageType = enum {
     pong,
     @"+ok",
     @"-err",
+    eos,
 
     fn parseMemEql(input: []const u8) ?MessageType {
         // if (std.mem.eql(u8, "INFO", input)) return .info;
@@ -44,6 +45,9 @@ pub const Message = union(MessageType) {
     pong,
     @"+ok": void,
     @"-err": void,
+    // Not an actual NATS message, but used to signal end of stream was reached in the input,
+    // and we should close the reader.
+    eos: void,
     pub const ServerInfo = struct {
         /// The unique identifier of the NATS server.
         server_id: []const u8,
@@ -75,7 +79,7 @@ pub const Message = union(MessageType) {
         allocator: std.heap.ArenaAllocator,
         connect: Connect,
 
-        pub fn deinit(self: *AllocatedConnect) void {
+        pub fn deinit(self: AllocatedConnect) void {
             self.allocator.deinit();
         }
     };
@@ -166,7 +170,14 @@ pub const Message = union(MessageType) {
                 try operation_string.appendBounded(byte);
                 try in.discardAll(1);
             } else break;
-        } else |err| return err;
+        } else |err| switch (err) {
+            error.EndOfStream => {
+                return .{ .eos = {} };
+            },
+            else => {
+                return err;
+            },
+        }
 
         const operation = parse(operation_string.items) orelse {
             return error.InvalidOperation;