1import { assert, describe, it } from "@effect/vitest"
2import { Array, Effect, Exit, Fiber, Latch, MutableList, PubSub, Scheduler, Scope, Stream } from "effect"
3import { pipe } from "effect/Function"
4
5describe("PubSub", () => {
6 it.effect("isPubSub type guard", () =>
7 Effect.gen(function*() {
8 const pubsub = yield* PubSub.bounded<string>(10)
9
10 assert.isTrue(PubSub.isPubSub(pubsub))
11 assert.isFalse(PubSub.isPubSub({}))
12 assert.isFalse(PubSub.isPubSub(null))
13 }))
14
15 for (const batch of [false, true]) {
16 it.effect.each([1, 2, 3])(
17 `sliding capacity %s delivers each retained message once per subscriber (batch: ${batch})`,
18 (capacity) =>
19 Effect.gen(function*() {
20 const pubsub = yield* PubSub.sliding<number>(capacity)
21 const first = yield* PubSub.subscribe(pubsub)
22 const second = yield* PubSub.subscribe(pubsub)
23 const values = Array.range(1, capacity + 2)
24 const retained = values.slice(-capacity)
25 yield* PubSub.publishAll(pubsub, values)
26
27 const received = batch
28 ? yield* PubSub.takeAll(first)
29 : yield* Effect.forEach(retained, () => PubSub.take(first))
30 const duplicate = yield* PubSub.takeUpTo(first, capacity)
31 const other = yield* PubSub.takeUpTo(second, capacity)
32
33 assert.deepStrictEqual([received, duplicate, other], [retained, [], retained])
34 })
35 )
36 }
37
38 it.effect.each([1, 2, 3])(
39 "unsubscribing after a slide preserves the other subscriber's messages (capacity: %s)",
40 (capacity) =>
41 Effect.gen(function*() {
42 const pubsub = yield* PubSub.sliding<number>(capacity)
43 const scope = yield* Scope.make()
44 const first = yield* PubSub.subscribe(pubsub).pipe(Scope.provide(scope))
45 const second = yield* PubSub.subscribe(pubsub)
46 const values = Array.range(1, capacity + 2)
47 const retained = values.slice(-capacity)
48 yield* PubSub.publishAll(pubsub, values)
49 yield* PubSub.takeAll(first)
50 yield* Scope.close(scope, Exit.void)
51
52 assert.deepStrictEqual(yield* PubSub.takeUpTo(second, capacity), retained)
53 })
54 )
55
56 it.effect("publishAll - capacity 2 (BoundedPubSubPow2)", () => {
57 const messages = [1, 2]
58 return PubSub.bounded<number>(2).pipe(
59 Effect.flatMap((pubsub) =>
60 Effect.scoped(
61 Effect.gen(function*() {
62 const sub1 = yield* PubSub.subscribe(pubsub)
63 const sub2 = yield* PubSub.subscribe(pubsub)
64 yield* PubSub.publishAll(pubsub, messages)
65 const takes1 = yield* PubSub.takeAll(sub1)
66 const takes2 = yield* PubSub.takeAll(sub2)
67 assert.deepStrictEqual(takes1, messages)
68 assert.deepStrictEqual(takes2, messages)
69 })
70 )
71 )
72 )
73 })
74 it.effect("publishAll - capacity 4 (BoundedPubSubPow2)", () => {
75 const messages = [1, 2]
76 return PubSub.bounded<number>(4).pipe(
77 Effect.flatMap((pubsub) =>
78 Effect.scoped(
79 Effect.gen(function*() {
80 const sub1 = yield* PubSub.subscribe(pubsub)
81 const sub2 = yield* PubSub.subscribe(pubsub)
82 yield* PubSub.publishAll(pubsub, messages)
83 const takes1 = yield* PubSub.takeAll(sub1)
84 const takes2 = yield* PubSub.takeAll(sub2)
85 assert.deepStrictEqual(takes1, messages)
86 assert.deepStrictEqual(takes2, messages)
87 })
88 )
89 )
90 )
91 })
92 it.effect("publishAll - capacity 3 (BoundedPubSubArb)", () => {
93 const messages = [1, 2]
94 return PubSub.bounded<number>(3).pipe(
95 Effect.flatMap((pubsub) =>
96 Effect.scoped(
97 Effect.gen(function*() {
98 const sub1 = yield* PubSub.subscribe(pubsub)
99 const sub2 = yield* PubSub.subscribe(pubsub)
100 yield* PubSub.publishAll(pubsub, messages)
101 const takes1 = yield* PubSub.takeAll(sub1)
102 const takes2 = yield* PubSub.takeAll(sub2)
103 assert.deepStrictEqual(takes1, messages)
104 assert.deepStrictEqual(takes2, messages)
105 })
106 )
107 )
108 )
109 })
110 it.effect("sequential publishers and subscribers with one publisher and one subscriber", () =>
111 Effect.gen(function*() {
112 const values = Array.range(0, 9)
113 const latch = yield* Latch.make()
114 const pubsub = yield* PubSub.bounded<number>(10)
115 const subscriber = yield* PubSub.subscribe(pubsub).pipe(
116 Effect.flatMap((subscription) =>
117 latch.await.pipe(
118 Effect.andThen(Effect.forEach(values, () => PubSub.take(subscription)))
119 )
120 ),
121 Effect.scoped,
122 Effect.forkChild({ startImmediately: true })
123 )
124 yield* PubSub.publishAll(pubsub, values)
125 yield* latch.open
126 const result = yield* Fiber.join(subscriber)
127 assert.deepStrictEqual(result, values)
128 }))
129 it.effect("sequential publishers and subscribers with one publisher and two subscribers", () =>
130 Effect.gen(function*() {
131 const values = Array.range(0, 9)
132 const latch = yield* Latch.make()
133 const pubsub = yield* PubSub.bounded<number>(10)
134 const subscriber1 = yield* pubsub.pipe(
135 PubSub.subscribe,
136 Effect.flatMap((subscription) =>
137 pipe(
138 latch.await,
139 Effect.andThen(pipe(values, Effect.forEach(() => PubSub.take(subscription))))
140 )
141 ),
142 Effect.scoped,
143 Effect.forkChild({ startImmediately: true })
144 )
145 const subscriber2 = yield* pubsub.pipe(
146 PubSub.subscribe,
147 Effect.flatMap((subscription) =>
148 pipe(
149 latch.await,
150 Effect.andThen(pipe(values, Effect.forEach(() => PubSub.take(subscription))))
151 )
152 ),
153 Effect.scoped,
154 Effect.forkChild({ startImmediately: true })
155 )
156 yield* PubSub.publishAll(pubsub, values)
157 yield* latch.open
158 const result1 = yield* Fiber.join(subscriber1)
159 const result2 = yield* Fiber.join(subscriber2)
160 assert.deepStrictEqual(result1, values)
161 assert.deepStrictEqual(result2, values)
162 }))
163 it.effect("backpressured concurrent publishers and subscribers - one to one", () =>
164 Effect.gen(function*() {
165 const values = Array.range(0, 64)
166 const pubsub = yield* PubSub.bounded<number>(64)
167 const subscriber = yield* PubSub.subscribe(pubsub).pipe(
168 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
169 Effect.scoped,
170 Effect.forkChild({ startImmediately: true })
171 )
172 yield* Effect.forkChild(PubSub.publishAll(pubsub, values))
173 const result = yield* Fiber.join(subscriber)
174 assert.deepStrictEqual(result, values)
175 }))
176 it.effect("backpressured concurrent publishers and subscribers - one to many", () =>
177 Effect.gen(function*() {
178 const values = Array.range(0, 64)
179 const pubsub = yield* PubSub.bounded<number>(64)
180 const subscriber1 = yield* PubSub.subscribe(pubsub).pipe(
181 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
182 Effect.scoped,
183 Effect.forkChild({ startImmediately: true })
184 )
185 const subscriber2 = yield* PubSub.subscribe(pubsub).pipe(
186 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
187 Effect.scoped,
188 Effect.forkChild({ startImmediately: true })
189 )
190 yield* Effect.forkChild(PubSub.publishAll(pubsub, values))
191 const result1 = yield* Fiber.join(subscriber1)
192 const result2 = yield* Fiber.join(subscriber2)
193 assert.deepStrictEqual(result1, values)
194 assert.deepStrictEqual(result2, values)
195 }))
196 it.effect("backpressured concurrent publishers and subscribers - many to many", () =>
197 Effect.gen(function*() {
198 const values = Array.range(1, 64)
199 const pubsub = yield* PubSub.bounded<number>(64 * 2)
200 const subscriber1 = yield* PubSub.subscribe(pubsub).pipe(
201 Effect.flatMap((subscription) => Effect.forEach(values.concat(values), (_) => PubSub.take(subscription))),
202 Effect.scoped,
203 Effect.forkChild({ startImmediately: true })
204 )
205 const subscriber2 = yield* PubSub.subscribe(pubsub).pipe(
206 Effect.flatMap((subscription) => Effect.forEach(values.concat(values), (_) => PubSub.take(subscription))),
207 Effect.scoped,
208 Effect.forkChild({ startImmediately: true })
209 )
210 const fiber = yield* Effect.forkChild(PubSub.publishAll(pubsub, values))
211 yield* Effect.forkChild(PubSub.publishAll(pubsub, Array.map(values, (n) => -n)))
212 const result1 = yield* Fiber.join(subscriber1)
213 const result2 = yield* Fiber.join(subscriber2)
214 yield* Fiber.join(fiber)
215 assert.deepStrictEqual(Array.filter(result1, (n) => n > 0), values)
216 assert.deepStrictEqual(
217 Array.filter(result1, (n) => n < 0),
218 Array.map(values, (n) => -n)
219 )
220 assert.deepStrictEqual(Array.filter(result2, (n) => n > 0), values)
221 assert.deepStrictEqual(
222 Array.filter(result2, (n) => n < 0),
223 Array.map(values, (n) => -n)
224 )
225 }))
226 it.effect("dropping concurrent publishers and subscribers - one to one", () =>
227 Effect.gen(function*() {
228 const values = Array.range(0, 64)
229 const pubsub = yield* PubSub.dropping<number>(64)
230 const subscriber = yield* PubSub.subscribe(pubsub).pipe(
231 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
232 Effect.scoped,
233 Effect.forkChild({ startImmediately: true })
234 )
235 yield* Effect.forkChild(Effect.forEach(values, (n) => PubSub.publish(pubsub, n)))
236 const result = yield* Fiber.join(subscriber)
237 assert.deepStrictEqual(result, values)
238 }))
239 it.effect("dropping concurrent publishers and subscribers - one to many", () =>
240 Effect.gen(function*() {
241 const values = Array.range(0, 64)
242 const pubsub = yield* PubSub.dropping<number>(64)
243 const subscriber1 = yield* PubSub.subscribe(pubsub).pipe(
244 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
245 Effect.scoped,
246 Effect.forkChild({ startImmediately: true })
247 )
248 const subscriber2 = yield* PubSub.subscribe(pubsub).pipe(
249 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
250 Effect.scoped,
251 Effect.forkChild({ startImmediately: true })
252 )
253 yield* Effect.forkChild(Effect.forEach(values, (n) => PubSub.publish(pubsub, n)))
254 const result1 = yield* Fiber.join(subscriber1)
255 const result2 = yield* Fiber.join(subscriber2)
256 assert.deepStrictEqual(result1, values)
257 assert.deepStrictEqual(result2, values)
258 }))
259 it.effect("dropping concurrent publishers and subscribers - many to many", () =>
260 Effect.gen(function*() {
261 const values = Array.range(1, 64)
262 const pubsub = yield* PubSub.dropping<number>(64 * 2)
263 const subscriber1 = yield* PubSub.subscribe(pubsub).pipe(
264 Effect.flatMap((subscription) => Effect.forEach(values.concat(values), (_) => PubSub.take(subscription))),
265 Effect.scoped,
266 Effect.forkChild({ startImmediately: true })
267 )
268 const subscriber2 = yield* PubSub.subscribe(pubsub).pipe(
269 Effect.flatMap((subscription) => Effect.forEach(values.concat(values), (_) => PubSub.take(subscription))),
270 Effect.scoped,
271 Effect.forkChild({ startImmediately: true })
272 )
273 const fiber = yield* Effect.forkChild(Effect.forEach(values, (n) => PubSub.publish(pubsub, n)))
274 yield* Effect.forkChild(Effect.forEach(values, (n) => PubSub.publish(pubsub, -n)))
275 const result1 = yield* Fiber.join(subscriber1)
276 const result2 = yield* Fiber.join(subscriber2)
277 yield* Fiber.join(fiber)
278 assert.deepStrictEqual(pipe(result1, Array.filter((n) => n > 0)), values)
279 assert.deepStrictEqual(
280 pipe(result1, Array.filter((n) => n < 0)),
281 pipe(values, Array.map((n) => -n))
282 )
283 assert.deepStrictEqual(pipe(result2, Array.filter((n) => n > 0)), values)
284 assert.deepStrictEqual(
285 pipe(result2, Array.filter((n) => n < 0)),
286 pipe(values, Array.map((n) => -n))
287 )
288 }))
289 it.effect("sliding concurrent publishers and subscribers - one to one", () =>
290 Effect.gen(function*() {
291 const values = Array.range(0, 64)
292 const pubsub = yield* PubSub.sliding<number>(64)
293 const subscriber = yield* PubSub.subscribe(pubsub).pipe(
294 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
295 Effect.scoped,
296 Effect.forkChild({ startImmediately: true })
297 )
298 yield* Effect.forkChild(Effect.forEach(values, (n) => PubSub.publish(pubsub, n)))
299 const result = yield* Fiber.join(subscriber)
300 assert.deepStrictEqual(result, values)
301 }))
302 it.effect("sliding concurrent publishers and subscribers - one to many", () =>
303 Effect.gen(function*() {
304 const values = Array.range(0, 64)
305 const pubsub = yield* PubSub.sliding<number>(64)
306 const subscriber1 = yield* PubSub.subscribe(pubsub).pipe(
307 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
308 Effect.scoped,
309 Effect.forkChild({ startImmediately: true })
310 )
311 const subscriber2 = yield* PubSub.subscribe(pubsub).pipe(
312 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
313 Effect.scoped,
314 Effect.forkChild({ startImmediately: true })
315 )
316 yield* Effect.forkChild(PubSub.publishAll(pubsub, values))
317 const result1 = yield* Fiber.join(subscriber1)
318 const result2 = yield* Fiber.join(subscriber2)
319 assert.deepStrictEqual(result1, values)
320 assert.deepStrictEqual(result2, values)
321 }))
322 it.effect("sliding concurrent publishers and subscribers - many to many", () =>
323 Effect.gen(function*() {
324 const values = Array.range(1, 64)
325 const pubsub = yield* PubSub.sliding<number>(64 * 2)
326 const subscriber1 = yield* PubSub.subscribe(pubsub).pipe(
327 Effect.flatMap((subscription) => Effect.forEach(values.concat(values), (_) => PubSub.take(subscription))),
328 Effect.scoped,
329 Effect.forkChild({ startImmediately: true })
330 )
331 const subscriber2 = yield* PubSub.subscribe(pubsub).pipe(
332 Effect.flatMap((subscription) => Effect.forEach(values.concat(values), (_) => PubSub.take(subscription))),
333 Effect.scoped,
334 Effect.forkChild({ startImmediately: true })
335 )
336 const fiber = yield* Effect.forkChild(PubSub.publishAll(pubsub, values))
337 yield* Effect.forkChild(PubSub.publishAll(pubsub, Array.map(values, (n) => -n)))
338 const result1 = yield* Fiber.join(subscriber1)
339 const result2 = yield* Fiber.join(subscriber2)
340 yield* Fiber.join(fiber)
341 assert.deepStrictEqual(pipe(result1, Array.filter((n) => n > 0)), values)
342 assert.deepStrictEqual(
343 pipe(result1, Array.filter((n) => n < 0)),
344 pipe(values, Array.map((n) => -n))
345 )
346 assert.deepStrictEqual(pipe(result2, Array.filter((n) => n > 0)), values)
347 assert.deepStrictEqual(
348 pipe(result2, Array.filter((n) => n < 0)),
349 pipe(values, Array.map((n) => -n))
350 )
351 }))
352 it.effect("unbounded concurrent publishers and subscribers - one to one", () =>
353 Effect.gen(function*() {
354 const values = Array.range(0, 64)
355 const pubsub = yield* PubSub.unbounded<number>()
356 const subscriber = yield* PubSub.subscribe(pubsub).pipe(
357 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
358 Effect.scoped,
359 Effect.forkChild({ startImmediately: true })
360 )
361 yield* Effect.forkChild(PubSub.publishAll(pubsub, values))
362
363 const result = yield* Fiber.join(subscriber)
364 assert.deepStrictEqual(result, values)
365 }))
366 it.effect("unbounded concurrent publishers and subscribers - one to many", () =>
367 Effect.gen(function*() {
368 const values = Array.range(0, 64)
369 const pubsub = yield* PubSub.unbounded<number>()
370 const subscriber1 = yield* PubSub.subscribe(pubsub).pipe(
371 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
372 Effect.scoped,
373 Effect.forkChild({ startImmediately: true })
374 )
375 const subscriber2 = yield* PubSub.subscribe(pubsub).pipe(
376 Effect.flatMap((subscription) => Effect.forEach(values, (_) => PubSub.take(subscription))),
377 Effect.scoped,
378 Effect.forkChild({ startImmediately: true })
379 )
380 yield* Effect.forkChild(PubSub.publishAll(pubsub, values))
381 const result1 = yield* Fiber.join(subscriber1)
382 const result2 = yield* Fiber.join(subscriber2)
383 assert.deepStrictEqual(result1, values)
384 assert.deepStrictEqual(result2, values)
385 }))
386 it.effect("unbounded concurrent publishers and subscribers - many to many", () =>
387 Effect.gen(function*() {
388 const values = Array.range(1, 64)
389 const pubsub = yield* PubSub.unbounded<number>()
390 const subscriber1 = yield* PubSub.subscribe(pubsub).pipe(
391 Effect.flatMap((subscription) => Effect.forEach(values.concat(values), (_) => PubSub.take(subscription))),
392 Effect.scoped,
393 Effect.forkChild({ startImmediately: true })
394 )
395
396 const subscriber2 = yield* PubSub.subscribe(pubsub).pipe(
397 Effect.flatMap((subscription) => Effect.forEach(values.concat(values), (_) => PubSub.take(subscription))),
398 Effect.scoped,
399 Effect.forkChild({ startImmediately: true })
400 )
401 const fiber = yield* Effect.forkChild(PubSub.publishAll(pubsub, values))
402 yield* Effect.forkChild(PubSub.publishAll(pubsub, Array.map(values, (n) => -n)))
403 const result1 = yield* Fiber.join(subscriber1)
404 const result2 = yield* Fiber.join(subscriber2)
405 yield* Fiber.join(fiber)
406 assert.deepStrictEqual(Array.filter(result1, (n) => n > 0), values)
407 assert.deepStrictEqual(
408 Array.filter(result1, (n) => n < 0),
409 Array.map(values, (n) => -n)
410 )
411 assert.deepStrictEqual(Array.filter(result2, (n) => n > 0), values)
412 assert.deepStrictEqual(
413 Array.filter(result2, (n) => n < 0),
414 Array.map(values, (n) => -n)
415 )
416 }))
417
418 it.effect("infinite capacity behaves as unbounded", () =>
419 Effect.gen(function*() {
420 const values = Array.range(0, 64)
421 for (const make of [PubSub.bounded, PubSub.dropping, PubSub.sliding]) {
422 const pubsub = yield* make<number>({ capacity: Infinity, replay: 2 })
423 const subscription = yield* PubSub.subscribe(pubsub)
424 yield* PubSub.publishAll(pubsub, values)
425 assert.deepStrictEqual(yield* PubSub.takeAll(subscription), values)
426 const late = yield* PubSub.subscribe(pubsub)
427 assert.deepStrictEqual(yield* PubSub.takeAll(late), [63, 64])
428 }
429 }))
430
431 it.effect("null values", () => {
432 const messages = [1, null]
433 return PubSub.unbounded<number | null>().pipe(
434 Effect.flatMap((pubsub) =>
435 Effect.scoped(
436 Effect.gen(function*() {
437 const sub1 = yield* PubSub.subscribe(pubsub)
438 const sub2 = yield* PubSub.subscribe(pubsub)
439 yield* PubSub.publishAll(pubsub, messages)
440 const takes1 = yield* PubSub.takeAll(sub1)
441 const takes2 = yield* PubSub.takeAll(sub2)
442 assert.deepStrictEqual([...takes1], messages)
443 assert.deepStrictEqual([...takes2], messages)
444 })
445 )
446 )
447 )
448 })
449
450 it.effect("publish does not increase size while no subscribers", () =>
451 Effect.gen(function*() {
452 const pubsub = yield* PubSub.dropping<number>(2)
453 yield* PubSub.publish(pubsub, 1)
454 yield* PubSub.publish(pubsub, 2)
455 assert.deepStrictEqual(PubSub.sizeUnsafe(pubsub), 0)
456 }))
457
458 it.effect("publishAll does not increase size while no subscribers", () =>
459 Effect.gen(function*() {
460 const pubsub = yield* PubSub.dropping<number>(2)
461 yield* PubSub.publishAll(pubsub, [1, 2])
462 assert.deepStrictEqual(PubSub.sizeUnsafe(pubsub), 0)
463 }))
464
465 it("normalizes low-level polling and replay counts", () => {
466 const implementations = [
467 PubSub.makeAtomicBounded<number>(1),
468 PubSub.makeAtomicBounded<number>(3),
469 PubSub.makeAtomicBounded<number>(4),
470 PubSub.makeAtomicUnbounded<number>()
471 ]
472
473 for (const pubsub of implementations) {
474 const subscription = pubsub.subscribe()
475 pubsub.publishAll([1, 2, 3])
476 assert.deepStrictEqual(subscription.pollUpTo(1.9), [1])
477 assert.deepStrictEqual(subscription.pollUpTo(Number.NaN), [])
478 }
479
480 const replayPubSub = PubSub.makeAtomicUnbounded<number>({ replay: 3 })
481 replayPubSub.publishAll([1, 2, 3])
482 const replay = replayPubSub.replayWindow()
483 assert.deepStrictEqual(replay.takeN(1.9), [1])
484 assert.deepStrictEqual(replay.takeN(Number.NaN), [])
485 assert.deepStrictEqual(replay.takeAll(), [2, 3])
486 })
487
488 it("polls MutableList.Empty as a value", () => {
489 const pubsub = PubSub.makeAtomicBounded<symbol>(1)
490 const subscription = pubsub.subscribe()
491 pubsub.publish(MutableList.Empty)
492
493 assert.deepStrictEqual(subscription.pollUpTo(1), [MutableList.Empty])
494 })
495
496 it("publishes after a capacity-one subscriber unsubscribes from a slid message", () => {
497 const pubsub = PubSub.makeAtomicBounded<number>(1)
498 const subscription = pubsub.subscribe()
499 pubsub.publish(1)
500 pubsub.slide()
501 subscription.unsubscribe()
502
503 const next = pubsub.subscribe()
504 assert.isTrue(pubsub.publish(2))
505 assert.strictEqual(pubsub.size(), 1)
506 assert.deepStrictEqual(next.pollUpTo(1), [2])
507 })
508
509 describe("replay", () => {
510 it("does not retain values published after the replay window is drained", () => {
511 const pubsub = PubSub.makeAtomicUnbounded<object>({ replay: 1 })
512 pubsub.publish({})
513 const replayWindow = pubsub.replayWindow()
514 replayWindow.take()
515
516 const slidOut = {}
517 pubsub.publish(slidOut)
518 pubsub.publish({})
519
520 assert.isFalse(retains(replayWindow, slidOut))
521 })
522
523 it("does not retain values published outside an undrained replay window", () => {
524 const pubsub = PubSub.makeAtomicUnbounded<object>({ replay: 1 })
525 const replayed = {}
526 pubsub.publish(replayed)
527 const replayWindow = pubsub.replayWindow()
528
529 const slidOut = {}
530 pubsub.publish(slidOut)
531 pubsub.publish({})
532
533 assert.isFalse(retains(replayWindow, slidOut))
534 assert.strictEqual(replayWindow.take(), replayed)
535 })
536
537 it("preserves replay order across multiple slides", () => {
538 const pubsub = PubSub.makeAtomicBounded<number>({ capacity: 4, replay: 3 })
539 pubsub.publishAll([1, 2, 3, 4, 5])
540 const subscription = pubsub.subscribe()
541 const replayWindow = pubsub.replayWindow()
542 pubsub.publishAll([6, 7, 8, 9])
543 for (const value of [10, 11, 12]) {
544 pubsub.slide()
545 pubsub.publish(value)
546 }
547
548 assert.deepStrictEqual(replayWindow.takeAll(), [6, 7, 8])
549 assert.deepStrictEqual(subscription.pollUpTo(Number.POSITIVE_INFINITY), [9, 10, 11, 12])
550 })
551
552 it.effect("unbounded", () =>
553 Effect.gen(function*() {
554 const messages = [1, 2, 3, 4, 5]
555 const pubsub = yield* PubSub.unbounded<number>({ replay: 3 })
556 yield* PubSub.publishAll(pubsub, messages)
557 const sub = yield* PubSub.subscribe(pubsub)
558 assert.deepStrictEqual(yield* PubSub.takeAll(sub), [3, 4, 5])
559 }))
560
561 it.effect("unbounded rounds up fractional replay", () =>
562 Effect.gen(function*() {
563 const pubsub = yield* PubSub.unbounded<number>({ replay: 1.5 })
564 yield* PubSub.publishAll(pubsub, [1, 2, 3, 4])
565 const sub = yield* PubSub.subscribe(pubsub)
566 assert.deepStrictEqual(yield* PubSub.takeAll(sub), [3, 4])
567 }))
568
569 it.effect("unbounded disables non-positive replay", () =>
570 Effect.gen(function*() {
571 const pubsub = yield* PubSub.unbounded<number>({ replay: -1 })
572 yield* PubSub.publishAll(pubsub, [1, 2])
573 const sub = yield* PubSub.subscribe(pubsub)
574 assert.deepStrictEqual(yield* PubSub.takeUpTo(sub, 2), [])
575 }))
576
577 it.effect("unbounded takeUpTo", () => {
578 const messages = [1, 2, 3, 4, 5]
579 return PubSub.unbounded<number>({ replay: 3 }).pipe(
580 Effect.flatMap((pubsub) =>
581 Effect.scoped(
582 Effect.gen(function*() {
583 yield* PubSub.publishAll(pubsub, messages)
584
585 const sub1 = yield* PubSub.subscribe(pubsub)
586 yield* PubSub.publish(pubsub, 6)
587 const sub2 = yield* PubSub.subscribe(pubsub)
588
589 assert.strictEqual(yield* PubSub.remaining(sub1), 4)
590 assert.strictEqual(yield* PubSub.remaining(sub2), 3)
591 assert.deepStrictEqual(yield* PubSub.takeUpTo(sub1, 2), [3, 4])
592 assert.deepStrictEqual(yield* PubSub.takeUpTo(sub1, 2), [5, 6])
593 assert.deepStrictEqual(yield* PubSub.takeUpTo(sub2, 3), [4, 5, 6])
594 })
595 )
596 )
597 )
598 })
599
600 it.effect("takeUpTo and takeBetween normalize message counts", () =>
601 Effect.gen(function*() {
602 const livePubSub = yield* PubSub.unbounded<number>()
603 const live = yield* PubSub.subscribe(livePubSub)
604 yield* PubSub.publishAll(livePubSub, [1, 2, 3])
605 assert.deepStrictEqual(yield* PubSub.takeUpTo(live, 1.9), [1])
606
607 const betweenPubSub = yield* PubSub.unbounded<number>()
608 const between = yield* PubSub.subscribe(betweenPubSub)
609 yield* PubSub.publishAll(betweenPubSub, [1, 2, 3])
610 assert.deepStrictEqual(yield* PubSub.takeBetween(between, 1.9, 2.9), [1, 2])
611
612 const replayPubSub = yield* PubSub.unbounded<number>({ replay: 3 })
613 yield* PubSub.publishAll(replayPubSub, [1, 2, 3])
614 const replay = yield* PubSub.subscribe(replayPubSub)
615 assert.deepStrictEqual(yield* PubSub.takeUpTo(replay, 1.9), [1])
616 assert.deepStrictEqual(yield* PubSub.takeUpTo(replay, Number.NaN), [])
617 assert.deepStrictEqual(yield* PubSub.takeAll(replay), [2, 3])
618 }))
619
620 it.effect("dropping", () =>
621 Effect.gen(function*() {
622 const messages = [1, 2, 3, 4, 5]
623 const pubsub = yield* PubSub.dropping<number>({ capacity: 2, replay: 3 })
624
625 yield* PubSub.publishAll(pubsub, messages)
626 const sub = yield* PubSub.subscribe(pubsub)
627 assert.deepStrictEqual(yield* PubSub.takeAll(sub), [3, 4, 5])
628 yield* PubSub.publishAll(pubsub, [6, 7])
629 assert.deepStrictEqual(yield* PubSub.takeAll(sub), [6, 7])
630
631 const sub2 = yield* PubSub.subscribe(pubsub)
632 assert.deepStrictEqual(yield* PubSub.takeAll(sub2), [5, 6, 7])
633
634 yield* PubSub.publishAll(pubsub, [8, 9, 10, 11])
635 assert.deepStrictEqual(yield* PubSub.takeAll(sub), [8, 9])
636 assert.deepStrictEqual(yield* PubSub.takeAll(sub2), [8, 9])
637
638 const sub3 = yield* PubSub.subscribe(pubsub)
639 assert.deepStrictEqual(yield* PubSub.takeAll(sub3), [7, 8, 9])
640 }))
641
642 it.effect("sliding", () =>
643 Effect.gen(function*() {
644 const messages = [1, 2, 3, 4, 5]
645 const pubsub = yield* PubSub.sliding<number>({ capacity: 4, replay: 3 })
646
647 yield* PubSub.publishAll(pubsub, messages)
648 const sub = yield* PubSub.subscribe(pubsub)
649 assert.deepStrictEqual(yield* PubSub.take(sub), 3)
650 yield* PubSub.publishAll(pubsub, [6, 7, 8, 9, 10])
651 assert.deepStrictEqual(yield* PubSub.takeAll(sub), [5, 6, 7, 8, 9, 10])
652
653 const sub2 = yield* PubSub.subscribe(pubsub)
654 assert.deepStrictEqual(yield* PubSub.takeAll(sub2), [8, 9, 10])
655
656 yield* PubSub.publishAll(pubsub, [11, 12, 13, 14, 15, 16])
657 assert.deepStrictEqual(yield* PubSub.takeAll(sub), [13, 14, 15, 16])
658 assert.deepStrictEqual(yield* PubSub.takeAll(sub2), [13, 14, 15, 16])
659
660 const sub3 = yield* PubSub.subscribe(pubsub)
661 assert.deepStrictEqual(yield* PubSub.takeAll(sub3), [14, 15, 16])
662 }))
663
664 it.effect("sliding preserves publish order with a lagging subscriber", () =>
665 Effect.gen(function*() {
666 const pubsub = yield* PubSub.sliding<number>({ capacity: 4, replay: 3 })
667 yield* PubSub.subscribe(pubsub)
668 yield* PubSub.publishAll(pubsub, [1, 2])
669 const subscription = yield* PubSub.subscribe(pubsub)
670 yield* PubSub.publishAll(pubsub, [3, 4, 5])
671
672 const values = yield* PubSub.takeAll(subscription)
673 assert.isTrue(values.every((value, index) => index === 0 || values[index - 1] <= value))
674 }))
675 })
676
677 it.effect("shutdown interrupts suspended subscribers", () =>
678 Effect.gen(function*() {
679 const pubsub = yield* PubSub.unbounded<number>()
680 const subscription = yield* PubSub.subscribe(pubsub)
681 const fiber = yield* Effect.forkChild(PubSub.take(subscription), { startImmediately: true })
682
683 yield* PubSub.shutdown(pubsub)
684
685 const exit = yield* Fiber.await(fiber)
686 assert.isTrue(Exit.hasInterrupts(exit!))
687 }))
688
689 it.effect("publish succeeds after interrupting a suspended subscriber", () =>
690 Effect.gen(function*() {
691 const pubsub = yield* PubSub.dropping<number>(1)
692 const subscription = yield* PubSub.subscribe(pubsub)
693 const fiber = yield* Effect.forkChild(PubSub.take(subscription), { startImmediately: true })
694
695 yield* Fiber.interrupt(fiber)
696
697 assert.isTrue(yield* PubSub.publish(pubsub, 42))
698 assert.strictEqual(yield* PubSub.take(subscription), 42)
699 }))
700
701 // A small op budget makes the fiber yield at each point between starting a
702 // wait and suspending on it; interrupting it there must leave nothing behind.
703 // (With a budget below 3 a fiber yields again before making progress.)
704 const interruptAtEveryYield = (
705 test: (
706 budget: (effect: Effect.Effect<unknown>) => Effect.Effect<unknown>
707 ) => Effect.Effect<boolean, never, Scope.Scope>
708 ) =>
709 Effect.gen(function*() {
710 const failed: Array<number> = []
711 for (let ops = 3; ops <= 64; ops++) {
712 const ok = yield* Effect.scoped(test(Effect.provideService(Scheduler.MaxOpsBeforeYield, ops)))
713 if (!ok) failed.push(ops)
714 }
715 assert.deepStrictEqual(failed, [])
716 })
717
718 it.effect("an interrupted take never swallows the next message", () =>
719 interruptAtEveryYield((budget) =>
720 Effect.gen(function*() {
721 const pubsub = yield* PubSub.bounded<number>(4)
722 const subscription = yield* PubSub.subscribe(pubsub)
723 const fiber = yield* Effect.forkChild(budget(PubSub.take(subscription)), { startImmediately: true })
724 yield* Fiber.interrupt(fiber)
725 yield* PubSub.publish(pubsub, 1)
726 return (yield* PubSub.takeUpTo(subscription, 1)).length === 1
727 })
728 ))
729
730 it.effect("a take that starts waiting after unsubscribe is interrupted", () =>
731 interruptAtEveryYield((budget) =>
732 Effect.gen(function*() {
733 const pubsub = yield* PubSub.unbounded<number>()
734 const scope = yield* Scope.make()
735 const subscription = yield* PubSub.subscribe(pubsub).pipe(Scope.provide(scope))
736 const fiber = yield* Effect.forkChild(budget(PubSub.take(subscription)), { startImmediately: true })
737 yield* Scope.close(scope, Exit.void)
738 for (let i = 0; i < 100 && fiber.pollUnsafe() === undefined; i++) yield* Effect.yieldNow
739 const exit = fiber.pollUnsafe()
740 return exit !== undefined && Exit.hasInterrupts(exit)
741 })
742 ))
743
744 it.effect("an interrupted backpressured publisher never publishes later", () =>
745 interruptAtEveryYield((budget) =>
746 Effect.gen(function*() {
747 const pubsub = yield* PubSub.bounded<number>(1)
748 const subscription = yield* PubSub.subscribe(pubsub)
749 yield* PubSub.publish(pubsub, 1)
750 const fiber = yield* Effect.forkChild(budget(PubSub.publish(pubsub, 2)), { startImmediately: true })
751 yield* Fiber.interrupt(fiber)
752 yield* PubSub.takeUpTo(subscription, 1)
753 return (yield* PubSub.takeUpTo(subscription, 1)).length === 0
754 })
755 ))
756
757 it.effect("shutdown interrupts suspended takeAll subscribers", () =>
758 Effect.gen(function*() {
759 const pubsub = yield* PubSub.unbounded<number>()
760 const subscription = yield* PubSub.subscribe(pubsub)
761 const fiber = yield* Effect.forkChild(PubSub.takeAll(subscription), { startImmediately: true })
762 yield* PubSub.shutdown(pubsub)
763 const exit = yield* Fiber.await(fiber)
764 assert.isTrue(Exit.hasInterrupts(exit))
765 }))
766
767 it.effect("Stream.fromPubSub completes after shutdown", () =>
768 Effect.gen(function*() {
769 const pubsub = yield* PubSub.unbounded<number>()
770 const fiber = yield* Effect.forkChild(Stream.runCollect(Stream.fromPubSub(pubsub)))
771
772 yield* Effect.yieldNow
773 assert.isUndefined(fiber.pollUnsafe())
774
775 yield* PubSub.shutdown(pubsub)
776
777 const result = yield* Fiber.join(fiber)
778 assert.deepStrictEqual(result, [])
779 }))
780
781 it.effect("publish returns false after shutdown", () =>
782 Effect.gen(function*() {
783 const pubsub = yield* PubSub.unbounded<number>()
784 yield* PubSub.shutdown(pubsub)
785
786 assert.strictEqual(yield* PubSub.publish(pubsub, 1), false)
787 }))
788
789 it.effect("publishAll returns false after shutdown", () =>
790 Effect.gen(function*() {
791 const pubsub = yield* PubSub.unbounded<number>()
792 yield* PubSub.shutdown(pubsub)
793
794 assert.strictEqual(yield* PubSub.publishAll(pubsub, [1, 2, 3]), false)
795 }))
796
797 describe("end", () => {
798 it.effect("delivers buffered messages before the final message and rejects later publishes", () =>
799 Effect.gen(function*() {
800 const pubsub = yield* PubSub.bounded<number>(4)
801 const subscription = yield* PubSub.subscribe(pubsub)
802 yield* PubSub.publishAll(pubsub, [1, 2])
803
804 assert.isTrue(yield* PubSub.end(pubsub, 0))
805 assert.isFalse(yield* PubSub.end(pubsub, -1))
806 assert.isFalse(yield* PubSub.publish(pubsub, 3))
807 assert.isFalse(yield* PubSub.publishAll(pubsub, [3]))
808 assert.isFalse(PubSub.publishUnsafe(pubsub, 3))
809
810 assert.deepStrictEqual(yield* PubSub.takeAll(subscription), [1, 2])
811 assert.strictEqual(yield* PubSub.take(subscription), 0)
812 assert.strictEqual(yield* PubSub.take(subscription), 0)
813 assert.deepStrictEqual(yield* PubSub.takeAll(subscription), [0])
814 }))
815
816 it.effect("wakes a suspended take", () =>
817 Effect.gen(function*() {
818 const pubsub = yield* PubSub.unbounded<number>()
819 const subscription = yield* PubSub.subscribe(pubsub)
820 const fiber = yield* Effect.forkChild(PubSub.take(subscription), { startImmediately: true })
821
822 yield* PubSub.end(pubsub, 0)
823
824 assert.strictEqual(yield* Fiber.join(fiber), 0)
825 }))
826
827 it.effect("is not dropped by a full dropping PubSub", () =>
828 Effect.gen(function*() {
829 const pubsub = yield* PubSub.dropping<number>(1)
830 const subscription = yield* PubSub.subscribe(pubsub)
831 yield* PubSub.publish(pubsub, 1)
832
833 yield* PubSub.end(pubsub, 0)
834
835 assert.strictEqual(yield* PubSub.take(subscription), 1)
836 assert.strictEqual(yield* PubSub.take(subscription), 0)
837 }))
838
839 it.effect("late subscribers receive the replayed messages and then the final message", () =>
840 Effect.gen(function*() {
841 const pubsub = yield* PubSub.unbounded<number>({ replay: 2 })
842 yield* PubSub.publishAll(pubsub, [1, 2, 3])
843 yield* PubSub.end(pubsub, 0)
844
845 const subscription = yield* PubSub.subscribe(pubsub)
846 assert.deepStrictEqual(yield* PubSub.takeBetween(subscription, 3, 3), [2, 3, 0])
847 }))
848
849 it.effect("rejects a suspended backpressured publish instead of delivering it after the final message", () =>
850 Effect.gen(function*() {
851 const pubsub = yield* PubSub.bounded<number>(1)
852 const fast = yield* PubSub.subscribe(pubsub)
853 const slow = yield* PubSub.subscribe(pubsub)
854 yield* PubSub.publish(pubsub, 1)
855 // The buffer stays full until the slow subscriber takes 1, so this
856 // publish suspends.
857 const publisher = yield* Effect.forkChild(PubSub.publish(pubsub, 2), { startImmediately: true })
858 assert.strictEqual(yield* PubSub.take(fast), 1)
859
860 yield* PubSub.end(pubsub, 0)
861
862 assert.strictEqual(yield* PubSub.take(fast), 0)
863 assert.strictEqual(yield* PubSub.take(slow), 1)
864 assert.isFalse(yield* Fiber.join(publisher))
865 assert.strictEqual(yield* PubSub.take(fast), 0)
866 assert.strictEqual(yield* PubSub.take(slow), 0)
867 }))
868
869 it.effect("shutdown still interrupts subscribers", () =>
870 Effect.gen(function*() {
871 const pubsub = yield* PubSub.unbounded<number>()
872 const subscription = yield* PubSub.subscribe(pubsub)
873 yield* PubSub.end(pubsub, 0)
874 yield* PubSub.shutdown(pubsub)
875
876 assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(PubSub.take(subscription))))
877 }))
878 })
879})
880
881const retains = (root: object, target: object): boolean => {
882 const objects = [root]
883 const seen = new Set<object>()
884 while (objects.length > 0) {
885 const current = objects.pop()!
886 if (current === target) {
887 return true
888 }
889 if (seen.has(current)) {
890 continue
891 }
892 seen.add(current)
893 for (const key of Reflect.ownKeys(current)) {
894 const descriptor = Object.getOwnPropertyDescriptor(current, key)
895 const value = descriptor && "value" in descriptor ? descriptor.value : undefined
896 if (typeof value === "object" && value !== null) {
897 objects.push(value)
898 }
899 }
900 }
901 return false
902}