feat: add unsubscribe and destroySubscriber to pubsub

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();
+