feat: add PubSubMessage and deinit to protocol

This commit is contained in:
dualfroz committed 2026-09-07 21:21:12 +02:00
1 parent 81a41dee0d
commit 5d433ed37b
1 file changed
+159
+159
View File
@@ -1877,3 +1877,162 @@ pub fn decodeUnsubscribeResult(payload: []const u8) !u32 {
return c.takeU32(); return c.takeU32();
} }
/// One pushed message on an already-open `subscribe` connection: which
/// channel it arrived on (a client can subscribe to several at once, so the
/// tag travels with every message) plus the payload bytes. Every frame a
/// subscribed connection receives after its initial ack uses this shape,
/// always with status `.ok` - there is no separate framing marker, because
/// once a client has sent `subscribe` it never sends another request on
/// that connection and so never has to disambiguate a push from a reply to
/// something else it asked (see `server.zig`'s `handleSubscribe`).
pub const PubSubMessage = struct {
channel: []u8,
payload: []u8,
pub fn deinit(self: PubSubMessage, allocator: Allocator) void {
allocator.free(self.channel);
allocator.free(self.payload);
}
};
pub fn encodePubSubMessage(allocator: Allocator, channel: []const u8, payload: []const u8) ![]u8 {
var buf = std.ArrayList(u8).init(allocator);
errdefer buf.deinit();
const w = encode.ByteWriter.init(&buf);
try putLenPrefixed(w, channel);
try w.slice(payload);
return buf.toOwnedSlice();
}
/// Decodes a pushed message into independently owned copies (the frame
/// they arrived in is freed right after decoding, in `Client.nextMessage` -
/// see its doc comment) - free with `PubSubMessage.deinit`.
pub fn decodePubSubMessage(allocator: Allocator, payload: []const u8) !PubSubMessage {
var c = Cursor{ .data = payload };
const channel = try c.takeLenPrefixed();
const message = c.data[c.pos..];
const channelCopy = try allocator.dupe(u8, channel);
errdefer allocator.free(channelCopy);
const payloadCopy = try allocator.dupe(u8, message);
return .{ .channel = channelCopy, .payload = payloadCopy };
}
// --- table explain ---
/// Wire request for a remote `table explain`: a table name and the raw text
/// of an `expr.zig` predicate expression (empty text means "no predicate",
/// the same convention `encodeTableSelect`'s `exprText` uses). No
/// offset/limit - explain reports a plan, not rows, so there is nothing to
/// page.
pub fn encodeTableExplain(allocator: Allocator, table: []const u8, exprText: []const u8) ![]u8 {
var buf = std.ArrayList(u8).init(allocator);
errdefer buf.deinit();
const w = encode.ByteWriter.init(&buf);
try putLenPrefixed(w, table);
try w.slice(exprText);
return buf.toOwnedSlice();
}
pub const TableExplainArgs = struct { table: []const u8, exprText: []const u8 };
pub fn decodeTableExplain(payload: []const u8) !TableExplainArgs {
var c = Cursor{ .data = payload };
const table = try c.takeLenPrefixed();
const exprText = c.data[c.pos..];
return .{ .table = table, .exprText = exprText };
}
/// Encodes a `query.Plan`. `indexName` uses the empty string as its "null"
/// sentinel (an index or field name can never itself be empty - `query.zig`
/// rejects that at `createIndex`/`createCompositeIndex` time) instead of a
/// separate presence byte, the same economy `exprText`'s "empty means no
/// predicate" convention already relies on elsewhere in this file.
pub fn encodePlan(allocator: Allocator, plan: querymod.Plan) ![]u8 {
var buf = std.ArrayList(u8).init(allocator);
errdefer buf.deinit();
const w = encode.ByteWriter.init(&buf);
try w.byte(@intFromEnum(plan.kind));
try putLenPrefixed(w, plan.indexName orelse "");
try w.putU32(@intCast(plan.matchedFields));
try w.putU32(@intCast(plan.totalFields));
return buf.toOwnedSlice();
}
/// Decodes a `query.Plan` - `indexName`, when present, is an owned copy the
/// caller must free with `Plan.deinit` (see `encodePlan`'s doc comment for
/// why an empty string decodes back to `null` instead of `""`).
pub fn decodePlan(allocator: Allocator, payload: []const u8) !querymod.Plan {
var c = Cursor{ .data = payload };
const kindByte = try c.takeByte();
const kind = std.meta.intToEnum(querymod.PlanKind, kindByte) catch return error.Malformed;
const rawName = try c.takeLenPrefixed();
const indexName = if (rawName.len == 0) null else try allocator.dupe(u8, rawName);
errdefer if (indexName) |n| allocator.free(n);
const matchedFields = try c.takeU32();
const totalFields = try c.takeU32();
return .{ .kind = kind, .indexName = indexName, .matchedFields = matchedFields, .totalFields = totalFields };
}
// --- table update ---
/// Wire request for a remote `table update`: a table name, the `SET`
/// clause's field list, and the raw text of an `expr.zig` predicate
/// expression (empty text updates every row, the same convention
/// `encodeTableSelect`'s `exprText` uses). The field list reuses
/// `Row.encode` exactly like `encodeTxnOp`'s per-entry `fields` does,
/// instead of inventing a second name+value framing just for this opcode -
/// and, like `encodeTableSelect`'s `orderBy`, sits BEFORE `exprText`
/// because `exprText` is decoded as "everything left in the payload",
/// so anything meant to survive decoding has to come first.
pub fn encodeTableUpdate(allocator: Allocator, table: []const u8, sets: []const rowmod.Field, exprText: []const u8) ![]u8 {
var buf = std.ArrayList(u8).init(allocator);
errdefer buf.deinit();
const w = encode.ByteWriter.init(&buf);
try putLenPrefixed(w, table);
var setsRow = rowmod.Row.init(allocator);
defer setsRow.deinit();
for (sets) |f| try setsRow.set(f.name, f.value);
const setsBytes = try setsRow.encode(allocator);
defer allocator.free(setsBytes);
try putLenPrefixed(w, setsBytes);
try w.slice(exprText);
return buf.toOwnedSlice();
}
pub const TableUpdateArgs = struct { table: []const u8, setsBytes: []const u8, exprText: []const u8 };
/// `table` and `exprText` are borrowed slices into `payload`, the same
/// convention `decodeTableExplain` uses. `setsBytes` is also borrowed and
/// still needs `rowmod.Row.decode` to become a real field list - kept
/// undecoded here so a malformed table name fails fast without touching
/// row-decoding at all, the same ownership split `DecodedTxnOpEntry` uses
/// for its own `fieldsBytes`.
pub fn decodeTableUpdate(payload: []const u8) !TableUpdateArgs {
var c = Cursor{ .data = payload };
const table = try c.takeLenPrefixed();
const setsBytes = try c.takeLenPrefixed();
const exprText = c.data[c.pos..];
return .{ .table = table, .setsBytes = setsBytes, .exprText = exprText };
}
// --- table delete ---
/// Wire request for a remote `table delete`: a table name and the raw text
/// of an `expr.zig` predicate expression (empty text deletes every row, the
/// same convention `encodeTableSelect`/`encodeTableExplain`'s `exprText`
/// uses). Identical wire shape to `encodeTableExplain` - both are just
/// "table + predicate", no `SET` clause like `encodeTableUpdate` needs - but
/// kept as its own named pair rather than reused directly, the same
/// per-opcode-gets-its-own-function precedent every other opcode here
/// follows even when two shapes coincide.
pub fn encodeTableDelete(allocator: Allocator, table: []const u8, exprText: []const u8) ![]u8 {
var buf = std.ArrayList(u8).init(allocator);
errdefer buf.deinit();
const w = encode.ByteWriter.init(&buf);
try putLenPrefixed(w, table);
try w.slice(exprText);
return buf.toOwnedSlice();
}