1/**
2 * Manages fibers by key inside a scope.
3 *
4 * A `FiberMap<K, A, E>` owns a map of running fibers, interrupts them when its
5 * scope closes, and automatically removes each entry when the corresponding
6 * fiber completes. Use it when a program needs to start, replace, join, or
7 * interrupt background work by a stable key while keeping all fibers tied to
8 * one scope.
9 *
10 * @since 2.0.0
11 */
12import * as Cause from "./Cause.ts"
13import * as Deferred from "./Deferred.ts"
14import * as Effect from "./Effect.ts"
15import * as Exit from "./Exit.ts"
16import * as Fiber from "./Fiber.ts"
17import * as Filter from "./Filter.ts"
18import { constVoid, dual } from "./Function.ts"
19import type * as Inspectable from "./Inspectable.ts"
20import { PipeInspectableProto } from "./internal/core.ts"
21import * as internalEffect from "./internal/effect.ts"
22import * as Iterable from "./Iterable.ts"
23import * as MutableHashMap from "./MutableHashMap.ts"
24import * as Option from "./Option.ts"
25import type { Pipeable } from "./Pipeable.ts"
26import * as Predicate from "./Predicate.ts"
27import type * as Scope from "./Scope.ts"
28
29const TypeId = "~effect/FiberMap"
30
31/**
32 * A FiberMap is a collection of fibers, indexed by a key. When the associated
33 * Scope is closed, all fibers in the map will be interrupted. Fibers are
34 * automatically removed from the map when they complete.
35 *
36 * **Example** (Managing fibers in a map)
37 *
38 * ```ts import.meta.vitest
39 * import { Effect, FiberMap } from "effect"
40 *
41 * // Create a FiberMap with string keys
42 * const program = Effect.gen(function*() {
43 * const map = yield* FiberMap.make<string>()
44 *
45 * // Add some fibers to the map
46 * yield* FiberMap.run(map, "task1", Effect.never)
47 * yield* FiberMap.run(map, "task2", Effect.never)
48 *
49 * // Get the size of the map
50 * return yield* FiberMap.size(map)
51 * })
52 *
53 * const actual = await Effect.runPromise(Effect.scoped(program))
54 * actual // => 2
55 * ```
56 *
57 * @category models
58 * @since 2.0.0
59 */
60export interface FiberMap<in out K, out A = unknown, out E = unknown>
61 extends Pipeable, Inspectable.Inspectable, Iterable<[K, Fiber.Fiber<A, E>]>
62{
63 readonly [TypeId]: typeof TypeId
64 readonly deferred: Deferred.Deferred<void, unknown>
65 state: {
66 readonly _tag: "Open"
67 readonly backing: MutableHashMap.MutableHashMap<K, Fiber.Fiber<A, E>>
68 } | {
69 readonly _tag: "Closed"
70 }
71}
72
73/**
74 * Returns `true` if a value is a `FiberMap`.
75 *
76 * **Details**
77 *
78 * This is a type guard that checks for the `FiberMap` runtime marker.
79 *
80 * **Example** (Checking if a value is a FiberMap)
81 *
82 * ```ts import.meta.vitest
83 * import { Effect, FiberMap } from "effect"
84 *
85 * const program = Effect.gen(function*() {
86 * const map = yield* FiberMap.make<string>()
87 *
88 * return [FiberMap.isFiberMap(map), FiberMap.isFiberMap({}), FiberMap.isFiberMap(null)]
89 * })
90 *
91 * const actual = await Effect.runPromise(Effect.scoped(program))
92 * actual // => [true, false, false]
93 * ```
94 *
95 * @category guards
96 * @since 2.0.0
97 */
98export const isFiberMap = (u: unknown): u is FiberMap<unknown> => Predicate.hasProperty(u, TypeId)
99
100const Proto = {
101 [TypeId]: TypeId,
102 [Symbol.iterator](this: FiberMap<unknown>) {
103 if (this.state._tag === "Closed") {
104 return Iterable.empty()
105 }
106 return this.state.backing[Symbol.iterator]()
107 },
108 ...PipeInspectableProto,
109 toJSON(this: FiberMap<unknown>) {
110 return {
111 _id: "FiberMap",
112 state: this.state
113 }
114 }
115}
116
117const makeUnsafe = <K, A = unknown, E = unknown>(
118 backing: MutableHashMap.MutableHashMap<K, Fiber.Fiber<A, E>>,
119 deferred: Deferred.Deferred<void, E>
120): FiberMap<K, A, E> => {
121 const self = Object.create(Proto)
122 self.state = { _tag: "Open", backing }
123 self.deferred = deferred
124 return self
125}
126
127/**
128 * Creates a scoped `FiberMap` for storing fibers by key.
129 *
130 * **Details**
131 *
132 * When the associated Scope is closed, all fibers in the map will be
133 * interrupted. You can add fibers to the map using `FiberMap.set` or
134 * `FiberMap.run`, and the fibers will be automatically removed from the
135 * `FiberMap` when they complete.
136 *
137 * **Example** (Creating a scoped FiberMap)
138 *
139 * ```ts import.meta.vitest
140 * import { Effect, FiberMap } from "effect"
141 *
142 * const program = Effect.gen(function*() {
143 * const map = yield* FiberMap.make<string>()
144 *
145 * // run some effects and add the fibers to the map
146 * yield* FiberMap.run(map, "fiber a", Effect.never)
147 * yield* FiberMap.run(map, "fiber b", Effect.never)
148 *
149 * yield* Effect.yieldNow
150 * return yield* FiberMap.size(map)
151 * }).pipe(
152 * Effect.scoped // The fibers will be interrupted when the scope is closed
153 * )
154 *
155 * const actual = await Effect.runPromise(program)
156 * actual // => 2
157 * ```
158 *
159 * @category constructors
160 * @since 2.0.0
161 */
162export const make = <K, A = unknown, E = unknown>(): Effect.Effect<FiberMap<K, A, E>, never, Scope.Scope> =>
163 Effect.acquireRelease(
164 Effect.sync(() =>
165 makeUnsafe<K, A, E>(
166 MutableHashMap.empty(),
167 Deferred.makeUnsafe()
168 )
169 ),
170 (map) =>
171 Effect.suspend(() => {
172 const state = map.state
173 if (state._tag === "Closed") return Effect.void
174 map.state = { _tag: "Closed" }
175 return Fiber.interruptAll(MutableHashMap.values(state.backing)).pipe(
176 Deferred.into(map.deferred)
177 )
178 })
179 )
180
181/**
182 * Creates a scoped run function that forks effects into a new `FiberMap`.
183 *
184 * **Details**
185 *
186 * Each call stores the forked fiber under the supplied key and returns that
187 * fiber. If the key already has a fiber, the previous fiber is interrupted
188 * unless `onlyIfMissing` is set. All managed fibers are interrupted when the
189 * map's scope closes.
190 *
191 * **Example** (Creating a scoped runtime)
192 *
193 * ```ts import.meta.vitest
194 * import { Effect, Fiber, FiberMap } from "effect"
195 *
196 * const program = Effect.gen(function*() {
197 * const run = yield* FiberMap.makeRuntime<never, string>()
198 *
199 * // Run effects and get back fibers
200 * const fiber1 = run("task1", Effect.succeed("Hello"))
201 * const fiber2 = run("task2", Effect.succeed("World"))
202 *
203 * // Join the fibers to get their successful values
204 * return [yield* Fiber.join(fiber1), yield* Fiber.join(fiber2)]
205 * })
206 *
207 * const actual = await Effect.runPromise(Effect.scoped(program))
208 * actual // => ["Hello", "World"]
209 * ```
210 *
211 * @category constructors
212 * @since 2.0.0
213 */
214export const makeRuntime = <R, K, E = unknown, A = unknown>(): Effect.Effect<
215 <XE extends E, XA extends A>(
216 key: K,
217 effect: Effect.Effect<XA, XE, R>,
218 options?:
219 | Effect.RunOptions & {
220 readonly onlyIfMissing?: boolean | undefined
221 }
222 | undefined
223 ) => Fiber.Fiber<XA, XE>,
224 never,
225 Scope.Scope | R
226> =>
227 Effect.flatMap(
228 make<K, A, E>(),
229 (self) => runtime(self)<R>()
230 )
231
232/**
233 * Creates a scoped run function that forks effects into a new `FiberMap` and
234 * returns a `Promise` for each effect result.
235 *
236 * **When to use**
237 *
238 * Use when keyed fibers must be managed in a scoped map while exposing their
239 * results through Promise-based APIs.
240 *
241 * **Details**
242 *
243 * Each call stores the fiber under the supplied key, interrupting any previous
244 * fiber for that key unless `onlyIfMissing` is set. The returned Promise
245 * resolves with the effect's success value or rejects with the squashed failure
246 * cause.
247 *
248 * **Example** (Creating a promise runtime)
249 *
250 * ```ts import.meta.vitest
251 * import { Effect, FiberMap } from "effect"
252 *
253 * const program = Effect.gen(function*() {
254 * const run = yield* FiberMap.makeRuntimePromise<never, string>()
255 *
256 * // Run effects and get back promises
257 * const promise1 = run("task1", Effect.succeed("Hello"))
258 * const promise2 = run("task2", Effect.succeed("World"))
259 *
260 * // Convert to Effect and await
261 * return [yield* Effect.promise(() => promise1), yield* Effect.promise(() => promise2)]
262 * })
263 *
264 * const actual = await Effect.runPromise(Effect.scoped(program))
265 * actual // => ["Hello", "World"]
266 * ```
267 *
268 * @category constructors
269 * @since 3.13.0
270 */
271export const makeRuntimePromise = <R, K, A = unknown, E = unknown>(): Effect.Effect<
272 <XE extends E, XA extends A>(
273 key: K,
274 effect: Effect.Effect<XA, XE, R>,
275 options?:
276 | Effect.RunOptions & {
277 readonly onlyIfMissing?: boolean | undefined
278 }
279 | undefined
280 ) => Promise<XA>,
281 never,
282 Scope.Scope | R
283> =>
284 Effect.flatMap(
285 make<K, A, E>(),
286 (self) => runtimePromise(self)<R>()
287 )
288
289const internalFiberId = -1
290const isInternalInterruption = Filter.toPredicate(Filter.compose(
291 Cause.filterInterruptors,
292 Filter.has(internalFiberId)
293))
294
295/**
296 * Adds a fiber to the `FiberMap` under a key using a synchronous, unsafe
297 * mutation.
298 *
299 * **When to use**
300 *
301 * Use when an existing forked fiber must be installed under a key immediately
302 * and synchronous interruption of the replaced fiber is acceptable.
303 *
304 * **Details**
305 *
306 * When the fiber completes, it is removed from the map. If the key already has
307 * a fiber, that previous fiber is interrupted unless `onlyIfMissing` is set;
308 * in that case a different new fiber is interrupted and the existing entry is
309 * kept, while re-registering the existing fiber is a no-op.
310 *
311 * **Example** (Adding a fiber unsafely)
312 *
313 * ```ts import.meta.vitest
314 * import { Deferred, Effect, Fiber, FiberMap } from "effect"
315 *
316 * const program = Effect.gen(function*() {
317 * const map = yield* FiberMap.make<string>()
318 * const deferred = yield* Deferred.make<string>()
319 *
320 * // Create a fiber and add it to the map
321 * const fiber = yield* Effect.forkChild(Deferred.await(deferred))
322 * FiberMap.setUnsafe(map, "greeting", fiber)
323 *
324 * yield* Deferred.succeed(deferred, "Hello")
325 *
326 * // Join the fiber to get its successful value
327 * return yield* Fiber.join(fiber)
328 * })
329 *
330 * const actual = await Effect.runPromise(Effect.scoped(program))
331 * actual // => "Hello"
332 * ```
333 *
334 * @category combinators
335 * @since 4.0.0
336 */
337export const setUnsafe: {
338 <K, A, E, XE extends E, XA extends A>(
339 key: K,
340 fiber: Fiber.Fiber<XA, XE>,
341 options?: {
342 readonly onlyIfMissing?: boolean | undefined
343 readonly propagateInterruption?: boolean | undefined
344 } | undefined
345 ): (self: FiberMap<K, A, E>) => void
346 <K, A, E, XE extends E, XA extends A>(
347 self: FiberMap<K, A, E>,
348 key: K,
349 fiber: Fiber.Fiber<XA, XE>,
350 options?: {
351 readonly onlyIfMissing?: boolean | undefined
352 readonly propagateInterruption?: boolean | undefined
353 } | undefined
354 ): void
355} = dual((args) => isFiberMap(args[0]), <K, A, E, XE extends E, XA extends A>(
356 self: FiberMap<K, A, E>,
357 key: K,
358 fiber: Fiber.Fiber<XA, XE>,
359 options?: {
360 readonly onlyIfMissing?: boolean | undefined
361 readonly propagateInterruption?: boolean | undefined
362 } | undefined
363): void => {
364 if (self.state._tag === "Closed") {
365 fiber.interruptUnsafe(internalFiberId)
366 return
367 }
368
369 const previous = MutableHashMap.get(self.state.backing, key)
370 if (previous._tag === "Some") {
371 if (previous.value === fiber) {
372 return
373 } else if (options?.onlyIfMissing === true) {
374 fiber.interruptUnsafe(internalFiberId)
375 return
376 }
377 }
378
379 // Install the replacement before interruption can re-enter the map through a finalizer.
380 MutableHashMap.set(self.state.backing, key, fiber)
381 if (previous._tag === "Some") {
382 previous.value.interruptUnsafe(internalFiberId)
383 }
384 fiber.addObserver((exit) => {
385 if (self.state._tag === "Closed") {
386 return
387 }
388 const current = MutableHashMap.get(self.state.backing, key)
389 if (Option.isSome(current) && fiber === current.value) {
390 MutableHashMap.remove(self.state.backing, key)
391 }
392 if (
393 Exit.isFailure(exit) &&
394 (
395 options?.propagateInterruption === true ?
396 !isInternalInterruption(exit.cause) :
397 !Cause.hasInterruptsOnly(exit.cause)
398 )
399 ) {
400 Deferred.doneUnsafe(self.deferred, exit as any)
401 }
402 })
403})
404
405/**
406 * Adds a fiber to the `FiberMap` under a key.
407 *
408 * **Details**
409 *
410 * When the fiber completes, it is removed from the map. If the key already has
411 * a fiber, that previous fiber is interrupted unless `onlyIfMissing` is set;
412 * in that case a different new fiber is interrupted and the existing entry is
413 * kept, while re-registering the existing fiber is a no-op.
414 *
415 * This is the Effect-wrapped version of `setUnsafe`.
416 *
417 * **Example** (Adding a fiber)
418 *
419 * ```ts import.meta.vitest
420 * import { Deferred, Effect, Fiber, FiberMap } from "effect"
421 *
422 * const program = Effect.gen(function*() {
423 * const map = yield* FiberMap.make<string>()
424 * const deferred = yield* Deferred.make<string>()
425 *
426 * // Create a fiber and add it to the map using Effect
427 * const fiber = yield* Effect.forkChild(Deferred.await(deferred))
428 * yield* FiberMap.set(map, "greeting", fiber)
429 *
430 * yield* Deferred.succeed(deferred, "Hello")
431 *
432 * // Join the fiber to get its successful value
433 * return yield* Fiber.join(fiber)
434 * })
435 *
436 * const actual = await Effect.runPromise(Effect.scoped(program))
437 * actual // => "Hello"
438 * ```
439 *
440 * @category combinators
441 * @since 2.0.0
442 */
443export const set: {
444 <K, A, E, XE extends E, XA extends A>(
445 key: K,
446 fiber: Fiber.Fiber<XA, XE>,
447 options?: {
448 readonly onlyIfMissing?: boolean | undefined
449 readonly propagateInterruption?: boolean | undefined
450 } | undefined
451 ): (self: FiberMap<K, A, E>) => Effect.Effect<void>
452 <K, A, E, XE extends E, XA extends A>(
453 self: FiberMap<K, A, E>,
454 key: K,
455 fiber: Fiber.Fiber<XA, XE>,
456 options?: {
457 readonly onlyIfMissing?: boolean | undefined
458 readonly propagateInterruption?: boolean | undefined
459 } | undefined
460 ): Effect.Effect<void>
461} = dual((args) => isFiberMap(args[0]), <K, A, E, XE extends E, XA extends A>(
462 self: FiberMap<K, A, E>,
463 key: K,
464 fiber: Fiber.Fiber<XA, XE>,
465 options?: {
466 readonly onlyIfMissing?: boolean | undefined
467 readonly propagateInterruption?: boolean | undefined
468 } | undefined
469): Effect.Effect<void> => Effect.sync(() => setUnsafe(self, key, fiber, options)))
470
471/**
472 * Retrieves a fiber from the FiberMap synchronously.
473 *
474 * **When to use**
475 *
476 * Use when synchronous keyed lookup of a fiber in a `FiberMap` is needed and an
477 * `Option` result is enough outside the Effect workflow.
478 *
479 * **Example** (Retrieving a fiber unsafely)
480 *
481 * ```ts import.meta.vitest
482 * import { Deferred, Effect, Fiber, FiberMap, Option } from "effect"
483 *
484 * const program = Effect.gen(function*() {
485 * const map = yield* FiberMap.make<string>()
486 * const deferred = yield* Deferred.make<string>()
487 *
488 * // Add a fiber to the map
489 * const fiber = yield* Effect.forkChild(Deferred.await(deferred))
490 * FiberMap.setUnsafe(map, "greeting", fiber)
491 *
492 * // Retrieve the fiber
493 * const retrieved = FiberMap.getUnsafe(map, "greeting")
494 * yield* Deferred.succeed(deferred, "Hello")
495 * const result = yield* Fiber.join(fiber)
496 * return Option.map(retrieved, () => result)
497 * })
498 *
499 * const actual = await Effect.runPromise(Effect.scoped(program))
500 * actual // => Option.some("Hello")
501 * ```
502 *
503 * @category combinators
504 * @since 4.0.0
505 */
506export const getUnsafe: {
507 <K>(key: K): <A, E>(self: FiberMap<K, A, E>) => Option.Option<Fiber.Fiber<A, E>>
508 <K, A, E>(self: FiberMap<K, A, E>, key: K): Option.Option<Fiber.Fiber<A, E>>
509} = dual(
510 2,
511 <K, A, E>(self: FiberMap<K, A, E>, key: K): Option.Option<Fiber.Fiber<A, E>> => {
512 return self.state._tag === "Closed" ? Option.none() : MutableHashMap.get(self.state.backing, key)
513 }
514)
515
516/**
517 * Retrieves a fiber from the FiberMap effectfully.
518 *
519 * **Details**
520 *
521 * Returns an `Option` wrapped in `Effect`.
522 *
523 * **Example** (Retrieving a fiber)
524 *
525 * ```ts import.meta.vitest
526 * import { Deferred, Effect, Fiber, FiberMap, Option } from "effect"
527 *
528 * const program = Effect.gen(function*() {
529 * const map = yield* FiberMap.make<string>()
530 * const deferred = yield* Deferred.make<string>()
531 *
532 * // Add a fiber to the map
533 * const fiber = yield* Effect.forkChild(Deferred.await(deferred))
534 * yield* FiberMap.set(map, "greeting", fiber)
535 *
536 * // Retrieve the fiber with error handling
537 * const retrieved = yield* FiberMap.get(map, "greeting")
538 * yield* Deferred.succeed(deferred, "Hello")
539 * const result = yield* Fiber.join(fiber)
540 * return Option.map(retrieved, () => result)
541 * })
542 *
543 * const actual = await Effect.runPromise(Effect.scoped(program))
544 * actual // => Option.some("Hello")
545 * ```
546 *
547 * @category combinators
548 * @since 2.0.0
549 */
550export const get: {
551 <K>(key: K): <A, E>(self: FiberMap<K, A, E>) => Effect.Effect<Option.Option<Fiber.Fiber<A, E>>>
552 <K, A, E>(self: FiberMap<K, A, E>, key: K): Effect.Effect<Option.Option<Fiber.Fiber<A, E>>>
553} = dual(
554 2,
555 <K, A, E>(self: FiberMap<K, A, E>, key: K): Effect.Effect<Option.Option<Fiber.Fiber<A, E>>> =>
556 Effect.suspend(() => Effect.succeed(getUnsafe(self, key)))
557)
558
559/**
560 * Checks whether a key exists in the FiberMap.
561 *
562 * **Example** (Checking if a key exists unsafely)
563 *
564 * ```ts import.meta.vitest
565 * import { Effect, FiberMap } from "effect"
566 *
567 * const program = Effect.gen(function*() {
568 * const map = yield* FiberMap.make<string>()
569 *
570 * // Add a fiber to the map
571 * yield* FiberMap.run(map, "task1", Effect.never)
572 *
573 * // Check if keys exist
574 * return [FiberMap.hasUnsafe(map, "task1"), FiberMap.hasUnsafe(map, "task2")]
575 * })
576 *
577 * const actual = await Effect.runPromise(Effect.scoped(program))
578 * actual // => [true, false]
579 * ```
580 *
581 * @category combinators
582 * @since 4.0.0
583 */
584export const hasUnsafe: {
585 <K>(key: K): <A, E>(self: FiberMap<K, A, E>) => boolean
586 <K, A, E>(self: FiberMap<K, A, E>, key: K): boolean
587} = dual(
588 2,
589 <K, A, E>(self: FiberMap<K, A, E>, key: K): boolean =>
590 self.state._tag === "Closed" ? false : MutableHashMap.has(self.state.backing, key)
591)
592
593/**
594 * Checks whether a key exists in the FiberMap.
595 * This is the Effect-wrapped version of `hasUnsafe`.
596 *
597 * **Example** (Checking if a key exists)
598 *
599 * ```ts import.meta.vitest
600 * import { Effect, FiberMap } from "effect"
601 *
602 * const program = Effect.gen(function*() {
603 * const map = yield* FiberMap.make<string>()
604 *
605 * // Add a fiber to the map
606 * yield* FiberMap.run(map, "task1", Effect.never)
607 *
608 * // Check if keys exist using Effect
609 * return [yield* FiberMap.has(map, "task1"), yield* FiberMap.has(map, "task2")]
610 * })
611 *
612 * const actual = await Effect.runPromise(Effect.scoped(program))
613 * actual // => [true, false]
614 * ```
615 *
616 * @category combinators
617 * @since 2.0.0
618 */
619export const has: {
620 <K>(key: K): <A, E>(self: FiberMap<K, A, E>) => Effect.Effect<boolean>
621 <K, A, E>(self: FiberMap<K, A, E>, key: K): Effect.Effect<boolean>
622} = dual(
623 2,
624 <K, A, E>(self: FiberMap<K, A, E>, key: K): Effect.Effect<boolean> => Effect.sync(() => hasUnsafe(self, key))
625)
626
627/**
628 * Removes a fiber from the FiberMap, interrupting it if it exists.
629 *
630 * **Example** (Removing a fiber)
631 *
632 * ```ts import.meta.vitest
633 * import { Effect, FiberMap } from "effect"
634 *
635 * const program = Effect.gen(function*() {
636 * const map = yield* FiberMap.make<string>()
637 *
638 * // Add some fibers to the map
639 * yield* FiberMap.run(map, "task1", Effect.never)
640 * yield* FiberMap.run(map, "task2", Effect.never)
641 *
642 * const sizeBefore = yield* FiberMap.size(map)
643 *
644 * // Remove a specific fiber (this will interrupt it)
645 * yield* FiberMap.remove(map, "task1")
646 *
647 * return [sizeBefore, yield* FiberMap.size(map)]
648 * })
649 *
650 * const actual = await Effect.runPromise(Effect.scoped(program))
651 * actual // => [2, 1]
652 * ```
653 *
654 * @category combinators
655 * @since 2.0.0
656 */
657export const remove: {
658 <K>(key: K): <A, E>(self: FiberMap<K, A, E>) => Effect.Effect<void>
659 <K, A, E>(self: FiberMap<K, A, E>, key: K): Effect.Effect<void>
660} = dual<
661 <K>(
662 key: K
663 ) => <A, E>(self: FiberMap<K, A, E>) => Effect.Effect<void>,
664 <K, A, E>(
665 self: FiberMap<K, A, E>,
666 key: K
667 ) => Effect.Effect<void>
668>(2, (self, key) =>
669 Effect.suspend(() => {
670 if (self.state._tag === "Closed") {
671 return Effect.void
672 }
673 const fiber = MutableHashMap.get(self.state.backing, key)
674 if (fiber._tag === "None") {
675 return Effect.void
676 }
677 return Fiber.interruptAs(fiber.value, internalFiberId)
678 }))
679
680/**
681 * Removes all fibers from the FiberMap, interrupting them.
682 *
683 * **Example** (Clearing all fibers)
684 *
685 * ```ts import.meta.vitest
686 * import { Effect, FiberMap } from "effect"
687 *
688 * const program = Effect.gen(function*() {
689 * const map = yield* FiberMap.make<string>()
690 *
691 * // Add some fibers to the map
692 * yield* FiberMap.run(map, "task1", Effect.never)
693 * yield* FiberMap.run(map, "task2", Effect.never)
694 * yield* FiberMap.run(map, "task3", Effect.never)
695 *
696 * const sizeBefore = yield* FiberMap.size(map)
697 *
698 * // Clear all fibers (this will interrupt all of them)
699 * yield* FiberMap.clear(map)
700 *
701 * return [sizeBefore, yield* FiberMap.size(map)]
702 * })
703 *
704 * const actual = await Effect.runPromise(Effect.scoped(program))
705 * actual // => [3, 0]
706 * ```
707 *
708 * @category combinators
709 * @since 2.0.0
710 */
711export const clear = <K, A, E>(self: FiberMap<K, A, E>): Effect.Effect<void> =>
712 Effect.suspend(() => {
713 if (self.state._tag === "Closed") {
714 return Effect.void
715 }
716 return Fiber.interruptAllAs(MutableHashMap.values(self.state.backing), internalFiberId)
717 })
718
719const constInterruptedFiber = (function() {
720 let fiber: Fiber.Fiber<never, never> | undefined = undefined
721 return () => {
722 if (fiber === undefined) {
723 fiber = Effect.runFork(Effect.interrupt)
724 }
725 return fiber
726 }
727})()
728
729/**
730 * Forks an Effect and stores the resulting fiber in the `FiberMap` under a key.
731 *
732 * **Details**
733 *
734 * When the fiber completes, it is removed from the map. If the key already has
735 * a fiber, the previous fiber is interrupted unless `onlyIfMissing` is set.
736 * Set `startImmediately: false` to defer startup. By default, the effect starts
737 * immediately.
738 *
739 * **Example** (Forking effects into a map)
740 *
741 * ```ts import.meta.vitest
742 * import { Effect, Fiber, FiberMap } from "effect"
743 *
744 * const program = Effect.gen(function*() {
745 * const map = yield* FiberMap.make<string>()
746 *
747 * // Run effects and add the fibers to the map
748 * const fiber1 = yield* FiberMap.run(map, "task1", Effect.succeed("Hello"))
749 * const fiber2 = yield* FiberMap.run(map, "task2", Effect.succeed("World"))
750 *
751 * // Join the fibers to get their successful values
752 * const result1 = yield* Fiber.join(fiber1)
753 * const result2 = yield* Fiber.join(fiber2)
754 * return [result1, result2, yield* FiberMap.size(map)]
755 * })
756 *
757 * const actual = await Effect.runPromise(Effect.scoped(program))
758 * actual // => ["Hello", "World", 0]
759 * ```
760 *
761 * @category combinators
762 * @since 2.0.0
763 */
764export const run: {
765 <K, A, E>(
766 self: FiberMap<K, A, E>,
767 key: K,
768 options?: {
769 readonly onlyIfMissing?: boolean | undefined
770 readonly propagateInterruption?: boolean | undefined
771 readonly startImmediately?: boolean | undefined
772 } | undefined
773 ): <R, XE extends E, XA extends A>(
774 effect: Effect.Effect<XA, XE, R>
775 ) => Effect.Effect<Fiber.Fiber<XA, XE>, never, R>
776 <K, A, E, R, XE extends E, XA extends A>(
777 self: FiberMap<K, A, E>,
778 key: K,
779 effect: Effect.Effect<XA, XE, R>,
780 options?: {
781 readonly onlyIfMissing?: boolean | undefined
782 readonly propagateInterruption?: boolean | undefined
783 readonly startImmediately?: boolean | undefined
784 } | undefined
785 ): Effect.Effect<Fiber.Fiber<XA, XE>, never, R>
786} = function() {
787 const self = arguments[0]
788 if (Effect.isEffect(arguments[2])) {
789 return runImpl(self, arguments[1], arguments[2], arguments[3]) as any
790 }
791 const key = arguments[1]
792 const options = arguments[2]
793 return (effect: Effect.Effect<any, any, any>) => runImpl(self, key, effect, options)
794}
795
796const runImpl = <K, A, E, R, XE extends E, XA extends A>(
797 self: FiberMap<K, A, E>,
798 key: K,
799 effect: Effect.Effect<XA, XE, R>,
800 options?: {
801 readonly startImmediately?: boolean | undefined
802 readonly onlyIfMissing?: boolean
803 readonly propagateInterruption?: boolean | undefined
804 }
805): Effect.Effect<Fiber.Fiber<XA, XE>, never, R> =>
806 Effect.withFiber((parent) => {
807 if (self.state._tag === "Closed") {
808 return Effect.interrupt
809 } else if (options?.onlyIfMissing === true && hasUnsafe(self, key)) {
810 return Effect.sync(constInterruptedFiber)
811 }
812 const fiber: Fiber.Fiber<XA, XE> = internalEffect.forkUnsafe(
813 parent,
814 effect,
815 options?.startImmediately ?? true,
816 true
817 )
818 setUnsafe(self, key, fiber, options)
819 return Effect.succeed(fiber)
820 })
821
822/**
823 * Captures the current runtime and returns a function for forking effects into
824 * an existing `FiberMap`.
825 *
826 * **Details**
827 *
828 * Each call stores the forked fiber under the supplied key. If that key already
829 * has a fiber, the previous fiber is interrupted unless `onlyIfMissing` is set.
830 *
831 * **Example** (Capturing a runtime)
832 *
833 * ```ts import.meta.vitest
834 * import { Context, Effect, Fiber, FiberMap } from "effect"
835 *
836 * class Users extends Context.Service<Users, {
837 * readonly getAll: Effect.Effect<Array<unknown>>
838 * }>()("Users") {}
839 *
840 * const program = Effect.gen(function*() {
841 * const map = yield* FiberMap.make<string>()
842 * const run = yield* FiberMap.runtime(map)<Users>()
843 *
844 * // run some effects and add the fibers to the map
845 * const fiberA = run("effect-a", Effect.andThen(Users, (_) => _.getAll))
846 * const fiberB = run("effect-b", Effect.andThen(Users, (_) => _.getAll))
847 * return [(yield* Fiber.join(fiberA)).length, (yield* Fiber.join(fiberB)).length]
848 * }).pipe(
849 * Effect.scoped // The fibers will be interrupted when the scope is closed
850 * )
851 *
852 * const actual = await Effect.runPromise(Effect.provideService(program, Users, {
853 * getAll: Effect.succeed([])
854 * }))
855 * actual // => [0, 0]
856 * ```
857 *
858 * @category combinators
859 * @since 2.0.0
860 */
861export const runtime: <K, A, E>(
862 self: FiberMap<K, A, E>
863) => <R = never>() => Effect.Effect<
864 <XE extends E, XA extends A>(
865 key: K,
866 effect: Effect.Effect<XA, XE, R>,
867 options?:
868 | Effect.RunOptions & {
869 readonly onlyIfMissing?: boolean | undefined
870 readonly propagateInterruption?: boolean | undefined
871 }
872 | undefined
873 ) => Fiber.Fiber<XA, XE>,
874 never,
875 R
876> = <K, A, E>(self: FiberMap<K, A, E>) => <R>() =>
877 Effect.map(
878 Effect.context<R>(),
879 (services) => {
880 const runFork = Effect.runForkWith(services)
881 return <XE extends E, XA extends A>(
882 key: K,
883 effect: Effect.Effect<XA, XE, R>,
884 options?:
885 | Effect.RunOptions & {
886 readonly onlyIfMissing?: boolean | undefined
887 readonly propagateInterruption?: boolean | undefined
888 }
889 | undefined
890 ) => {
891 if (self.state._tag === "Closed") {
892 return constInterruptedFiber()
893 } else if (options?.onlyIfMissing === true && hasUnsafe(self, key)) {
894 return constInterruptedFiber()
895 }
896 const fiber = runFork(effect, options)
897 setUnsafe(self, key, fiber, options)
898 return fiber
899 }
900 }
901 )
902
903/**
904 * Captures the current runtime and returns a function for running effects in
905 * an existing `FiberMap` as Promises.
906 *
907 * **Details**
908 *
909 * Each call stores the forked fiber under the supplied key, interrupting any
910 * previous fiber for that key unless `onlyIfMissing` is set. The Promise
911 * resolves with the effect's success value or rejects with the squashed failure
912 * cause.
913 *
914 * **Example** (Running effects as promises)
915 *
916 * ```ts import.meta.vitest
917 * import { Effect, FiberMap } from "effect"
918 *
919 * const program = Effect.gen(function*() {
920 * const map = yield* FiberMap.make<string>()
921 * const runPromise = yield* FiberMap.runtimePromise(map)<never>()
922 *
923 * // Create promises that will be backed by fibers in the map
924 * const promise1 = runPromise("task1", Effect.succeed("Hello"))
925 * const promise2 = runPromise("task2", Effect.succeed("World"))
926 *
927 * // Convert promises back to Effects and await
928 * return [yield* Effect.promise(() => promise1), yield* Effect.promise(() => promise2)]
929 * })
930 *
931 * const actual = await Effect.runPromise(Effect.scoped(program))
932 * actual // => ["Hello", "World"]
933 * ```
934 *
935 * @category combinators
936 * @since 3.13.0
937 */
938export const runtimePromise = <K, A, E>(self: FiberMap<K, A, E>): <R = never>() => Effect.Effect<
939 <XE extends E, XA extends A>(
940 key: K,
941 effect: Effect.Effect<XA, XE, R>,
942 options?:
943 | Effect.RunOptions & {
944 readonly onlyIfMissing?: boolean | undefined
945 readonly propagateInterruption?: boolean | undefined
946 }
947 | undefined
948 ) => Promise<XA>,
949 never,
950 R
951> =>
952<R>() =>
953 Effect.map(
954 runtime(self)<R>(),
955 (runFork) =>
956 <XE extends E, XA extends A>(
957 key: K,
958 effect: Effect.Effect<XA, XE, R>,
959 options?:
960 | Effect.RunOptions & { readonly propagateInterruption?: boolean | undefined }
961 | undefined
962 ): Promise<XA> =>
963 new Promise((resolve, reject) =>
964 runFork(key, effect, options).addObserver((exit) => {
965 if (Exit.isSuccess(exit)) {
966 resolve(exit.value)
967 } else {
968 reject(Cause.squash(exit.cause))
969 }
970 })
971 )
972 )
973
974/**
975 * Gets the number of fibers currently in the FiberMap.
976 *
977 * **Example** (Checking the map size)
978 *
979 * ```ts import.meta.vitest
980 * import { Effect, FiberMap } from "effect"
981 *
982 * const program = Effect.gen(function*() {
983 * const map = yield* FiberMap.make<string>()
984 *
985 * const sizeBefore = yield* FiberMap.size(map)
986 *
987 * // Add some fibers
988 * yield* FiberMap.run(map, "task1", Effect.never)
989 * yield* FiberMap.run(map, "task2", Effect.never)
990 *
991 * return [sizeBefore, yield* FiberMap.size(map)]
992 * })
993 *
994 * const actual = await Effect.runPromise(Effect.scoped(program))
995 * actual // => [0, 2]
996 * ```
997 *
998 * @category combinators
999 * @since 2.0.0
1000 */
1001export const size = <K, A, E>(self: FiberMap<K, A, E>): Effect.Effect<number> =>
1002 Effect.sync(() => self.state._tag === "Closed" ? 0 : MutableHashMap.size(self.state.backing))
1003
1004/**
1005 * Waits for the `FiberMap` to fail or close.
1006 *
1007 * **Details**
1008 *
1009 * The returned Effect fails with the first managed fiber failure that is not
1010 * ignored by the map's interruption rules. Normal successful completion
1011 * removes fibers from the map; use `awaitEmpty` to wait until the map has no
1012 * fibers.
1013 *
1014 * **Example** (Joining failing fibers)
1015 *
1016 * ```ts import.meta.vitest
1017 * import { Effect, Exit, FiberMap } from "effect"
1018 *
1019 * const program = Effect.gen(function*() {
1020 * const map = yield* FiberMap.make()
1021 * yield* FiberMap.set(map, "a", Effect.runFork(Effect.fail("error")))
1022 *
1023 * // parent fiber will fail with "error"
1024 * yield* FiberMap.join(map)
1025 * })
1026 *
1027 * const actual = await Effect.runPromise(Effect.exit(Effect.scoped(program)))
1028 * actual // => Exit.fail("error")
1029 * ```
1030 *
1031 * @category combinators
1032 * @since 2.0.0
1033 */
1034export const join = <K, A, E>(self: FiberMap<K, A, E>): Effect.Effect<void, E> =>
1035 Deferred.await(self.deferred as Deferred.Deferred<void, E>)
1036
1037/**
1038 * Waits for the FiberMap to be empty.
1039 * This will wait for all currently running fibers to complete.
1040 *
1041 * **Example** (Waiting for an empty map)
1042 *
1043 * ```ts import.meta.vitest
1044 * import { Effect, FiberMap } from "effect"
1045 *
1046 * const program = Effect.gen(function*() {
1047 * const map = yield* FiberMap.make<string>()
1048 *
1049 * yield* FiberMap.run(map, "task1", Effect.yieldNow)
1050 * yield* FiberMap.run(map, "task2", Effect.yieldNow)
1051 *
1052 * // Wait for the map to be empty
1053 * yield* FiberMap.awaitEmpty(map)
1054 *
1055 * return yield* FiberMap.size(map)
1056 * })
1057 *
1058 * const actual = await Effect.runPromise(Effect.scoped(program))
1059 * actual // => 0
1060 * ```
1061 *
1062 * @category combinators
1063 * @since 3.13.0
1064 */
1065export const awaitEmpty = <K, A, E>(self: FiberMap<K, A, E>): Effect.Effect<void, E> =>
1066 Effect.whileLoop({
1067 while: () => self.state._tag === "Open" && MutableHashMap.size(self.state.backing) > 0,
1068 body: () => Fiber.await(Iterable.headUnsafe(self)[1]),
1069 step: constVoid
1070 })