feat: add Waiter and run to pubsub

src/pubsub.zig

@@ -468,3 +468,163 @@ test "publish to a channel with no subscribers returns 0 without crashing" {
     var hub = Hub.init(allocator);
     defer hub.deinit();
 
+    const delivered = try hub.publish(allocator, "nobody-home", "hello");
+    try std.testing.expectEqual(@as(usize, 0), delivered);
+}
+
+test "every subscriber on a channel gets its own independent copy" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const a = try hub.subscribe(&.{"orders"}, defaultCapacity);
+    defer destroy(&hub, a);
+    const b = try hub.subscribe(&.{"orders"}, defaultCapacity);
+    defer destroy(&hub, b);
+
+    const delivered = try hub.publish(allocator, "orders", "order-42");
+    try std.testing.expectEqual(@as(usize, 2), delivered);
+
+    const ma = a.waitMessage().?;
+    defer ma.deinit(allocator);
+    const mb = b.waitMessage().?;
+    defer mb.deinit(allocator);
+    try std.testing.expectEqualStrings("order-42", ma.payload);
+    try std.testing.expectEqualStrings("order-42", mb.payload);
+    // Independent allocations, not aliased - freeing one must not affect
+    // the other's still-valid pointer.
+    try std.testing.expect(ma.payload.ptr != mb.payload.ptr);
+}
+
+test "a subscriber on several channels gets messages tagged with the right channel" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{ "a", "b" }, defaultCapacity);
+    defer destroy(&hub, sub);
+
+    try std.testing.expectEqual(@as(usize, 1), try hub.publish(allocator, "b", "from-b"));
+    try std.testing.expectEqual(@as(usize, 1), try hub.publish(allocator, "a", "from-a"));
+
+    const first = sub.waitMessage().?;
+    defer first.deinit(allocator);
+    try std.testing.expectEqualStrings("b", first.channel);
+    try std.testing.expectEqualStrings("from-b", first.payload);
+
+    const second = sub.waitMessage().?;
+    defer second.deinit(allocator);
+    try std.testing.expectEqualStrings("a", second.channel);
+    try std.testing.expectEqualStrings("from-a", second.payload);
+}
+
+test "subscribing to the same channel name twice still delivers only once" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{ "dup", "dup" }, defaultCapacity);
+    defer destroy(&hub, sub);
+
+    try std.testing.expectEqual(@as(usize, 1), try hub.publish(allocator, "dup", "x"));
+    const msg = sub.waitMessage().?;
+    defer msg.deinit(allocator);
+    // Only one message was ever queued - a second `waitMessage` in a
+    // non-blocking check would hang, so instead assert there is exactly
+    // one by closing right after and expecting a drained-then-null result.
+    sub.close();
+    try std.testing.expect(sub.waitMessage() == null);
+}
+
+test "closing a subscriber unregisters it - a later publish no longer reaches it" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    hub.closeSubscriber(sub);
+
+    try std.testing.expectEqual(@as(usize, 0), try hub.publish(allocator, "ch", "too-late"));
+    try std.testing.expectEqual(@as(usize, 0), hub.channelSubscriberCount("ch"));
+
+    hub.destroySubscriber(sub);
+}
+
+test "waitMessage drains already-queued messages before returning null on close" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    defer hub.destroySubscriber(sub);
+
+    _ = try hub.publish(allocator, "ch", "one");
+    _ = try hub.publish(allocator, "ch", "two");
+    hub.closeSubscriber(sub);
+
+    const m1 = sub.waitMessage().?;
+    defer m1.deinit(allocator);
+    try std.testing.expectEqualStrings("one", m1.payload);
+    const m2 = sub.waitMessage().?;
+    defer m2.deinit(allocator);
+    try std.testing.expectEqualStrings("two", m2.payload);
+    try std.testing.expect(sub.waitMessage() == null);
+}
+
+test "a full mailbox drops the oldest message, not the newest" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, 2);
+    defer destroy(&hub, sub);
+
+    _ = try hub.publish(allocator, "ch", "1");
+    _ = try hub.publish(allocator, "ch", "2");
+    _ = try hub.publish(allocator, "ch", "3");
+    try std.testing.expectEqual(@as(usize, 1), sub.droppedCount());
+
+    const m1 = sub.waitMessage().?;
+    defer m1.deinit(allocator);
+    try std.testing.expectEqualStrings("2", m1.payload);
+    const m2 = sub.waitMessage().?;
+    defer m2.deinit(allocator);
+    try std.testing.expectEqualStrings("3", m2.payload);
+}
+
+test "capacity zero drops every message without ever panicking" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, 0);
+    defer destroy(&hub, sub);
+
+    _ = try hub.publish(allocator, "ch", "gone");
+    _ = try hub.publish(allocator, "ch", "also-gone");
+    try std.testing.expectEqual(@as(usize, 2), sub.droppedCount());
+    sub.close();
+    try std.testing.expect(sub.waitMessage() == null);
+}
+
+test "closing a subscriber wakes a thread already blocked in waitMessage" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    defer hub.destroySubscriber(sub);
+
+    const Waiter = struct {
+        fn run(s: *Subscriber, result: *?Message) void {
+            result.* = s.waitMessage();
+        }
+    };
+    var result: ?Message = undefined;
+    result = Message{ .channel = &.{}, .payload = &.{} };
+    const t = try std.Thread.spawn(.{}, Waiter.run, .{ sub, &result });
+    hub.closeSubscriber(sub);
+    t.join();
+    try std.testing.expect(result == null);
+}
+