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

Browse Source
This commit is contained in:
Robby Zambito
2025-11-21 13:21:23 -05:00
parent dbf189adc0
commit 474df6deed
1 changed files with 59 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
src/server/message_parser.zig
View File

@@ -94,9 +94,11 @@ const Message = union(MessageType) {
};
fn parseJsonMessage(T: type, alloc: std.mem.Allocator, in: *std.Io.Reader) !T {
const slice = try in.takeDelimiterInclusive('}');
return std.json.parseFromSliceLeaky(T, alloc, slice, .{});
var reader: std.json.Reader = .init(alloc, in);
return std.json.innerParse(T, alloc, &reader, .{
.max_value_len = std.json.default_max_value_len,
.allocate = .alloc_always,
});
}
fn parsePub(in: *std.Io.Reader) !Message.Pub {
@@ -136,22 +138,69 @@ pub fn parseNextMessage(alloc: std.mem.Allocator, in: *std.Io.Reader) ?Message {
break :blk MessageType.parse(word[0..len]) orelse return null;
};
std.debug.print("buffered: '{s}'\n", .{in.buffered()});
defer in.toss(2); // CRLF
// defer in.toss(2); // CRLF
return switch (message_type) {
.connect => .{ .connect = parseJsonMessage(Message.Connect, alloc, in) catch return null },
.connect => blk: {
const value: ?Message = .{ .connect = parseJsonMessage(Message.Connect, alloc, in) catch return null };
std.debug.print("value: {s}\n", .{value.?.connect.name.?});
std.debug.print("buffered: '{d}'\n", .{in.buffered().len});
break :blk value;
},
.@"pub" => .{ .@"pub" = parsePub(in) catch |err| std.debug.panic("{}", .{err}) },
.ping => .ping,
else => null,
};
}
// 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}\r\naftertheendoftheinput";
// var reader: std.Io.Reader = .fixed(input);
// var arena: std.heap.ArenaAllocator = .init(std.testing.allocator);
// defer arena.deinit();
// const gpa = arena.allocator();
// const msg: ?Message = parseNextMessage(gpa, &reader);
// const expected: ?Message = .{ .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,
// } };
// try std.testing.expect(msg != null);
// try std.testing.expectEqualDeep(msg, expected);
// }
// {
// const input = "PUB hi 3\r\nfoo\r\n";
// var reader: std.Io.Reader = .fixed(input);
// var arena: std.heap.ArenaAllocator = .init(std.testing.allocator);
// defer arena.deinit();
// const gpa = arena.allocator();
// const msg: ?Message = parseNextMessage(gpa, &reader);
// std.debug.print("msg: {any}\n", .{msg});
// const expected: ?Message = .{ .@"pub" = .{
// .subject = "hi",
// .bytes = 3,
// .payload = "foo",
// } };
// try std.testing.expect(msg != null);
// try std.testing.expectEqualDeep(msg, expected);
// }
// }
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}\r\n";
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}\r\nPUB hi 3\r\nfoo\r\n";
var reader: std.Io.Reader = .fixed(input);
var arena: std.heap.ArenaAllocator = .init(std.testing.allocator);
defer arena.deinit();
const gpa = arena.allocator();
{
const msg: ?Message = parseNextMessage(gpa, &reader);
const expected: ?Message = .{ .connect = .{
.verbose = false,
@@ -169,11 +218,6 @@ test parseNextMessage {
try std.testing.expectEqualDeep(msg, expected);
}
{
const input = "PUB hi 3\r\nfoo\r\n";
var reader: std.Io.Reader = .fixed(input);
var arena: std.heap.ArenaAllocator = .init(std.testing.allocator);
defer arena.deinit();
const gpa = arena.allocator();
const msg: ?Message = parseNextMessage(gpa, &reader);
std.debug.print("msg: {any}\n", .{msg});
const expected: ?Message = .{ .@"pub" = .{