From 6d310ef01a024af20a7fca265adfa2b4faa7876d Mon Sep 17 00:00:00 2001 From: DualFroz Date: Tue, 8 Sep 2026 16:53:58 +0200 Subject: [PATCH] feat: add Hub and deinit to pubsub --- src/pubsub.zig | 158 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/src/pubsub.zig b/src/pubsub.zig index 4b297e8..3b43d22 100644 --- a/src/pubsub.zig +++ b/src/pubsub.zig @@ -151,3 +151,161 @@ pub const Subscriber = struct { } }; +/// The channel registry every `Subscriber` is filed under. Completely +/// independent of `Db` - publish/subscribe never touches storage, so a +/// `Hub`'s own mutex is never held anywhere near `Db`'s (see `server.zig`'s +/// `dispatch`, which handles the `subscribe`/`publish` opcodes before it +/// ever reaches for `Server.mutex`, so a slow or permanently-open +/// subscriber connection can never stall a read or write against the +/// store). +pub const Hub = struct { + allocator: Allocator, + mutex: std.Thread.Mutex = .{}, + channels: std.StringHashMap(std.ArrayList(*Subscriber)), + /// Every currently-live subscriber, keyed by the `id` `subscribe` + /// assigned it - lets `unsubscribe` (see its doc comment) look one up by + /// the bare number a client saw in its subscribe ack, without needing + /// the original `*Subscriber` pointer, which never crosses a second TCP + /// connection. Entries are removed exactly when `closeSubscriber` runs + /// (or `unsubscribe` empties a subscriber's channel set and closes it + /// itself), so a stale or already-closed id reliably misses here rather + /// than risking a dangling pointer. + byId: std.AutoHashMap(u64, *Subscriber), + nextId: u64 = 0, + + pub fn init(allocator: Allocator) Hub { + return .{ + .allocator = allocator, + .channels = std.StringHashMap(std.ArrayList(*Subscriber)).init(allocator), + .byId = std.AutoHashMap(u64, *Subscriber).init(allocator), + }; + } + + /// Frees every remaining channel's subscriber list and key copy, and the + /// id registry. Does NOT touch any `Subscriber` still registered - + /// callers must `closeSubscriber` (which also unregisters, so in an + /// orderly shutdown this map is already empty by the time `deinit` runs) + /// and `destroySubscriber` every subscriber they created first, the same + /// "caller drains, then frees the container" ordering `Db`/`Server` + /// already use elsewhere in this codebase. + pub fn deinit(self: *Hub) void { + var it = self.channels.iterator(); + while (it.next()) |entry| { + entry.value_ptr.deinit(); + self.allocator.free(entry.key_ptr.*); + } + self.channels.deinit(); + self.byId.deinit(); + } + + /// Creates a new subscriber registered under every name in + /// `channelNames` (duplicate names collapse to one registration, so a + /// message published on a channel a caller listed twice is still only + /// ever delivered once). `capacity` bounds its mailbox (see + /// `Subscriber.push`). Returns a pointer the caller must eventually + /// pass to `closeSubscriber` then `destroySubscriber`, in that order. + pub fn subscribe(self: *Hub, channelNames: []const []const u8, capacity: usize) !*Subscriber { + var unique = std.ArrayList([]const u8).init(self.allocator); + defer unique.deinit(); + for (channelNames) |name| { + var seen = false; + for (unique.items) |u| { + if (std.mem.eql(u8, u, name)) { + seen = true; + break; + } + } + if (!seen) try unique.append(name); + } + + const owned = try self.allocator.alloc([]u8, unique.items.len); + var ownedCount: usize = 0; + errdefer { + for (owned[0..ownedCount]) |c| self.allocator.free(c); + self.allocator.free(owned); + } + for (unique.items, 0..) |name, i| { + owned[i] = try self.allocator.dupe(u8, name); + ownedCount += 1; + } + + const sub = try self.allocator.create(Subscriber); + errdefer self.allocator.destroy(sub); + sub.* = Subscriber.init(self.allocator, capacity, owned); + + self.mutex.lock(); + defer self.mutex.unlock(); + + const id = self.nextId; + self.nextId += 1; + sub.id = id; + try self.byId.put(id, sub); + errdefer _ = self.byId.remove(id); + + var registered: usize = 0; + errdefer { + for (owned[0..registered]) |name| { + if (self.channels.getPtr(name)) |list| { + for (list.items, 0..) |s, idx| { + if (s == sub) { + _ = list.swapRemove(idx); + break; + } + } + } + } + } + for (owned) |name| { + const gop = try self.channels.getOrPut(name); + if (!gop.found_existing) { + gop.key_ptr.* = self.allocator.dupe(u8, name) catch |err| { + // The map entry was already inserted with `name`'s + // pointer as a placeholder key by `getOrPut` itself; + // remove it before propagating so `deinit`/`subscribe` + // never iterate over a key pointing into memory this + // function is about to free via `owned`'s errdefer. + _ = self.channels.remove(name); + return err; + }; + gop.value_ptr.* = std.ArrayList(*Subscriber).init(self.allocator); + } + try gop.value_ptr.append(sub); + registered += 1; + } + + return sub; + } + + /// Stops delivering new messages to `sub` and unregisters it from every + /// channel it was subscribed to, wakes any thread blocked in its + /// `waitMessage`, and drops a channel's map entry entirely once its + /// subscriber list becomes empty (an unsubscribed-from channel is a + /// completely normal, permanent steady state, not a reason to keep an + /// empty list around forever). Safe to call more than once (from a + /// watcher thread and a failed publish both noticing the same dead + /// connection, say) - both `Subscriber.close` and the unregister loop + /// below are no-ops the second time. Does NOT free `sub` itself - see + /// `destroySubscriber`. + pub fn closeSubscriber(self: *Hub, sub: *Subscriber) void { + sub.close(); + self.mutex.lock(); + defer self.mutex.unlock(); + _ = self.byId.remove(sub.id); + for (sub.channels) |name| { + if (self.channels.getPtr(name)) |list| { + for (list.items, 0..) |s, idx| { + if (s == sub) { + _ = list.swapRemove(idx); + break; + } + } + if (list.items.len == 0) { + if (self.channels.fetchRemove(name)) |kv| { + kv.value.deinit(); + self.allocator.free(kv.key); + } + } + } + } + } +