diff --git a/src/protocol.zig b/src/protocol.zig index ff60c92..f85ed71 100644 --- a/src/protocol.zig +++ b/src/protocol.zig @@ -1717,3 +1717,163 @@ pub const SubscribeArgs = struct { } }; +/// Decodes a `subscribe` request into independently owned channel-name +/// copies - free with `SubscribeArgs.deinit`. Owned (rather than borrowed +/// slices into `payload`, as most other decoders here return) because +/// `handleSubscribe` hands these straight to `pubsub.Hub.subscribe`, which +/// keeps its own copies anyway - decoding once into memory that outlives +/// the request frame avoids a second, redundant copy inside the hub for +/// every subscribe call. +pub fn decodeSubscribe(allocator: Allocator, payload: []const u8) !SubscribeArgs { + var c = Cursor{ .data = payload }; + const count = try c.takeU32(); + var list = try std.ArrayList([]u8).initCapacity(allocator, count); + errdefer { + for (list.items) |ch| allocator.free(ch); + list.deinit(); + } + var i: u32 = 0; + while (i < count) : (i += 1) { + const name = try c.takeLenPrefixed(); + const owned = try allocator.dupe(u8, name); + list.appendAssumeCapacity(owned); + } + return .{ .channels = try list.toOwnedSlice() }; +} + +/// A `subscribe` request's one and only reply frame (status `.ok`, this as +/// its payload) - sent once the connection is actually registered in the +/// `pubsub.Hub`, so a client that has received this ack is guaranteed not +/// to miss a message published right after it returns. Carries the +/// (deduplicated) channel count purely as a sanity-checkable echo, the same +/// spirit as `encodeAuthOk` echoing a token back - plus the `pubsub.Hub`- +/// assigned subscriber id, the only way a later `unsubscribe` request (on a +/// SEPARATE connection - this one is receive-only from here on, see +/// `server.zig`'s `handleSubscribe`) can ever name this subscription again. +pub fn encodeSubscribeAck(allocator: Allocator, subscriberId: u64, channelCount: u32) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU64(subscriberId); + try w.putU32(channelCount); + return buf.toOwnedSlice(); +} + +pub const SubscribeAck = struct { subscriberId: u64, channelCount: u32 }; + +pub fn decodeSubscribeAck(payload: []const u8) !SubscribeAck { + var c = Cursor{ .data = payload }; + const subscriberId = try c.takeU64(); + const channelCount = try c.takeU32(); + return .{ .subscriberId = subscriberId, .channelCount = channelCount }; +} + +/// Wire request for `publish`: a channel name and the message bytes (binary +/// safe, "rest of the payload" like `encodeSet`'s value). Any connection +/// may publish regardless of whether it has ever subscribed to anything - +/// publishing and subscribing are independent roles on the same opcode +/// family, exactly like a `Db` writer and a `Db` reader are independent +/// roles over the same store. +pub fn encodePublish(allocator: Allocator, channel: []const u8, message: []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(message); + return buf.toOwnedSlice(); +} + +pub const PublishArgs = struct { channel: []const u8, message: []const u8 }; + +pub fn decodePublish(payload: []const u8) !PublishArgs { + var c = Cursor{ .data = payload }; + const channel = try c.takeLenPrefixed(); + const message = c.data[c.pos..]; + return .{ .channel = channel, .message = message }; +} + +/// A `publish` request's reply payload: how many subscribers `pubsub.Hub` +/// handed the message to (see `Hub.publish`'s doc comment - 0 is a normal, +/// non-error outcome for a channel nobody is listening to). +pub fn encodePublishResult(allocator: Allocator, subscriberCount: u64) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU64(subscriberCount); + return buf.toOwnedSlice(); +} + +pub fn decodePublishResult(payload: []const u8) !u64 { + var c = Cursor{ .data = payload }; + return c.takeU64(); +} + +/// Wire request for `unsubscribe`: the `pubsub.Hub`-assigned subscriber id +/// (from `encodeSubscribeAck`) plus the list of channel names to drop from +/// that subscription - the same channel-list shape `encodeSubscribe` uses, +/// with the id prefixed. Unlike `subscribe`, this is an ordinary +/// request/response opcode (see `server.zig`'s `handleUnsubscribe`): it is +/// always sent on a DIFFERENT connection than the one that subscribed, +/// since that connection became receive-only the moment its subscribe ack +/// arrived and can never send another request on itself again. +pub fn encodeUnsubscribe(allocator: Allocator, subscriberId: u64, channels: []const []const u8) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU64(subscriberId); + try w.putU32(@intCast(channels.len)); + for (channels) |c| try putLenPrefixed(w, c); + return buf.toOwnedSlice(); +} + +pub const UnsubscribeArgs = struct { + subscriberId: u64, + channels: [][]u8, + + pub fn deinit(self: UnsubscribeArgs, allocator: Allocator) void { + for (self.channels) |c| allocator.free(c); + allocator.free(self.channels); + } +}; + +/// Decodes an `unsubscribe` request - same owned-copy rationale as +/// `decodeSubscribe`: `handleUnsubscribe` hands the channel names straight +/// to `pubsub.Hub.unsubscribe`, which only ever borrows them for the +/// duration of that call, so decoding into memory this function's caller +/// frees (via `UnsubscribeArgs.deinit`) avoids a redundant second copy. +pub fn decodeUnsubscribe(allocator: Allocator, payload: []const u8) !UnsubscribeArgs { + var c = Cursor{ .data = payload }; + const subscriberId = try c.takeU64(); + const count = try c.takeU32(); + var list = try std.ArrayList([]u8).initCapacity(allocator, count); + errdefer { + for (list.items) |ch| allocator.free(ch); + list.deinit(); + } + var i: u32 = 0; + while (i < count) : (i += 1) { + const name = try c.takeLenPrefixed(); + const owned = try allocator.dupe(u8, name); + list.appendAssumeCapacity(owned); + } + return .{ .subscriberId = subscriberId, .channels = try list.toOwnedSlice() }; +} + +/// An `unsubscribe` request's reply payload: how many of the requested +/// channel names were actually removed from the named subscriber's set (see +/// `pubsub.Hub.unsubscribe`'s doc comment on why naming a channel it was +/// never on is a harmless no-op that simply does not count here, not an +/// error). +pub fn encodeUnsubscribeResult(allocator: Allocator, removedCount: u32) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU32(removedCount); + return buf.toOwnedSlice(); +} + +pub fn decodeUnsubscribeResult(payload: []const u8) !u32 { + var c = Cursor{ .data = payload }; + return c.takeU32(); +} +