feat: add decodeTxnOp and encodeTxnResult to protocol
This commit is contained in:
1 parent
c928cc7302
commit
aad2726772
1 file changed
+152
@@ -801,3 +801,155 @@ pub const DecodedTxnOpEntry = struct {
|
||||
fieldsBytes: []const u8,
|
||||
};
|
||||
|
||||
pub fn decodeTxnOp(allocator: Allocator, payload: []const u8) ![]DecodedTxnOpEntry {
|
||||
var c = Cursor{ .data = payload };
|
||||
const count = try c.takeU32();
|
||||
var list = try std.ArrayList(DecodedTxnOpEntry).initCapacity(allocator, count);
|
||||
errdefer list.deinit();
|
||||
var i: u32 = 0;
|
||||
while (i < count) : (i += 1) {
|
||||
const verb = try TxnVerb.fromByte(try c.takeByte());
|
||||
const table = try c.takeLenPrefixed();
|
||||
const pkBytes = try c.takeLenPrefixed();
|
||||
const fieldsBytes = try c.takeLenPrefixed();
|
||||
list.appendAssumeCapacity(.{ .verb = verb, .table = table, .pkBytes = pkBytes, .fieldsBytes = fieldsBytes });
|
||||
}
|
||||
return list.toOwnedSlice();
|
||||
}
|
||||
|
||||
/// Encodes the response to a successful `txn_op`: how many operations were
|
||||
/// committed - always every entry in the request, since a partial commit is
|
||||
/// impossible (`error.Conflict` aborts the whole transaction instead, and
|
||||
/// surfaces as `Status.conflict` with no payload, the same shape as
|
||||
/// `user_admin`'s `UserExists` conflict).
|
||||
pub fn encodeTxnResult(allocator: Allocator, applied: u32) ![]u8 {
|
||||
var buf = std.ArrayList(u8).init(allocator);
|
||||
errdefer buf.deinit();
|
||||
const w = encode.ByteWriter.init(&buf);
|
||||
try w.putU32(applied);
|
||||
return buf.toOwnedSlice();
|
||||
}
|
||||
|
||||
pub fn decodeTxnResult(payload: []const u8) !u32 {
|
||||
var c = Cursor{ .data = payload };
|
||||
return c.takeU32();
|
||||
}
|
||||
|
||||
// --- scan / prefix ---
|
||||
|
||||
pub const ScanArgs = struct {
|
||||
start: ?[]const u8 = null,
|
||||
end: ?[]const u8 = null,
|
||||
limit: u32 = 0,
|
||||
reverse: bool = false,
|
||||
};
|
||||
|
||||
const scanFlagHasStart: u8 = 1 << 0;
|
||||
const scanFlagHasEnd: u8 = 1 << 1;
|
||||
const scanFlagReverse: u8 = 1 << 2;
|
||||
|
||||
pub fn encodeScan(allocator: Allocator, args: ScanArgs) ![]u8 {
|
||||
var buf = std.ArrayList(u8).init(allocator);
|
||||
errdefer buf.deinit();
|
||||
const w = encode.ByteWriter.init(&buf);
|
||||
var flags: u8 = 0;
|
||||
if (args.start != null) flags |= scanFlagHasStart;
|
||||
if (args.end != null) flags |= scanFlagHasEnd;
|
||||
if (args.reverse) flags |= scanFlagReverse;
|
||||
try w.byte(flags);
|
||||
if (args.start) |s| try putLenPrefixed(w, s);
|
||||
if (args.end) |e| try putLenPrefixed(w, e);
|
||||
try w.putU32(args.limit);
|
||||
return buf.toOwnedSlice();
|
||||
}
|
||||
|
||||
pub fn decodeScan(payload: []const u8) !ScanArgs {
|
||||
var c = Cursor{ .data = payload };
|
||||
const flags = try c.takeByte();
|
||||
var args = ScanArgs{};
|
||||
if (flags & scanFlagHasStart != 0) args.start = try c.takeLenPrefixed();
|
||||
if (flags & scanFlagHasEnd != 0) args.end = try c.takeLenPrefixed();
|
||||
args.reverse = flags & scanFlagReverse != 0;
|
||||
args.limit = try c.takeU32();
|
||||
return args;
|
||||
}
|
||||
|
||||
pub fn encodePrefix(allocator: Allocator, prefix: []const u8, limit: u32) ![]u8 {
|
||||
var buf = std.ArrayList(u8).init(allocator);
|
||||
errdefer buf.deinit();
|
||||
const w = encode.ByteWriter.init(&buf);
|
||||
try w.putU32(limit);
|
||||
try w.slice(prefix);
|
||||
return buf.toOwnedSlice();
|
||||
}
|
||||
|
||||
pub const PrefixArgs = struct { prefix: []const u8, limit: u32 };
|
||||
|
||||
pub fn decodePrefix(payload: []const u8) !PrefixArgs {
|
||||
var c = Cursor{ .data = payload };
|
||||
const limit = try c.takeU32();
|
||||
const prefix = c.data[c.pos..];
|
||||
return .{ .prefix = prefix, .limit = limit };
|
||||
}
|
||||
|
||||
// --- entry lists (scan / prefix responses) ---
|
||||
|
||||
pub fn encodeEntries(allocator: Allocator, entries: []const dbmod.Entry) ![]u8 {
|
||||
var buf = std.ArrayList(u8).init(allocator);
|
||||
errdefer buf.deinit();
|
||||
const w = encode.ByteWriter.init(&buf);
|
||||
try w.putU32(@intCast(entries.len));
|
||||
for (entries) |e| {
|
||||
try putLenPrefixed(w, e.key);
|
||||
try putLenPrefixed(w, e.value);
|
||||
}
|
||||
return buf.toOwnedSlice();
|
||||
}
|
||||
|
||||
pub const OwnedEntry = struct {
|
||||
key: []u8,
|
||||
value: []u8,
|
||||
|
||||
pub fn deinit(self: *OwnedEntry, allocator: Allocator) void {
|
||||
allocator.free(self.key);
|
||||
allocator.free(self.value);
|
||||
}
|
||||
};
|
||||
|
||||
/// Copies every key and value out of `payload` so the caller can free
|
||||
/// the frame that carried them and keep the entries alive independently.
|
||||
pub fn decodeEntries(allocator: Allocator, payload: []const u8) ![]OwnedEntry {
|
||||
var c = Cursor{ .data = payload };
|
||||
const count = try c.takeU32();
|
||||
var list = try std.ArrayList(OwnedEntry).initCapacity(allocator, count);
|
||||
errdefer {
|
||||
for (list.items) |*e| e.deinit(allocator);
|
||||
list.deinit();
|
||||
}
|
||||
var i: u32 = 0;
|
||||
while (i < count) : (i += 1) {
|
||||
const key = try c.takeLenPrefixed();
|
||||
const value = try c.takeLenPrefixed();
|
||||
const ownedKey = try allocator.dupe(u8, key);
|
||||
errdefer allocator.free(ownedKey);
|
||||
const ownedValue = try allocator.dupe(u8, value);
|
||||
list.appendAssumeCapacity(.{ .key = ownedKey, .value = ownedValue });
|
||||
}
|
||||
return list.toOwnedSlice();
|
||||
}
|
||||
|
||||
pub fn freeOwnedEntries(allocator: Allocator, entries: []OwnedEntry) void {
|
||||
for (entries) |*e| e.deinit(allocator);
|
||||
allocator.free(entries);
|
||||
}
|
||||
|
||||
// --- batch ---
|
||||
|
||||
pub const BatchOp = union(enum) {
|
||||
put: struct { key: []const u8, value: []const u8 },
|
||||
del: []const u8,
|
||||
};
|
||||
|
||||
const batchKindPut: u8 = 0;
|
||||
const batchKindDel: u8 = 1;
|
||||
|
||||
Reference in new issue
Block a user