1import { assert, describe, it } from "@effect/vitest"
2import { Effect, Exit, Fiber, Option, Scope, TxPubSub, TxQueue } from "effect"
3
4describe("TxPubSub", () => {
5 describe("constructors", () => {
6 it.effect("bounded creates pubsub with specified capacity", () =>
7 Effect.gen(function*() {
8 const hub = yield* Effect.tx(TxPubSub.bounded<number>(16))
9 assert.strictEqual(TxPubSub.capacity(hub), 16)
10 assert.strictEqual(yield* Effect.tx(TxPubSub.isShutdown(hub)), false)
11 }))
12
13 it.effect("unbounded creates pubsub with unlimited capacity", () =>
14 Effect.gen(function*() {
15 const hub = yield* Effect.tx(TxPubSub.unbounded<string>())
16 assert.strictEqual(TxPubSub.capacity(hub), Number.POSITIVE_INFINITY)
17 }))
18
19 it.effect("dropping creates pubsub with dropping strategy", () =>
20 Effect.gen(function*() {
21 const hub = yield* Effect.tx(TxPubSub.dropping<number>(4))
22 assert.strictEqual(TxPubSub.capacity(hub), 4)
23 }))
24
25 it.effect("sliding creates pubsub with sliding strategy", () =>
26 Effect.gen(function*() {
27 const hub = yield* Effect.tx(TxPubSub.sliding<number>(4))
28 assert.strictEqual(TxPubSub.capacity(hub), 4)
29 }))
30 })
31
32 describe("publish and subscribe", () => {
33 it.effect("publish with no subscribers is a no-op", () =>
34 Effect.gen(function*() {
35 const hub = yield* Effect.tx(TxPubSub.unbounded<string>())
36 const result = yield* Effect.tx(TxPubSub.publish(hub, "hello"))
37 assert.strictEqual(result, true)
38 }))
39
40 it.effect("subscribe then publish then take", () =>
41 Effect.gen(function*() {
42 const hub = yield* Effect.tx(TxPubSub.unbounded<string>())
43
44 yield* Effect.scoped(
45 Effect.gen(function*() {
46 const sub = yield* TxPubSub.subscribe(hub)
47 yield* Effect.tx(TxPubSub.publish(hub, "hello"))
48 const msg = yield* Effect.tx(TxQueue.take(sub))
49 assert.strictEqual(msg, "hello")
50 })
51 )
52 }))
53
54 it.effect("multiple subscribers each receive all messages", () =>
55 Effect.gen(function*() {
56 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
57
58 yield* Effect.scoped(
59 Effect.gen(function*() {
60 const sub1 = yield* TxPubSub.subscribe(hub)
61 const sub2 = yield* TxPubSub.subscribe(hub)
62
63 yield* Effect.tx(TxPubSub.publish(hub, 1))
64 yield* Effect.tx(TxPubSub.publish(hub, 2))
65
66 const v1a = yield* Effect.tx(TxQueue.take(sub1))
67 const v1b = yield* Effect.tx(TxQueue.take(sub1))
68 const v2a = yield* Effect.tx(TxQueue.take(sub2))
69 const v2b = yield* Effect.tx(TxQueue.take(sub2))
70
71 assert.strictEqual(v1a, 1)
72 assert.strictEqual(v1b, 2)
73 assert.strictEqual(v2a, 1)
74 assert.strictEqual(v2b, 2)
75 })
76 )
77 }))
78
79 it.effect("publishAll delivers all messages", () =>
80 Effect.gen(function*() {
81 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
82
83 yield* Effect.scoped(
84 Effect.gen(function*() {
85 const sub = yield* TxPubSub.subscribe(hub)
86 yield* Effect.tx(TxPubSub.publishAll(hub, [1, 2, 3]))
87
88 const v1 = yield* Effect.tx(TxQueue.take(sub))
89 const v2 = yield* Effect.tx(TxQueue.take(sub))
90 const v3 = yield* Effect.tx(TxQueue.take(sub))
91 assert.strictEqual(v1, 1)
92 assert.strictEqual(v2, 2)
93 assert.strictEqual(v3, 3)
94 })
95 )
96 }))
97
98 it.effect("publishAll preserves a one-shot iterable across retries", () =>
99 Effect.gen(function*() {
100 const hub = yield* TxPubSub.bounded<number>(1)
101
102 yield* Effect.scoped(
103 Effect.gen(function*() {
104 const sub = yield* TxPubSub.subscribe(hub)
105 yield* TxPubSub.publish(hub, 1)
106
107 let iterations = 0
108 const hasIterated = () => iterations > 0
109 const values = (function*() {
110 iterations++
111 yield 2
112 })()
113 const fiber = yield* Effect.forkChild(TxPubSub.publishAll(hub, values))
114 while (!hasIterated()) {
115 yield* Effect.yieldNow
116 }
117
118 assert.strictEqual(yield* TxQueue.take(sub), 1)
119 assert.strictEqual(yield* Fiber.join(fiber), true)
120 assert.deepStrictEqual(yield* TxQueue.poll(sub), Option.some(2))
121 })
122 )
123 }))
124
125 it.effect("subscriber only receives messages published after subscription", () =>
126 Effect.gen(function*() {
127 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
128
129 // Publish before subscription
130 yield* Effect.tx(TxPubSub.publish(hub, 1))
131
132 yield* Effect.scoped(
133 Effect.gen(function*() {
134 const sub = yield* TxPubSub.subscribe(hub)
135 yield* Effect.tx(TxPubSub.publish(hub, 2))
136 yield* Effect.tx(TxPubSub.publish(hub, 3))
137
138 const v1 = yield* Effect.tx(TxQueue.take(sub))
139 const v2 = yield* Effect.tx(TxQueue.take(sub))
140 assert.strictEqual(v1, 2)
141 assert.strictEqual(v2, 3)
142 })
143 )
144 }))
145 })
146
147 describe("strategies", () => {
148 it.effect("bounded publisher blocks when subscriber queue is full", () =>
149 Effect.gen(function*() {
150 const hub = yield* Effect.tx(TxPubSub.bounded<number>(2))
151
152 yield* Effect.scoped(
153 Effect.gen(function*() {
154 const sub = yield* TxPubSub.subscribe(hub)
155
156 yield* Effect.tx(TxPubSub.publish(hub, 1))
157 yield* Effect.tx(TxPubSub.publish(hub, 2))
158
159 // Publisher should block since subscriber queue is full
160 const fiber = yield* Effect.forkChild(Effect.tx(TxPubSub.publish(hub, 3)))
161
162 // Take one to make space, unblocking the publisher
163 const v1 = yield* Effect.tx(TxQueue.take(sub))
164 assert.strictEqual(v1, 1)
165
166 // Publisher should now complete
167 yield* Fiber.join(fiber)
168
169 const v2 = yield* Effect.tx(TxQueue.take(sub))
170 const v3 = yield* Effect.tx(TxQueue.take(sub))
171 assert.strictEqual(v2, 2)
172 assert.strictEqual(v3, 3)
173 })
174 )
175 }))
176
177 it.effect("dropping strategy drops messages when subscriber queue is full", () =>
178 Effect.gen(function*() {
179 const hub = yield* Effect.tx(TxPubSub.dropping<number>(2))
180
181 yield* Effect.scoped(
182 Effect.gen(function*() {
183 const sub = yield* TxPubSub.subscribe(hub)
184
185 yield* Effect.tx(TxPubSub.publish(hub, 1))
186 yield* Effect.tx(TxPubSub.publish(hub, 2))
187 const accepted = yield* Effect.tx(TxPubSub.publish(hub, 3)) // should be dropped
188 assert.strictEqual(accepted, false)
189
190 const v1 = yield* Effect.tx(TxQueue.take(sub))
191 const v2 = yield* Effect.tx(TxQueue.take(sub))
192 assert.strictEqual(v1, 1)
193 assert.strictEqual(v2, 2)
194 })
195 )
196 }))
197
198 it.effect("sliding strategy drops oldest messages when subscriber queue is full", () =>
199 Effect.gen(function*() {
200 const hub = yield* Effect.tx(TxPubSub.sliding<number>(2))
201
202 yield* Effect.scoped(
203 Effect.gen(function*() {
204 const sub = yield* TxPubSub.subscribe(hub)
205
206 yield* Effect.tx(TxPubSub.publish(hub, 1))
207 yield* Effect.tx(TxPubSub.publish(hub, 2))
208 yield* Effect.tx(TxPubSub.publish(hub, 3)) // evicts 1
209
210 const v1 = yield* Effect.tx(TxQueue.take(sub))
211 const v2 = yield* Effect.tx(TxQueue.take(sub))
212 assert.strictEqual(v1, 2)
213 assert.strictEqual(v2, 3)
214 })
215 )
216 }))
217
218 it.effect("unbounded strategy always accepts", () =>
219 Effect.gen(function*() {
220 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
221
222 yield* Effect.scoped(
223 Effect.gen(function*() {
224 const sub = yield* TxPubSub.subscribe(hub)
225
226 for (let i = 0; i < 100; i++) {
227 const accepted = yield* Effect.tx(TxPubSub.publish(hub, i))
228 assert.strictEqual(accepted, true)
229 }
230
231 for (let i = 0; i < 100; i++) {
232 const v = yield* Effect.tx(TxQueue.take(sub))
233 assert.strictEqual(v, i)
234 }
235 })
236 )
237 }))
238 })
239
240 describe("getters", () => {
241 it.effect("size returns max subscriber queue size", () =>
242 Effect.gen(function*() {
243 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
244
245 assert.strictEqual(yield* Effect.tx(TxPubSub.size(hub)), 0)
246
247 yield* Effect.scoped(
248 Effect.gen(function*() {
249 const sub = yield* TxPubSub.subscribe(hub)
250 yield* Effect.tx(TxPubSub.publish(hub, 1))
251 yield* Effect.tx(TxPubSub.publish(hub, 2))
252 assert.strictEqual(yield* Effect.tx(TxPubSub.size(hub)), 2)
253
254 yield* Effect.tx(TxQueue.take(sub))
255 assert.strictEqual(yield* Effect.tx(TxPubSub.size(hub)), 1)
256 })
257 )
258 }))
259
260 it.effect("isEmpty and isNonEmpty reflect subscriber queues", () =>
261 Effect.gen(function*() {
262 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
263 assert.strictEqual(yield* Effect.tx(TxPubSub.isEmpty(hub)), true)
264 assert.strictEqual(yield* Effect.tx(TxPubSub.isNonEmpty(hub)), false)
265
266 yield* Effect.scoped(
267 Effect.gen(function*() {
268 const sub = yield* TxPubSub.subscribe(hub)
269
270 yield* Effect.tx(TxPubSub.publish(hub, 1))
271 assert.strictEqual(yield* Effect.tx(TxPubSub.isEmpty(hub)), false)
272 assert.strictEqual(yield* Effect.tx(TxPubSub.isNonEmpty(hub)), true)
273
274 yield* Effect.tx(TxQueue.take(sub))
275 assert.strictEqual(yield* Effect.tx(TxPubSub.isEmpty(hub)), true)
276 assert.strictEqual(yield* Effect.tx(TxPubSub.isNonEmpty(hub)), false)
277 })
278 )
279 }))
280
281 it.effect("isFull checks if any subscriber queue is at capacity", () =>
282 Effect.gen(function*() {
283 const hub = yield* Effect.tx(TxPubSub.bounded<number>(2))
284 assert.strictEqual(yield* Effect.tx(TxPubSub.isFull(hub)), false)
285
286 yield* Effect.scoped(
287 Effect.gen(function*() {
288 yield* TxPubSub.subscribe(hub)
289 yield* Effect.tx(TxPubSub.publish(hub, 1))
290 assert.strictEqual(yield* Effect.tx(TxPubSub.isFull(hub)), false)
291
292 yield* Effect.tx(TxPubSub.publish(hub, 2))
293 assert.strictEqual(yield* Effect.tx(TxPubSub.isFull(hub)), true)
294 })
295 )
296 }))
297 })
298
299 describe("shutdown", () => {
300 it.effect("shutdown prevents further publishes", () =>
301 Effect.gen(function*() {
302 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
303 yield* Effect.tx(TxPubSub.shutdown(hub))
304
305 assert.strictEqual(yield* Effect.tx(TxPubSub.isShutdown(hub)), true)
306 const accepted = yield* Effect.tx(TxPubSub.publish(hub, 1))
307 assert.strictEqual(accepted, false)
308 }))
309
310 it.effect("shutdown is idempotent", () =>
311 Effect.gen(function*() {
312 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
313 yield* Effect.tx(TxPubSub.shutdown(hub))
314 yield* Effect.tx(TxPubSub.shutdown(hub)) // second call should not throw
315 assert.strictEqual(yield* Effect.tx(TxPubSub.isShutdown(hub)), true)
316 }))
317
318 it.effect("awaitShutdown blocks until shutdown", () =>
319 Effect.gen(function*() {
320 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
321
322 const fiber = yield* Effect.forkChild(Effect.tx(TxPubSub.awaitShutdown(hub)))
323 yield* Effect.tx(TxPubSub.shutdown(hub))
324 yield* Fiber.join(fiber)
325 }))
326
327 it.effect("awaitShutdown returns immediately if already shutdown", () =>
328 Effect.gen(function*() {
329 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
330 yield* Effect.tx(TxPubSub.shutdown(hub))
331 yield* Effect.tx(TxPubSub.awaitShutdown(hub)) // should not block
332 }))
333 })
334
335 describe("scope cleanup", () => {
336 it.effect("releases a subscriber after hub shutdown without interruption", () =>
337 Effect.gen(function*() {
338 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
339 const scope = yield* Scope.make()
340 yield* TxPubSub.subscribe(hub).pipe(Scope.provide(scope))
341 yield* Effect.tx(TxPubSub.shutdown(hub))
342
343 const release = yield* Effect.exit(Scope.close(scope, Exit.void))
344 assert(Exit.isSuccess(release))
345 }))
346
347 it.effect("closing scope removes subscriber", () =>
348 Effect.gen(function*() {
349 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
350
351 // Subscribe inside a scope then close it
352 yield* Effect.scoped(
353 Effect.gen(function*() {
354 yield* TxPubSub.subscribe(hub)
355 })
356 )
357
358 // Publish should succeed with no subscribers
359 const result = yield* Effect.tx(TxPubSub.publish(hub, 1))
360 assert.strictEqual(result, true)
361 assert.strictEqual(yield* Effect.tx(TxPubSub.size(hub)), 0)
362 }))
363
364 it.effect("subscriber receives messages only while scope is open", () =>
365 Effect.gen(function*() {
366 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
367 const results: Array<number> = []
368
369 yield* Effect.scoped(
370 Effect.gen(function*() {
371 const sub = yield* TxPubSub.subscribe(hub)
372 yield* Effect.tx(TxPubSub.publish(hub, 1))
373 yield* Effect.tx(TxPubSub.publish(hub, 2))
374 results.push(yield* Effect.tx(TxQueue.take(sub)))
375 results.push(yield* Effect.tx(TxQueue.take(sub)))
376 })
377 )
378
379 // After scope closes, new messages should not accumulate
380 yield* Effect.tx(TxPubSub.publish(hub, 3))
381 assert.strictEqual(yield* Effect.tx(TxPubSub.size(hub)), 0)
382 assert.deepStrictEqual(results, [1, 2])
383 }))
384 })
385
386 describe("guards", () => {
387 it.effect("isTxPubSub returns true for TxPubSub instances", () =>
388 Effect.gen(function*() {
389 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
390 assert.strictEqual(TxPubSub.isTxPubSub(hub), true)
391 assert.strictEqual(TxPubSub.isTxPubSub({}), false)
392 assert.strictEqual(TxPubSub.isTxPubSub(null), false)
393 assert.strictEqual(TxPubSub.isTxPubSub(42), false)
394 }))
395 })
396
397 describe("concurrency", () => {
398 it.effect("concurrent publishers and subscribers", () =>
399 Effect.gen(function*() {
400 const hub = yield* Effect.tx(TxPubSub.unbounded<number>())
401
402 yield* Effect.scoped(
403 Effect.gen(function*() {
404 const sub1 = yield* TxPubSub.subscribe(hub)
405 const sub2 = yield* TxPubSub.subscribe(hub)
406
407 // Fork publishers
408 const f1 = yield* Effect.forkChild(Effect.tx(TxPubSub.publishAll(hub, [1, 2, 3])))
409 const f2 = yield* Effect.forkChild(Effect.tx(TxPubSub.publishAll(hub, [4, 5, 6])))
410
411 yield* Fiber.join(f1)
412 yield* Fiber.join(f2)
413
414 // Each subscriber should get all 6 messages
415 const items1: Array<number> = []
416 const items2: Array<number> = []
417 for (let i = 0; i < 6; i++) {
418 items1.push(yield* Effect.tx(TxQueue.take(sub1)))
419 items2.push(yield* Effect.tx(TxQueue.take(sub2)))
420 }
421
422 assert.strictEqual(items1.length, 6)
423 assert.strictEqual(items2.length, 6)
424 // Both should have the same items (though order across publishers may vary)
425 assert.deepStrictEqual(items1.sort(), [1, 2, 3, 4, 5, 6])
426 assert.deepStrictEqual(items2.sort(), [1, 2, 3, 4, 5, 6])
427 })
428 )
429 }))
430 })
431})