Change msgpack to use API similar to std.json

Author: lalinsky

lib/msgpack/README.md

@@ -14,14 +14,16 @@ const Message = struct {
 var buffer = std.ArrayList(u8).init(allocator);
 defer buffer.deinit();
 
-try msgpack.encode(buffer.writer(), Message, .{
+try msgpack.encode(Message{
     .name = "John",
     .age = 20,
-});
+}, buffer.writer());
 
-var stream = std.io.fixedBufferStream(buffer.items);
-const message = try decode(stream.reader(), allocator, Message);
-defer allocator.free(message.name);
+const decoded = try msgpack.decodeFromSlice(Message, allocator, buffer.items);
+defer decoded.deinit();
+
+std.debug.assert(std.mem.eql(u8, decoded.name, "John"));
+std.debug.assert(decoded.age == 20);
 ```
 
 Change the default format from using field names to field indexes:

lib/msgpack/msgpack.zig

@@ -1,5 +1,6 @@
 const std = @import("std");
 const Allocator = std.mem.Allocator;
+const ArenaAllocator = std.heap.ArenaAllocator;
 
 const NoAllocator = @import("utils.zig").NoAllocator;
 
@@ -245,18 +246,40 @@ pub fn unpacker(reader: anytype, allocator: ?Allocator) Unpacker(@TypeOf(reader)
     return Unpacker(@TypeOf(reader)).init(reader, allocator orelse NoAllocator.allocator());
 }
 
-pub fn unpackerNoAlloc(reader: anytype) Unpacker(@TypeOf(reader)) {
-    return Unpacker(@TypeOf(reader)).init(reader, NoAllocator.allocator());
+pub fn encode(value: anytype, writer: anytype) !void {
+    return try packer(writer).write(@TypeOf(value), value);
 }
 
-pub fn encode(writer: anytype, comptime T: type, value: T) !void {
-    return try packer(writer).write(T, value);
+pub const Decoded = std.json.Parsed;
+
+pub fn decode(comptime T: type, allocator: Allocator, reader: anytype) !Decoded(T) {
+    var parsed = Decoded(T){
+        .arena = try allocator.create(ArenaAllocator),
+        .value = undefined,
+    };
+    errdefer allocator.destroy(parsed.arena);
+    parsed.arena.* = ArenaAllocator.init(allocator);
+    errdefer parsed.arena.deinit();
+
+    parsed.value = try decodeLeaky(T, parsed.arena.allocator(), reader);
+
+    return parsed;
 }
 
-pub fn decode(reader: anytype, allocator: ?Allocator, comptime T: type) !T {
+pub fn decodeLeaky(comptime T: type, allocator: ?Allocator, reader: anytype) !T {
     return try unpacker(reader, allocator).read(T);
 }
 
+pub fn decodeFromSlice(comptime T: type, allocator: Allocator, data: []const u8) !Decoded(T) {
+    var stream = std.io.fixedBufferStream(data);
+    return try decode(T, allocator, stream.reader());
+}
+
+pub fn decodeFromSliceLeaky(comptime T: type, allocator: ?Allocator, data: []const u8) !T {
+    var stream = std.io.fixedBufferStream(data);
+    return try decodeLeaky(T, allocator, stream.reader());
+}
+
 test {
     _ = std.testing.refAllDecls(@This());
 }

src/Oplog.zig

@@ -211,10 +211,10 @@ pub fn write(self: *Self, changes: []const Change) !u64 {
     var bufferred_writer = std.io.bufferedWriter(file.writer());
     const writer = bufferred_writer.writer();
 
-    try msgpack.encode(writer, Transaction, .{
+    try msgpack.encode(Transaction{
         .id = commit_id,
         .changes = changes,
-    });
+    }, writer);
 
     try bufferred_writer.flush();
 
@@ -341,7 +341,7 @@ pub const OplogFileIterator = struct {
     pub fn next(self: *OplogFileIterator) !?Transaction {
         _ = self.arena.reset(.retain_capacity);
 
-        return msgpack.decode(self.buffered_reader.reader(), self.arena.allocator(), Transaction) catch |err| {
+        return msgpack.decodeLeaky(Transaction, self.arena.allocator(), self.buffered_reader.reader()) catch |err| {
             if (err == error.EndOfStream) {
                 return null;
             }

src/filefmt.zig

@@ -384,7 +384,7 @@ pub fn readSegmentFile(dir: fs.Dir, id: SegmentId, segment: *FileSegment) !void
     var fixed_buffer_stream = std.io.fixedBufferStream(raw_data[0..]);
     const reader = fixed_buffer_stream.reader();
 
-    const unpacker = msgpack.unpackerNoAlloc(reader);
+    const unpacker = msgpack.unpacker(reader, null);
 
     const header = try unpacker.read(SegmentFileHeader);
 
@@ -513,8 +513,8 @@ pub fn writeIndexFile(dir: std.fs.Dir, segments: []const SegmentId) !void {
     var buffered_writer = std.io.bufferedWriter(file.file.writer());
     const writer = buffered_writer.writer();
 
-    try msgpack.encode(writer, IndexFileHeader, .{});
-    try msgpack.encode(writer, []const SegmentId, segments);
+    try msgpack.encode(IndexFileHeader{}, writer);
+    try msgpack.encode(segments, writer);
 
     try buffered_writer.flush();
 
@@ -537,12 +537,12 @@ pub fn readIndexFile(dir: std.fs.Dir, allocator: std.mem.Allocator) ![]SegmentId
     var buffered_reader = std.io.bufferedReader(file.reader());
     const reader = buffered_reader.reader();
 
-    const header = try msgpack.decode(reader, allocator, IndexFileHeader);
+    const header = try msgpack.decodeLeaky(IndexFileHeader, null, reader);
     if (header.magic != index_header_magic_v1) {
         return error.InvalidIndexfile;
     }
 
-    return try msgpack.decode(reader, allocator, []SegmentId);
+    return try msgpack.decodeLeaky([]SegmentId, allocator, reader);
 }
 
 test "readIndexFile/writeIndexFile" {

src/server.zig

@@ -160,7 +160,7 @@ fn writeResponse(value: anytype, req: *httpz.Request, res: *httpz.Response) !voi
         .json => try res.json(value, .{}),
         .msgpack => {
             res.header("content-type", "application/vnd.msgpack");
-            try msgpack.encode(res.writer(), @TypeOf(value), value);
+            try msgpack.encode(value, res.writer());
         },
     }
 }
@@ -192,8 +192,7 @@ fn getRequestBody(comptime T: type, req: *httpz.Request, res: *httpz.Response) !
             };
         },
         .msgpack => {
-            var stream = std.io.fixedBufferStream(content);
-            return msgpack.decode(stream.reader(), req.arena, T) catch {
+            return msgpack.decodeFromSliceLeaky(T, req.arena, content) catch {
                 res.status = 400;
                 try writeResponse(.{ .status = "invalid body" }, req, res);
                 return null;