62 lines
1.9 KiB
TypeScript
62 lines
1.9 KiB
TypeScript
import { test, expect } from "bun:test";
|
|
import { createPubSub, createResilientPubSub, PresenceChannel } from "../src/index.ts";
|
|
|
|
test("publish reaches exact + wildcard subscribers", async () => {
|
|
const bus = createPubSub();
|
|
const got: string[] = [];
|
|
bus.subscribe("order:created", (m: { id: number }) => {
|
|
got.push(`exact:${m.id}`);
|
|
});
|
|
bus.subscribe("order:*", (_m, topic) => {
|
|
got.push(`prefix:${topic}`);
|
|
});
|
|
bus.subscribe("*", (_m, topic) => {
|
|
got.push(`all:${topic}`);
|
|
});
|
|
|
|
await bus.publish("order:created", { id: 7 });
|
|
await bus.publish("user:login", { id: 1 });
|
|
|
|
expect(got).toContain("exact:7");
|
|
expect(got).toContain("prefix:order:created");
|
|
expect(got).toContain("all:order:created");
|
|
expect(got).toContain("all:user:login");
|
|
expect(got).not.toContain("prefix:user:login"); // order:* doesn't match user:*
|
|
});
|
|
|
|
test("unsubscribe stops delivery", async () => {
|
|
const bus = createPubSub();
|
|
let n = 0;
|
|
const off = bus.subscribe("t", () => {
|
|
n++;
|
|
});
|
|
await bus.publish("t", 1);
|
|
off();
|
|
await bus.publish("t", 2);
|
|
expect(n).toBe(1);
|
|
});
|
|
|
|
test("publish awaits async handlers and close rejects new work", async () => {
|
|
const bus = createPubSub();
|
|
let completed = false;
|
|
bus.subscribe("task", async () => {
|
|
await Promise.resolve();
|
|
completed = true;
|
|
});
|
|
await bus.publish("task", {});
|
|
expect(completed).toBe(true);
|
|
await bus.close();
|
|
await bus.close();
|
|
await expect(bus.publish("task", {})).rejects.toThrow("WRN-PUBSUB-CLOSED");
|
|
expect(() => bus.subscribe("task", () => {})).toThrow("WRN-PUBSUB-CLOSED");
|
|
});
|
|
|
|
test("resilient pubsub validates retry and presence settings", () => {
|
|
const driver = {
|
|
publish: async () => {},
|
|
subscribe: () => () => {},
|
|
};
|
|
expect(() => createResilientPubSub(driver, { retries: -1 })).toThrow("retries");
|
|
expect(() => new PresenceChannel(0)).toThrow("ttlMs");
|
|
});
|