diff --git a/src/pubsub.zig b/src/pubsub.zig index ddd6755..18652b5 100644 --- a/src/pubsub.zig +++ b/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); +} +