mirrors/zits
1
0
Fork 0
mirror of https://git.robbyzambito.me/zits synced 2026-02-04 03:34:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Parse connect properly

Browse Source
This commit is contained in:
Robby Zambito
2025-11-19 08:45:17 -05:00
parent 51008cd7e1
commit 938504e12e
1 changed files with 5 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/server/message_parser.zig
View File

@@ -95,10 +95,9 @@ const Message = union(MessageType) {
}; };
fn parseJsonMessage(T: type, alloc: std.mem.Allocator, in: *std.Io.Reader) !T { fn parseJsonMessage(T: type, alloc: std.mem.Allocator, in: *std.Io.Reader) !T {
var json_reader: std.json.Reader = .init(alloc, in); const slice = try in.takeDelimiterInclusive('}');
defer json_reader.deinit();
return std.json.parseFromTokenSourceLeaky(T, alloc, &json_reader, .{}); return std.json.parseFromSliceLeaky(T, alloc, slice, .{});
} }
fn parsePub(in: *std.Io.Reader) !Message.Pub { fn parsePub(in: *std.Io.Reader) !Message.Pub {
@@ -109,7 +108,7 @@ fn parsePub(in: *std.Io.Reader) !Message.Pub {
reply_to = next; reply_to = next;
break :blk try std.fmt.parseUnsigned(usize, (try in.takeDelimiter(' ')) orelse return error.EndOfStream, 10); 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); const payload = try in.take(bytes);
return .{ return .{
@@ -127,7 +126,7 @@ pub fn parseNextMessage(alloc: std.mem.Allocator, in: *std.Io.Reader) ?Message {
std.debug.print("word: {s}\n", .{word}); std.debug.print("word: {s}\n", .{word});
break :blk MessageType.parse(word) orelse return null; break :blk MessageType.parse(word) orelse return null;
}; };
// defer in.toss(2); // CRLF defer in.toss(2); // CRLF
return switch (message_type) { return switch (message_type) {
.connect => .{ .connect = parseJsonMessage(Message.Connect, alloc, in) catch return null }, .connect => .{ .connect = parseJsonMessage(Message.Connect, alloc, in) catch return null },
.@"pub" => .{ .@"pub" = parsePub(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 { test parseNextMessage {
const input = 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 reader: std.Io.Reader = .fixed(input);
var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); var arena: std.heap.ArenaAllocator = .init(std.testing.allocator);