From e06f1a622d3617ef97a1f18ed017b14f7fd349c1 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Tue, 8 Sep 2026 17:09:00 +0200 Subject: [PATCH] feat: add unsubscribe and destroySubscriber to pubsub --- src/pubsub.zig | 159 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) diff --git a/src/pubsub.zig b/src/pubsub.zig index 3b43d22..ddd6755 100644 --- a/src/pubsub.zig +++ b/src/pubsub.zig @@ -309,3 +309,162 @@ pub const Hub = struct { } } + /// Removes every name in `channelNames` from `id`'s subscription set + /// without disturbing any channel it stays registered on, so a client + /// that subscribed to several channels can drop a subset from a SEPARATE + /// connection - the control channel `handleSubscribe`'s doc comment + /// says a live subscribe connection can never multiplex further + /// requests onto itself. `channelNames` naming a channel `id` was never + /// on is harmless and simply does not count toward the returned total + /// (matching `publish`-to-an-empty-channel's "no-op, not an error" + /// posture). Returns `error.UnknownSubscriber` if `id` names no + /// currently-live subscriber - already closed (a normal disconnect + /// race, not a caller bug) or never issued (a caller typo) look + /// identical from here, both correctly rejected as "there is nothing + /// left to unsubscribe." + /// + /// If this empties `id`'s entire channel set, the subscriber is closed + /// outright (`sub.close()`, unregistered from `byId`) rather than left + /// alive with nothing left to ever receive - the same terminal state a + /// disconnect reaches through `closeSubscriber`, just reached from the + /// opposite direction. `handleSubscribe`'s push loop notices this + /// exactly like any other close: `sub.waitMessage()` returns `null`, the + /// loop ends, and the connection's own `closeSubscriber`/ + /// `destroySubscriber` pair runs as usual - by then `sub.channels` is + /// already empty, so that second `closeSubscriber` call's unregister + /// loop simply does nothing (idempotent, like every other double-close + /// path in this file). + pub fn unsubscribe(self: *Hub, id: u64, channelNames: []const []const u8) !usize { + self.mutex.lock(); + defer self.mutex.unlock(); + const sub = self.byId.get(id) orelse return error.UnknownSubscriber; + + var kept = try std.ArrayList([]u8).initCapacity(self.allocator, sub.channels.len); + errdefer kept.deinit(); + var removed: usize = 0; + for (sub.channels) |name| { + var matches = false; + for (channelNames) |target| { + if (std.mem.eql(u8, name, target)) { + matches = true; + break; + } + } + if (!matches) { + kept.appendAssumeCapacity(name); + continue; + } + removed += 1; + 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); + } + } + } + self.allocator.free(name); + } + self.allocator.free(sub.channels); + sub.channels = try kept.toOwnedSlice(); + + if (sub.channels.len == 0) { + _ = self.byId.remove(id); + sub.close(); + } + return removed; + } + + /// Frees `sub`'s own memory (mailbox contents, channel-name copies, the + /// struct itself). Must only be called after `closeSubscriber(sub)` has + /// already run and every thread that might still call `sub.waitMessage`/ + /// `sub.push` (directly or through `publish`) has stopped touching it - + /// `closeSubscriber` unregisters `sub` from `Hub.channels` precisely so + /// `publish` can never hand this function a dangling pointer to race + /// against. + pub fn destroySubscriber(self: *Hub, sub: *Subscriber) void { + sub.deinitContents(); + self.allocator.destroy(sub); + } + + /// Number of subscribers currently registered under `channel` (0 if it + /// has none, including a channel name nobody has ever used). Mainly for + /// tests and metrics - `publish` recomputes its own snapshot rather than + /// calling this, to avoid taking `self.mutex` twice per publish. + pub fn channelSubscriberCount(self: *Hub, channel: []const u8) usize { + self.mutex.lock(); + defer self.mutex.unlock(); + if (self.channels.get(channel)) |list| return list.items.len; + return 0; + } + + /// Delivers an independent copy of `payload` to every subscriber + /// currently registered under `channel`, returning how many subscribers + /// it was handed to (0 if the channel currently has none - not an + /// error, matching real pub/sub semantics where publishing to an empty + /// channel is a routine no-op, not a failure). Never blocks on a slow + /// subscriber - see `Subscriber.push`. `allocator` is used only for the + /// short-lived snapshot of subscriber pointers, not the per-subscriber + /// message copies, which are always allocated from `self.allocator` + /// (each subscriber frees its own mailbox contents independently, so + /// they must not depend on a caller-supplied allocator outliving this + /// call). + pub fn publish(self: *Hub, allocator: Allocator, channel: []const u8, payload: []const u8) !usize { + self.mutex.lock(); + errdefer self.mutex.unlock(); + const subs = if (self.channels.get(channel)) |list| list.items else &[_]*Subscriber{}; + // Snapshot the subscriber list under the lock, but do the actual + // per-subscriber copy+push OUTSIDE it: `Subscriber.push` takes its + // own independent mutex, never `self.mutex`, so holding this one + // across every subscriber's push would serialize every publish + // behind every subscriber's own lock contention for no reason. + const snapshot = try allocator.alloc(*Subscriber, subs.len); + defer allocator.free(snapshot); + @memcpy(snapshot, subs); + self.mutex.unlock(); + + for (snapshot) |sub| { + const channelCopy = self.allocator.dupe(u8, channel) catch continue; + const payloadCopy = self.allocator.dupe(u8, payload) catch { + self.allocator.free(channelCopy); + continue; + }; + sub.push(.{ .channel = channelCopy, .payload = payloadCopy }); + } + return snapshot.len; + } +}; + +fn destroy(hub: *Hub, sub: *Subscriber) void { + hub.closeSubscriber(sub); + hub.destroySubscriber(sub); +} + +test "publish delivers a copy to a single subscriber on the channel it named" { + const allocator = std.testing.allocator; + var hub = Hub.init(allocator); + defer hub.deinit(); + + const sub = try hub.subscribe(&.{"weather"}, defaultCapacity); + defer destroy(&hub, sub); + + const delivered = try hub.publish(allocator, "weather", "sunny"); + try std.testing.expectEqual(@as(usize, 1), delivered); + + const msg = sub.waitMessage().?; + defer msg.deinit(allocator); + try std.testing.expectEqualStrings("weather", msg.channel); + try std.testing.expectEqualStrings("sunny", msg.payload); +} + +test "publish to a channel with no subscribers returns 0 without crashing" { + const allocator = std.testing.allocator; + var hub = Hub.init(allocator); + defer hub.deinit(); +