1/**
2 * Consumes values from a `Stream` and produces one final result.
3 *
4 * A `Sink` may read no input, a fixed amount of input, or keep reading until a
5 * condition is met. If it reads more than it needs, it can return leftovers so
6 * the stream can continue from those values. Sinks are used to collect, fold,
7 * search, count, or otherwise reduce streamed input, and they can be composed
8 * when a stream needs more than one consuming step.
9 *
10 * @since 2.0.0
11 */
12import type { NonEmptyReadonlyArray } from "./Array.ts"
13import * as Arr from "./Array.ts"
14import * as Cause from "./Cause.ts"
15import * as Channel from "./Channel.ts"
16import * as Clock from "./Clock.ts"
17import type * as Context from "./Context.ts"
18import * as Duration from "./Duration.ts"
19import * as Effect from "./Effect.ts"
20import * as Exit from "./Exit.ts"
21import type * as Filter from "./Filter.ts"
22import type { LazyArg } from "./Function.ts"
23import { constant, constFalse, constTrue, constVoid, dual, identity, pipe } from "./Function.ts"
24import * as Count from "./internal/count.ts"
25import * as internalStream from "./internal/stream.ts"
26import * as Option from "./Option.ts"
27import { type Pipeable, pipeArguments } from "./Pipeable.ts"
28import type { Predicate, Refinement } from "./Predicate.ts"
29import { hasProperty } from "./Predicate.ts"
30import * as PubSub from "./PubSub.ts"
31import * as Pull from "./Pull.ts"
32import * as Queue from "./Queue.ts"
33import * as Result from "./Result.ts"
34import * as Scope from "./Scope.ts"
35import type { Stream } from "./Stream.ts"
36import type * as Types from "./Types.ts"
37import type * as Unify from "./Unify.ts"
38
39const TypeId = "~effect/Sink"
40
41/**
42 * A `Sink<A, In, L, E, R>` is used to consume elements produced by a `Stream`.
43 * You can think of a sink as a function that will consume a variable amount of
44 * `In` elements (could be 0, 1, or many), might fail with an error of type `E`,
45 * and will eventually yield a value of type `A` together with a remainder of
46 * type `L` (i.e. any leftovers).
47 *
48 * **Example** (Running a sink with a stream)
49 *
50 * ```ts import.meta.vitest
51 * import { Effect, Sink, Stream } from "effect"
52 *
53 * // Create a simple sink that always succeeds with a value
54 * const sink: Sink.Sink<number> = Sink.succeed(42)
55 *
56 * // Use the sink to consume a stream
57 * const stream = Stream.make(1, 2, 3)
58 * await Effect.runPromise(Stream.run(stream, sink)) // => 42
59 * ```
60 *
61 * @category models
62 * @since 2.0.0
63 */
64export interface Sink<out A, in In = unknown, out L = never, out E = never, out R = never>
65 extends Sink.Variance<A, In, L, E, R>, Pipeable
66{
67 readonly transform: (
68 upstream: Pull.Pull<NonEmptyReadonlyArray<In>, never, void>,
69 scope: Scope.Scope
70 ) => Effect.Effect<End<A, L>, E, R>
71 [Unify.typeSymbol]?: unknown
72 [Unify.unifySymbol]?: SinkUnify<this>
73 [Unify.ignoreSymbol]?: SinkUnifyIgnore
74}
75
76/**
77 * Tuple returned when a `Sink` finishes.
78 *
79 * **Details**
80 *
81 * The first element is the sink result. The optional second element contains a
82 * non-empty array of leftover input that was pulled but not consumed.
83 *
84 * @category models
85 * @since 4.0.0
86 */
87export type End<A, L = never> = readonly [value: A, leftover?: NonEmptyReadonlyArray<L> | undefined]
88
89const endVoid = Effect.succeed([void 0] as End<void, never>)
90
91/**
92 * Type-level unification support for `Sink` values.
93 *
94 * **Details**
95 *
96 * This preserves the result, input, leftover, error, and service type
97 * parameters when Effect's `Unify` machinery normalizes generic values that
98 * include sinks. Users normally do not need to reference this interface
99 * directly.
100 *
101 * @category models
102 * @since 2.0.0
103 */
104export interface SinkUnify<A extends { [Unify.typeSymbol]?: any }> extends Effect.EffectUnify<A> {
105 Sink?: () => A[Unify.typeSymbol] extends
106 | Sink<
107 infer A,
108 infer In,
109 infer L,
110 infer E,
111 infer R
112 >
113 | infer _ ? Sink<A, In, L, E, R>
114 : never
115}
116
117/**
118 * Marker used by Effect's `Unify` machinery for `Sink` values.
119 *
120 * **Details**
121 *
122 * It prevents the inherited `Effect` unifier from being selected when
123 * sink-specific unification should preserve the `Sink` type parameters. Users
124 * normally do not need to reference this interface directly.
125 *
126 * @category models
127 * @since 2.0.0
128 */
129export interface SinkUnifyIgnore {
130 Effect?: true
131}
132
133/**
134 * Namespace containing types and interfaces for Sink variance and type relationships.
135 *
136 * @since 2.0.0
137 */
138export declare namespace Sink {
139 /**
140 * Type-level variance marker for `Sink`.
141 *
142 * **Details**
143 *
144 * The result `A`, leftovers `L`, errors `E`, and services `R` are
145 * covariant. The input type `In` is contravariant because values flow into
146 * the sink.
147 *
148 * @category models
149 * @since 2.0.0
150 */
151 export interface Variance<out A, in In, out L, out E, out R> {
152 readonly [TypeId]: VarianceStruct<A, In, L, E, R>
153 }
154 /**
155 * Structural encoding used by `Sink.Variance` to record each `Sink` type
156 * parameter's variance.
157 *
158 * **Details**
159 *
160 * `_A`, `_L`, `_E`, and `_R` are covariant markers. `_In` is a
161 * contravariant marker.
162 *
163 * @category models
164 * @since 2.0.0
165 */
166 export interface VarianceStruct<out A, in In, out L, out E, out R> {
167 _A: Types.Covariant<A>
168 _In: Types.Contravariant<In>
169 _L: Types.Covariant<L>
170 _E: Types.Covariant<E>
171 _R: Types.Covariant<R>
172 }
173}
174
175const sinkVariance = {
176 _A: identity,
177 _In: identity,
178 _L: identity,
179 _E: identity,
180 _R: identity
181}
182
183const SinkProto = {
184 [TypeId]: sinkVariance,
185 pipe() {
186 return pipeArguments(this, arguments)
187 }
188}
189
190/**
191 * Checks whether a value is a Sink.
192 *
193 * **Example** (Checking for a sink)
194 *
195 * ```ts import.meta.vitest
196 * import { Sink } from "effect"
197 *
198 * const sink = Sink.never
199 * const notStream = { data: [1, 2, 3] }
200 *
201 * Sink.isSink(sink) // => true
202 * Sink.isSink(notStream) // => false
203 * ```
204 *
205 * @category guards
206 * @since 4.0.0
207 */
208export const isSink = (u: unknown): u is Sink<unknown, never, unknown, unknown, unknown> => hasProperty(u, TypeId)
209
210/**
211 * Creates a sink from a `Channel`.
212 *
213 * **When to use**
214 *
215 * Use to create a `Sink` from a `Channel` that processes non-empty arrays of
216 * input values.
217 *
218 * **Example** (Using channel completion as the sink result)
219 *
220 * ```ts import.meta.vitest
221 * import { Channel, Effect, Sink, Stream } from "effect"
222 *
223 * const channel = Channel.identity<readonly [number, ...Array<number>], never, void>().pipe(
224 * Channel.drain,
225 * Channel.mapDone(() => ["consumed"] as const)
226 * )
227 * const sink = Sink.fromChannel(channel)
228 *
229 * await Effect.runPromise(Stream.run(Stream.make(1, 2, 3), sink)) // => "consumed"
230 * ```
231 *
232 * @see {@link toChannel} for converting a `Sink` back to a `Channel`
233 * @category constructors
234 * @since 2.0.0
235 */
236export const fromChannel = <L, In, E, A, R>(
237 channel: Channel.Channel<
238 never,
239 E,
240 End<A, L>,
241 NonEmptyReadonlyArray<In>,
242 never,
243 void,
244 R
245 >
246): Sink<A, In, L, E, R> =>
247 fromTransform((upstream, scope) =>
248 Channel.toTransform(channel)(upstream, scope).pipe(
249 Effect.flatMap(Effect.forever({ disableYield: true })),
250 Pull.catchDone(Effect.succeed)
251 ) as Effect.Effect<End<A, L>, E, R>
252 )
253
254/**
255 * Creates a sink that writes its input to a Web `WritableStream`.
256 *
257 * **Example** (Collecting values in a Web stream)
258 *
259 * ```ts import.meta.vitest
260 * import { Effect, Sink, Stream } from "effect"
261 *
262 * const written: Array<number> = []
263 * const sink = Sink.fromWritableStream({
264 * evaluate: () => new WritableStream<number>({
265 * write(value) {
266 * written.push(value)
267 * }
268 * }),
269 * onError: (cause) => new Error(String(cause))
270 * })
271 *
272 * await Effect.runPromise(Stream.run(Stream.make(1, 2, 3), sink))
273 * written // => [1, 2, 3]
274 * ```
275 *
276 * @category constructors
277 * @since 4.0.0
278 */
279export const fromWritableStream = <A, E>(options: {
280 readonly evaluate: LazyArg<WritableStream<A>>
281 readonly onError: (error: unknown) => E
282 readonly closeOnDone?: boolean | undefined
283}): Sink<void, A, never, E> =>
284 fromChannel(
285 Channel.mapDone(
286 Channel.fromWritableStream<never, E, A>(options),
287 (_) => [_]
288 )
289 )
290
291/**
292 * Creates a `Sink` from a low-level transform function.
293 *
294 * **Details**
295 *
296 * The transform receives the upstream pull of non-empty input arrays and the
297 * active scope, and returns an effect that completes with the sink's `End`
298 * value.
299 *
300 * @category constructors
301 * @since 4.0.0
302 */
303export const fromTransform = <In, A, E, R, L = never>(
304 transform: (
305 upstream: Pull.Pull<NonEmptyReadonlyArray<In>, never, void>,
306 scope: Scope.Scope
307 ) => Effect.Effect<End<A, L>, E, R>
308): Sink<A, In, L, E, R> => {
309 const self = Object.create(SinkProto)
310 self.transform = transform
311 return self
312}
313
314/**
315 * Creates a `Channel` from a Sink.
316 *
317 * **Example** (Running a sink as a channel)
318 *
319 * ```ts import.meta.vitest
320 * import { Channel, Effect, Sink, Stream } from "effect"
321 *
322 * const channel = Stream.toChannel(Stream.make(1, 2, 3)).pipe(
323 * Channel.pipeTo(Sink.toChannel(Sink.sum))
324 * )
325 *
326 * await Effect.runPromise(Channel.runDrain(channel)) // => [6]
327 * ```
328 *
329 * @category constructors
330 * @since 2.0.0
331 */
332export const toChannel = <A, In, L, E, R>(
333 self: Sink<A, In, L, E, R>
334): Channel.Channel<never, E, End<A, L>, NonEmptyReadonlyArray<In>, never, void, R> =>
335 Channel.fromTransform((upstream, scope) =>
336 Effect.succeed(Effect.flatMap(
337 self.transform(upstream, scope),
338 Cause.done
339 ))
340 )
341
342/**
343 * Creates a pipe-style constructor for sinks over input type `In`.
344 *
345 * **Details**
346 *
347 * The returned function exposes the sink input as a `Stream<In>`, applies the
348 * provided pipeline, and uses the final effect's success value as the sink
349 * result.
350 *
351 * @category constructors
352 * @since 4.0.0
353 */
354export const make = <In>(): make.Constructor<In> => (...fns: []) =>
355 fromTransform((upstream, scope) =>
356 pipe(
357 internalStream.fromChannel(Channel.fromPull(Effect.succeed(upstream))),
358 ...fns as any as [() => Effect.Effect<any>],
359 Effect.flatMap((a) => Cause.done<End<any>>([a])),
360 Scope.provide(scope)
361 )
362 )
363
364/**
365 * Companion namespace containing overload types for the pipe-style sink
366 * constructor returned by `Sink.make`.
367 *
368 * @since 4.0.0
369 */
370export declare namespace make {
371 /**
372 * Overloaded function type returned by `Sink.make`.
373 *
374 * **Details**
375 *
376 * The first pipeline function receives the sink input as a `Stream<In>`. The
377 * final pipeline step must return an `Effect`, whose success value becomes
378 * the sink result.
379 *
380 * @category models
381 * @since 4.0.0
382 */
383 export interface Constructor<In> {
384 <E, R, B = never>(ab: (_: Stream<In>) => Effect.Effect<B, E, R>): Sink<B, In, never, E, Exclude<R, Scope.Scope>>
385 <E, R, B = never, C = never>(
386 ab: (_: Stream<In>) => B,
387 bc: (_: B) => Effect.Effect<C, E, R>
388 ): Sink<C, In, never, E, Exclude<R, Scope.Scope>>
389 <E, R, B = never, C = never, D = never>(
390 ab: (_: Stream<In>) => B,
391 bc: (_: B) => C,
392 cd: (_: C) => Effect.Effect<D, E, R>
393 ): Sink<D, In, never, E, Exclude<R, Scope.Scope>>
394 <E, R, B = never, C = never, D = never, F = never>(
395 ab: (_: Stream<In>) => B,
396 bc: (_: B) => C,
397 cd: (_: C) => D,
398 df: (_: D) => Effect.Effect<F, E, R>
399 ): Sink<F, In, never, E, Exclude<R, Scope.Scope>>
400 <E, R, B = never, C = never, D = never, F = never, G = never>(
401 ab: (_: Stream<In>) => B,
402 bc: (_: B) => C,
403 cd: (_: C) => D,
404 df: (_: D) => F,
405 fg: (_: F) => Effect.Effect<G, E, R>
406 ): Sink<G, In, never, E, Exclude<R, Scope.Scope>>
407 <E, R, B = never, C = never, D = never, F = never, G = never, H = never>(
408 ab: (_: Stream<In>) => B,
409 bc: (_: B) => C,
410 cd: (_: C) => D,
411 df: (_: D) => F,
412 fg: (_: F) => G,
413 gh: (_: G) => Effect.Effect<H, E, R>
414 ): Sink<H, In, never, E, Exclude<R, Scope.Scope>>
415 <E, R, B = never, C = never, D = never, F = never, G = never, H = never, I = never>(
416 ab: (_: Stream<In>) => B,
417 bc: (_: B) => C,
418 cd: (_: C) => D,
419 df: (_: D) => F,
420 fg: (_: F) => G,
421 gh: (_: G) => H,
422 hi: (_: H) => Effect.Effect<I, E, R>
423 ): Sink<I, In, never, E, Exclude<R, Scope.Scope>>
424 <E, R, B = never, C = never, D = never, F = never, G = never, H = never, I = never, J = never>(
425 ab: (_: Stream<In>) => B,
426 bc: (_: B) => C,
427 cd: (_: C) => D,
428 df: (_: D) => F,
429 fg: (_: F) => G,
430 gh: (_: G) => H,
431 hi: (_: H) => I,
432 ij: (_: I) => Effect.Effect<J, E, R>
433 ): Sink<J, In, never, E, Exclude<R, Scope.Scope>>
434 <E, R, B = never, C = never, D = never, F = never, G = never, H = never, I = never, J = never, K = never>(
435 ab: (_: Stream<In>) => B,
436 bc: (_: B) => C,
437 cd: (_: C) => D,
438 df: (_: D) => F,
439 fg: (_: F) => G,
440 gh: (_: G) => H,
441 hi: (_: H) => I,
442 ij: (_: I) => J,
443 jk: (_: J) => Effect.Effect<K, E, R>
444 ): Sink<K, In, never, E, Exclude<R, Scope.Scope>>
445 <
446 E,
447 R,
448 B = never,
449 C = never,
450 D = never,
451 F = never,
452 G = never,
453 H = never,
454 I = never,
455 J = never,
456 K = never,
457 L = never
458 >(
459 ab: (_: Stream<In>) => B,
460 bc: (_: B) => C,
461 cd: (_: C) => D,
462 df: (_: D) => F,
463 fg: (_: F) => G,
464 gh: (_: G) => H,
465 hi: (_: H) => I,
466 ij: (_: I) => J,
467 jk: (_: J) => K,
468 kl: (_: K) => Effect.Effect<L, E, R>
469 ): Sink<L, In, never, E, Exclude<R, Scope.Scope>>
470 }
471}
472
473/**
474 * Creates a sink that ignores upstream input and completes from an effect that
475 * already returns an `End`.
476 *
477 * **When to use**
478 *
479 * Use when you need to create a sink from an effect that returns both the sink
480 * result value and optional leftovers.
481 *
482 * @category constructors
483 * @since 4.0.0
484 */
485export const fromEffectEnd = <A, E, R, L = never>(
486 effect: Effect.Effect<End<A, L>, E, R>
487): Sink<A, unknown, L, E, R> => fromTransform(() => effect)
488
489/**
490 * Creates a sink that ignores upstream input and completes with the success
491 * value of the provided effect.
492 *
493 * **Details**
494 *
495 * If the effect fails, the sink fails with the same error.
496 *
497 * @category constructors
498 * @since 2.0.0
499 */
500export const fromEffect = <A, E, R>(
501 effect: Effect.Effect<A, E, R>
502): Sink<A, unknown, never, E, R> => fromEffectEnd(Effect.map(effect, (a) => [a]))
503
504/**
505 * Creates a sink that offers every consumed input element to a queue.
506 *
507 * **Details**
508 *
509 * When the upstream stream ends, the sink ends the queue and completes with
510 * `void`.
511 *
512 * @category constructors
513 * @since 2.0.0
514 */
515export const fromQueue = <A>(
516 queue: Queue.Queue<A, Cause.Done>
517): Sink<void, A> =>
518 fromTransform((upstream) =>
519 upstream.pipe(
520 Effect.flatMap((arr) => Queue.offerAll(queue, arr)),
521 Effect.forever({ disableYield: true }),
522 Pull.catchDone((_) => {
523 Queue.endUnsafe(queue)
524 return endVoid
525 })
526 )
527 )
528
529/**
530 * Creates a sink that publishes every consumed input element to a `PubSub`.
531 *
532 * **Details**
533 *
534 * The sink completes with `void` when the upstream stream ends.
535 *
536 * @category constructors
537 * @since 2.0.0
538 */
539export const fromPubSub = <A>(
540 pubsub: PubSub.PubSub<A>
541): Sink<void, A> => forEachArray((arr) => PubSub.publishAll(pubsub, arr))
542
543/**
544 * A sink that immediately ends with the specified value.
545 *
546 * **Example** (Succeeding with a value)
547 *
548 * ```ts import.meta.vitest
549 * import { Effect, Sink, Stream } from "effect"
550 *
551 * // Create a sink that always yields the same value
552 * const sink = Sink.succeed(42)
553 *
554 * // Use it with a stream
555 * const stream = Stream.make(1, 2, 3)
556 * await Effect.runPromise(Stream.run(stream, sink)) // => 42
557 * ```
558 *
559 * @category constructors
560 * @since 2.0.0
561 */
562export const succeed = <A, L = never>(a: A, leftovers?: NonEmptyReadonlyArray<L> | undefined): Sink<A, unknown, L> =>
563 fromEffectEnd(Effect.succeed([a, leftovers]))
564
565/**
566 * A sink that immediately ends with the specified lazily evaluated value.
567 *
568 * @category constructors
569 * @since 2.0.0
570 */
571export const sync = <A>(a: LazyArg<A>): Sink<A> => fromEffect(Effect.sync(a))
572
573/**
574 * A sink that is created from a lazily evaluated sink.
575 *
576 * @category constructors
577 * @since 2.0.0
578 */
579export const suspend = <A, In, L, E, R>(evaluate: LazyArg<Sink<A, In, L, E, R>>): Sink<A, In, L, E, R> =>
580 fromTransform((upstream, scope) => evaluate().transform(upstream, scope))
581
582/**
583 * A sink that always fails with the specified error.
584 *
585 * **Example** (Failing with an error)
586 *
587 * ```ts import.meta.vitest
588 * import { Effect, Exit, Sink, Stream } from "effect"
589 *
590 * // Create a sink that always fails
591 * const sink = Sink.fail("Sink failed")
592 *
593 * // Use it with a stream
594 * const stream = Stream.make(1, 2, 3)
595 * await Effect.runPromiseExit(Stream.run(stream, sink)) // => Exit.fail("Sink failed")
596 * ```
597 *
598 * @category constructors
599 * @since 2.0.0
600 */
601export const fail = <E>(e: E): Sink<never, unknown, never, E> => fromEffectEnd(Effect.fail(e))
602
603/**
604 * A sink that always fails with the specified lazily evaluated error.
605 *
606 * **Example** (Failing with a lazy error)
607 *
608 * ```ts import.meta.vitest
609 * import { Effect, Exit, Sink, Stream } from "effect"
610 *
611 * // Create a sink that fails with a lazy error
612 * const sink = Sink.failSync(() => "Lazy error")
613 *
614 * // Use it with a stream
615 * const stream = Stream.make(1, 2, 3)
616 * await Effect.runPromiseExit(Stream.run(stream, sink)) // => Exit.fail("Lazy error")
617 * ```
618 *
619 * @category constructors
620 * @since 2.0.0
621 */
622export const failSync = <E>(evaluate: LazyArg<E>): Sink<never, unknown, never, E> =>
623 fromEffectEnd(Effect.failSync(evaluate))
624
625/**
626 * Creates a sink halting with a specified `Cause`.
627 *
628 * **Example** (Failing with a cause)
629 *
630 * ```ts import.meta.vitest
631 * import { Cause, Effect, Exit, Sink, Stream } from "effect"
632 *
633 * // Create a sink that fails with a specific cause
634 * const sink = Sink.failCause(Cause.fail("Custom cause"))
635 *
636 * // Use it with a stream
637 * const stream = Stream.make(1, 2, 3)
638 * await Effect.runPromiseExit(Stream.run(stream, sink)) // => Exit.fail("Custom cause")
639 * ```
640 *
641 * @category constructors
642 * @since 2.0.0
643 */
644export const failCause = <E>(cause: Cause.Cause<E>): Sink<never, unknown, never, E> =>
645 fromEffectEnd(Effect.failCause(cause))
646
647/**
648 * Creates a sink halting with a specified lazily evaluated `Cause`.
649 *
650 * **Example** (Failing with a lazy cause)
651 *
652 * ```ts import.meta.vitest
653 * import { Cause, Effect, Exit, Sink, Stream } from "effect"
654 *
655 * // Create a sink that fails with a lazy cause
656 * const sink = Sink.failCauseSync(() => Cause.fail("Lazy cause"))
657 *
658 * // Use it with a stream
659 * const stream = Stream.make(1, 2, 3)
660 * await Effect.runPromiseExit(Stream.run(stream, sink)) // => Exit.fail("Lazy cause")
661 * ```
662 *
663 * @category constructors
664 * @since 2.0.0
665 */
666export const failCauseSync = <E>(evaluate: LazyArg<Cause.Cause<E>>): Sink<never, unknown, never, E> =>
667 fromEffectEnd(Effect.failCauseSync(evaluate))
668
669/**
670 * Creates a sink halting with a specified defect.
671 *
672 * **Example** (Dying with a defect)
673 *
674 * ```ts import.meta.vitest
675 * import { Effect, Exit, Sink, Stream } from "effect"
676 *
677 * // Create a sink that dies with a defect
678 * const sink = Sink.die("Defect error")
679 *
680 * // Use it with a stream
681 * const stream = Stream.make(1, 2, 3)
682 * await Effect.runPromiseExit(Stream.run(stream, sink)) // => Exit.die("Defect error")
683 * ```
684 *
685 * @category constructors
686 * @since 2.0.0
687 */
688export const die = (defect: unknown): Sink<never> => fromEffectEnd(Effect.die(defect))
689
690/**
691 * A sink that never completes.
692 *
693 * @category constructors
694 * @since 2.0.0
695 */
696export const never: Sink<unknown> = fromEffectEnd(Effect.never)
697
698/**
699 * Drops leftovers produced by a sink.
700 *
701 * **Details**
702 *
703 * The sink result is preserved, but any leftover elements are discarded
704 * instead of being returned to downstream sink composition. This does not
705 * continue pulling additional elements from the upstream stream.
706 *
707 * @category filtering
708 * @since 2.0.0
709 */
710export const ignoreLeftover = <A, In, L, E, R>(self: Sink<A, In, L, E, R>): Sink<A, In, never, E, R> =>
711 mapEnd(self, ([a]) => [a])
712
713/**
714 * Consumes and ignores all stream inputs.
715 *
716 * **When to use**
717 *
718 * Use to consume all upstream input and complete with void when the input
719 * values and any aggregate result are not needed.
720 *
721 * @see {@link count} for consuming all input while returning the number of elements
722 * @see {@link forEach} for consuming all input while running an effect for each element
723 *
724 * @category constructors
725 * @since 2.0.0
726 */
727export const drain: Sink<void, unknown> = fromTransform((upstream) =>
728 Pull.catchDone(
729 Effect.forever(upstream, { disableYield: true }),
730 () => endVoid
731 )
732)
733
734/**
735 * A sink that folds its inputs with the provided function, termination
736 * predicate and initial state.
737 *
738 * **When to use**
739 *
740 * Use to accumulate stream input element by element with an effectful step and
741 * stop based on the accumulated state.
742 *
743 * **Details**
744 *
745 * The initial state is evaluated lazily. Each input element is folded with the
746 * effectful function, and the sink continues while `contFn` returns `true`. If
747 * the sink stops in the middle of a pulled array, the remaining elements from
748 * that array are returned as leftovers.
749 *
750 * @see {@link foldArray} for folding each pulled non-empty input array at once
751 * @see {@link foldUntil} for folding until a fixed maximum number of elements is consumed
752 *
753 * @category folding
754 * @since 2.0.0
755 */
756export const fold = <S, In, E = never, R = never>(
757 s: LazyArg<S>,
758 contFn: Predicate<S>,
759 f: (s: S, input: In) => Effect.Effect<S, E, R>
760): Sink<S, In, In, E, R> =>
761 fromTransform((upstream) => {
762 let state = s()
763 return Effect.gen(function*() {
764 while (true) {
765 const arr = yield* upstream
766 for (let i = 0; i < arr.length; i++) {
767 state = yield* f(state, arr[i])
768 if (contFn(state)) continue
769 return [
770 state,
771 (i + 1) < arr.length ? (arr.slice(i + 1) as any) : undefined
772 ] as const
773 }
774 }
775 }).pipe(
776 Pull.catchDone(() => Effect.succeed<End<S, In>>([state]))
777 )
778 })
779
780/**
781 * Folds non-empty input arrays into state with an effectful function.
782 *
783 * **When to use**
784 *
785 * Use to update state with an effectful function once per pulled non-empty
786 * input array when batch-level processing is the natural unit.
787 *
788 * **Details**
789 *
790 * The initial state is evaluated lazily. After each pulled array is folded,
791 * the sink continues while `contFn` returns `true`; otherwise it completes
792 * with the current state.
793 *
794 * @see {@link fold} for folding element by element and returning leftovers when stopping mid-array
795 * @see {@link reduceWhileArrayEffect} for array-level effectful reducing that checks the predicate before consuming input
796 *
797 * @category folding
798 * @since 4.0.0
799 */
800export const foldArray = <S, In, E = never, R = never>(
801 s: LazyArg<S>,
802 contFn: Predicate<S>,
803 f: (s: S, input: Arr.NonEmptyReadonlyArray<In>) => Effect.Effect<S, E, R>
804): Sink<S, In, never, E, R> =>
805 fromTransform((upstream) => {
806 let state = s()
807 return Effect.gen(function*() {
808 while (true) {
809 const arr = yield* upstream
810 state = yield* f(state, arr)
811 if (contFn(state)) continue
812 return [state] as const
813 }
814 }).pipe(
815 Pull.catchDone(() => Effect.succeed<End<S>>([state]))
816 )
817 })
818
819/**
820 * Folds input elements into state until the specified maximum number of
821 * elements has been consumed or the upstream stream ends.
822 *
823 * **Details**
824 *
825 * Finite fractional values of `max` are rounded down. If `max` is `NaN` or
826 * non-positive, the sink completes with the initial state without consuming
827 * input.
828 *
829 * If the sink stops in the middle of a pulled array, the remaining elements
830 * from that array are returned as leftovers.
831 *
832 * @category folding
833 * @since 2.0.0
834 */
835export const foldUntil = <S, In, E = never, R = never>(
836 s: LazyArg<S>,
837 max: number,
838 f: (s: S, input: In) => Effect.Effect<S, E, R>
839): Sink<S, In, In, E, R> => {
840 const count = Count.normalize(max)
841 if (count === 0) return sync(s)
842 return fold<readonly [S, number], In, E, R>(
843 () => [s(), 0],
844 (tuple) => tuple[1] < count,
845 ([output, consumed], input) => Effect.map(f(output, input), (s) => [s, consumed + 1] as const)
846 ).pipe(
847 map((tuple) => tuple[0])
848 )
849}
850
851/**
852 * A sink that returns whether all elements satisfy the specified predicate.
853 *
854 * **When to use**
855 *
856 * Use to reduce a stream to a boolean that is true only when every input
857 * satisfies a pure predicate.
858 *
859 * @see {@link some} for the dual any-match check
860 *
861 * @category constructors
862 * @since 2.0.0
863 */
864export const every = <In>(predicate: Predicate<In>): Sink<boolean, In, In> =>
865 fold(
866 constTrue,
867 identity,
868 (_, a) => Effect.succeed(predicate(a))
869 )
870
871/**
872 * A sink that returns whether an element satisfies the specified predicate.
873 *
874 * **When to use**
875 *
876 * Use to reduce a stream to a boolean that is true when any input satisfies a
877 * pure predicate.
878 *
879 * @see {@link every} for the all-match check
880 *
881 * @category constructors
882 * @since 2.0.0
883 */
884export const some = <In>(predicate: Predicate<In>): Sink<boolean, In, In> =>
885 fold(
886 constFalse,
887 (b) => !b,
888 (_, a) => Effect.succeed(predicate(a))
889 )
890
891/**
892 * Transforms this sink's result.
893 *
894 * **When to use**
895 *
896 * Use to compute a new result from the original sink result while preserving
897 * the sink's input consumption behavior.
898 *
899 * **Details**
900 *
901 * The transformed sink preserves the original sink's input type, leftovers,
902 * errors, and requirements.
903 *
904 * @see {@link mapEffect} for effectful result transformations
905 * @see {@link as} for replacing the result with a constant value
906 * @see {@link mapEnd} for transforming both the result and leftovers
907 *
908 * @category mapping
909 * @since 2.0.0
910 */
911export const map: {
912 <A, A2>(f: (a: A) => A2): <In, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A2, In, L, E, R>
913 <A, In, L, E, R, A2>(self: Sink<A, In, L, E, R>, f: (a: A) => A2): Sink<A2, In, L, E, R>
914} = dual(
915 2,
916 <A, In, L, E, R, A2>(self: Sink<A, In, L, E, R>, f: (a: A) => A2): Sink<A2, In, L, E, R> =>
917 mapEnd(self, ([a, l]) => [f(a), l])
918)
919
920/**
921 * Sets the sink's result to a constant value.
922 *
923 * **When to use**
924 *
925 * Use to keep a sink's input consumption, errors, requirements, and leftovers
926 * while replacing only its result with a known value.
927 *
928 * @see {@link map} for computing the replacement from the original result
929 *
930 * @category mapping
931 * @since 2.0.0
932 */
933export const as: {
934 <A2>(a2: A2): <A, In, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A2, In, L, E, R>
935 <A, In, L, E, R, A2>(self: Sink<A, In, L, E, R>, a2: A2): Sink<A2, In, L, E, R>
936} = dual(
937 2,
938 <A, In, L, E, R, A2>(self: Sink<A, In, L, E, R>, a2: A2): Sink<A2, In, L, E, R> => map(self, () => a2)
939)
940
941/**
942 * Transforms this sink's input elements.
943 *
944 * @category mapping
945 * @since 2.0.0
946 */
947export const mapInput: {
948 <In0, In>(f: (input: In0) => In): <A, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A, In0, L, E, R>
949 <A, In, L, E, R, In0>(self: Sink<A, In, L, E, R>, f: (input: In0) => In): Sink<A, In0, L, E, R>
950} = dual(
951 2,
952 <A, In, L, E, R, In0>(self: Sink<A, In, L, E, R>, f: (input: In0) => In): Sink<A, In0, L, E, R> =>
953 mapInputArray(self, Arr.map(f))
954)
955
956/**
957 * Transforms this sink's input elements effectfully.
958 *
959 * @category mapping
960 * @since 2.0.0
961 */
962export const mapInputEffect: {
963 <In0, In, E2, R2>(
964 f: (input: In0) => Effect.Effect<In, E2, R2>
965 ): <A, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A, In0, L, E2 | E, R2 | R>
966 <A, In, L, E, R, In0, E2, R2>(
967 self: Sink<A, In, L, E, R>,
968 f: (input: In0) => Effect.Effect<In, E2, R2>
969 ): Sink<A, In0, L, E | E2, R | R2>
970} = dual(
971 2,
972 <A, In, L, E, R, In0, E2, R2>(
973 self: Sink<A, In, L, E, R>,
974 f: (input: In0) => Effect.Effect<In, E2, R2>
975 ): Sink<A, In0, L, E | E2, R | R2> => mapInputArrayEffect(self, Effect.forEach(f))
976)
977
978/**
979 * Transforms each non-empty array of upstream input before it is fed to this
980 * sink.
981 *
982 * @category mapping
983 * @since 4.0.0
984 */
985export const mapInputArray: {
986 <In0, In>(
987 f: (input: Arr.NonEmptyReadonlyArray<In0>) => Arr.NonEmptyReadonlyArray<In>
988 ): <A, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A, In0, L, E, R>
989 <A, In, L, E, R, In0>(
990 self: Sink<A, In, L, E, R>,
991 f: (input: Arr.NonEmptyReadonlyArray<In0>) => Arr.NonEmptyReadonlyArray<In>
992 ): Sink<A, In0, L, E, R>
993} = dual(
994 2,
995 <A, In, L, E, R, In0>(
996 self: Sink<A, In, L, E, R>,
997 f: (input: Arr.NonEmptyReadonlyArray<In0>) => Arr.NonEmptyReadonlyArray<In>
998 ): Sink<A, In0, L, E, R> => fromTransform((upstream, scope) => self.transform(Effect.map(upstream, f), scope))
999)
1000
1001/**
1002 * Transforms each non-empty array of upstream input effectfully before it is
1003 * fed to this sink.
1004 *
1005 * @category mapping
1006 * @since 4.0.0
1007 */
1008export const mapInputArrayEffect: {
1009 <In0, In, E2, R2>(
1010 f: (input: Arr.NonEmptyReadonlyArray<In0>) => Effect.Effect<Arr.NonEmptyReadonlyArray<In>, E2, R2>
1011 ): <A, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A, In0, L, E2 | E, R2 | R>
1012 <A, In, L, E, R, In0, E2, R2>(
1013 self: Sink<A, In, L, E, R>,
1014 f: (input: Arr.NonEmptyReadonlyArray<In0>) => Effect.Effect<Arr.NonEmptyReadonlyArray<In>, E2, R2>
1015 ): Sink<A, In0, L, E | E2, R | R2>
1016} = dual(
1017 2,
1018 <A, In, L, E, R, In0, E2, R2>(
1019 self: Sink<A, In, L, E, R>,
1020 f: (input: Arr.NonEmptyReadonlyArray<In0>) => Effect.Effect<Arr.NonEmptyReadonlyArray<In>, E2, R2>
1021 ): Sink<A, In0, L, E | E2, R | R2> =>
1022 fromTransform((upstream, scope) =>
1023 self.transform(
1024 Effect.flatMap(upstream, f) as any,
1025 scope
1026 )
1027 )
1028)
1029
1030/**
1031 * Transforms the full `End` produced by this sink.
1032 *
1033 * **Details**
1034 *
1035 * This can change both the result value and the optional leftovers.
1036 *
1037 * @category mapping
1038 * @since 4.0.0
1039 */
1040export const mapEnd: {
1041 <A, L, A2, L2 = never>(
1042 f: (a: End<A, L>) => End<A2, L2>
1043 ): <In, E, R>(self: Sink<A, In, L, E, R>) => Sink<A2, In, L2, E, R>
1044 <A, In, L, E, R, A2, L2 = never>(self: Sink<A, In, L, E, R>, f: (a: End<A, L>) => End<A2, L2>): Sink<A2, In, L2, E, R>
1045} = dual(
1046 2,
1047 <A, In, L, E, R, A2, L2 = never>(
1048 self: Sink<A, In, L, E, R>,
1049 f: (a: End<A, L>) => End<A2, L2>
1050 ): Sink<A2, In, L2, E, R> =>
1051 fromTransform((upstream, scope) =>
1052 Effect.map(
1053 self.transform(upstream, scope),
1054 f
1055 )
1056 )
1057)
1058
1059const transformEffect = <A, In, L, E, R, A2, E2, R2, L2 = never>(
1060 self: Sink<A, In, L, E, R>,
1061 f: (effect: Effect.Effect<End<A, L>, E, R>) => Effect.Effect<End<A2, L2>, E2, R2>
1062): Sink<A2, In, L2, E2, R2> => fromTransform((upstream, scope) => f(self.transform(upstream, scope)))
1063
1064/**
1065 * Transforms the full `End` produced by this sink effectfully.
1066 *
1067 * **Details**
1068 *
1069 * This can change both the result value and the optional leftovers, and the
1070 * transformation can fail or require services.
1071 *
1072 * @category mapping
1073 * @since 4.0.0
1074 */
1075export const mapEffectEnd: {
1076 <A, L, A2, E2, R2, L2 = never>(
1077 f: (end: End<A, L>) => Effect.Effect<End<A2, L2>, E2, R2>
1078 ): <In, E, R>(self: Sink<A, In, L, E, R>) => Sink<A2, In, L2, E2 | E, R2 | R>
1079 <A, In, L, E, R, A2, E2, R2, L2 = never>(
1080 self: Sink<A, In, L, E, R>,
1081 f: (end: End<A, L>) => Effect.Effect<End<A2, L2>, E2, R2>
1082 ): Sink<A2, In, L2, E | E2, R | R2>
1083} = dual(2, <A, In, L, E, R, A2, E2, R2, L2 = never>(
1084 self: Sink<A, In, L, E, R>,
1085 f: (end: End<A, L>) => Effect.Effect<End<A2, L2>, E2, R2>
1086): Sink<A2, In, L2, E | E2, R | R2> => transformEffect(self, Effect.flatMap(f)))
1087
1088/**
1089 * Transforms this sink's result effectfully.
1090 *
1091 * **When to use**
1092 *
1093 * Use when you need a sink result transformation that is effectful, can fail,
1094 * or requires services.
1095 *
1096 * **Details**
1097 *
1098 * The transformed sink preserves the original sink's input consumption and
1099 * leftovers while adding the errors and requirements of the transformation.
1100 *
1101 * @see {@link map} for pure result transformations
1102 * @see {@link mapEffectEnd} for effectfully transforming both the result and leftovers
1103 * @see {@link flatMap} for continuing with another sink based on the result
1104 *
1105 * @category mapping
1106 * @since 2.0.0
1107 */
1108export const mapEffect: {
1109 <A, A2, E2, R2>(
1110 f: (a: A) => Effect.Effect<A2, E2, R2>
1111 ): <In, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A2, In, L, E2 | E, R2 | R>
1112 <A, In, L, E, R, A2, E2, R2>(
1113 self: Sink<A, In, L, E, R>,
1114 f: (a: A) => Effect.Effect<A2, E2, R2>
1115 ): Sink<A2, In, L, E | E2, R | R2>
1116} = dual(2, <A, In, L, E, R, A2, E2, R2>(
1117 self: Sink<A, In, L, E, R>,
1118 f: (a: A) => Effect.Effect<A2, E2, R2>
1119): Sink<A2, In, L, E | E2, R | R2> => mapEffectEnd(self, ([a, l]) => Effect.map(f(a), (a2) => [a2, l] as End<A2, L>)))
1120
1121/**
1122 * Transforms the errors emitted by this sink using `f`.
1123 *
1124 * @category mapping
1125 * @since 2.0.0
1126 */
1127export const mapError: {
1128 <E, E2>(f: (error: E) => E2): <A, In, L, R>(self: Sink<A, In, L, E, R>) => Sink<A, In, L, E2, R>
1129 <A, In, L, E, R, E2>(self: Sink<A, In, L, E, R>, f: (error: E) => E2): Sink<A, In, L, E2, R>
1130} = dual(2, <A, In, L, E, R, E2>(
1131 self: Sink<A, In, L, E, R>,
1132 f: (error: E) => E2
1133): Sink<A, In, L, E2, R> => transformEffect(self, Effect.mapError(f)))
1134
1135/**
1136 * Transforms the leftovers emitted by this sink using `f`.
1137 *
1138 * @category mapping
1139 * @since 2.0.0
1140 */
1141export const mapLeftover: {
1142 <L, L2>(f: (leftover: L) => L2): <A, In, E, R>(self: Sink<A, In, L, E, R>) => Sink<A, In, L2, E, R>
1143 <A, In, L, E, R, L2>(self: Sink<A, In, L, E, R>, f: (leftover: L) => L2): Sink<A, In, L2, E, R>
1144} = dual(2, <A, In, L, E, R, L2>(
1145 self: Sink<A, In, L, E, R>,
1146 f: (leftover: L) => L2
1147): Sink<A, In, L2, E, R> => mapEnd(self, ([a, l]) => [a, l && Arr.map(l, f)]))
1148
1149/**
1150 * Collects up to `n` input elements into an array.
1151 *
1152 * **Details**
1153 *
1154 * Finite fractional values of `n` are rounded down. If `n` is `NaN` or
1155 * non-positive, the sink completes with an empty array. If more elements are
1156 * pulled than needed, the remaining elements from the same array are returned
1157 * as leftovers.
1158 *
1159 * @category constructors
1160 * @since 2.0.0
1161 */
1162export const take = <In>(n: number): Sink<Array<In>, In, In> => {
1163 const count = Count.normalize(n)
1164 return fromTransform((upstream) => {
1165 const taken: Array<In> = []
1166 if (count === 0) {
1167 return Effect.succeed([taken] as const)
1168 }
1169 let leftover: NonEmptyReadonlyArray<In> | undefined = undefined
1170 return upstream.pipe(
1171 Effect.flatMap((arr) => {
1172 if (taken.length + arr.length <= count) {
1173 taken.push(...arr)
1174 if (taken.length === count) {
1175 return Cause.done()
1176 }
1177 return Effect.void
1178 }
1179 for (let i = 0; i < arr.length; i++) {
1180 taken.push(arr[i])
1181 if (taken.length === count) {
1182 if ((i + 1) < arr.length) {
1183 leftover = arr.slice(i + 1) as any
1184 }
1185 return Cause.done()
1186 }
1187 }
1188 return Effect.void
1189 }),
1190 Effect.forever({ disableYield: true }),
1191 Pull.catchDone(() => Effect.succeed([taken, leftover] as const))
1192 )
1193 })
1194}
1195
1196/**
1197 * Runs this sink until it yields a result, then uses that result to create
1198 * another sink from the provided function which will continue to run until it
1199 * yields a result.
1200 *
1201 * **When to use**
1202 *
1203 * Use to compose sinks when the next sink depends on the result produced by the
1204 * previous sink.
1205 *
1206 * **Details**
1207 *
1208 * Leftovers from the first sink are fed to the sink returned by `f` before more
1209 * upstream input is pulled.
1210 *
1211 * @see {@link map} for transforming the result without switching sinks
1212 * @see {@link mapEffect} for effectfully transforming the result without switching sinks
1213 *
1214 * @category sequencing
1215 * @since 2.0.0
1216 */
1217export const flatMap: {
1218 <A, A1, L, In1 extends L, L1, E1, R1>(
1219 f: (a: A) => Sink<A1, In1, L1, E1, R1>
1220 ): <In, E, R>(self: Sink<A, In, L, E, R>) => Sink<A1, In & In1, L1 | L, E1 | E, R1 | R>
1221 <A, In, L, E, R, A1, In1 extends L, L1, E1, R1>(
1222 self: Sink<A, In, L, E, R>,
1223 f: (a: A) => Sink<A1, In1, L1, E1, R1>
1224 ): Sink<A1, In & In1, L | L1, E | E1, R | R1>
1225} = dual(2, <A, In, L, E, R, A1, In1 extends L, L1, E1, R1>(
1226 self: Sink<A, In, L, E, R>,
1227 f: (a: A) => Sink<A1, In1, L1, E1, R1>
1228): Sink<A1, In & In1, L | L1, E | E1, R | R1> =>
1229 fromTransform((upstream, scope) => {
1230 let upstreamDone = false
1231 const pull = Effect.catchCause(upstream, (cause) => {
1232 upstreamDone = true
1233 return Effect.failCause(cause)
1234 })
1235 return Effect.flatMap(
1236 self.transform(pull, scope),
1237 ([a, leftover]) =>
1238 f(a).transform(
1239 Effect.suspend(() => {
1240 if (leftover) {
1241 const arr = leftover as Arr.NonEmptyReadonlyArray<In1>
1242 leftover = undefined
1243 return Effect.succeed(arr)
1244 } else if (upstreamDone) {
1245 return Cause.done()
1246 }
1247 return upstream
1248 }),
1249 scope
1250 ).pipe(Effect.map((end): End<A1, L | L1> => {
1251 if (!leftover) {
1252 return end
1253 }
1254 return [end[0], end[1] ? [...end[1], ...leftover] : leftover]
1255 }))
1256 )
1257 }))
1258
1259/**
1260 * A sink that reduces input elements from the provided `initial` state with
1261 * `f` while the specified `predicate` returns `true`.
1262 *
1263 * @category folding
1264 * @since 4.0.0
1265 */
1266export const reduceWhile = <S, In>(
1267 initial: LazyArg<S>,
1268 predicate: Predicate<S>,
1269 f: (s: S, input: In) => S
1270): Sink<S, In, In> =>
1271 fromTransform((upstream) => {
1272 let state = initial()
1273 let leftover: NonEmptyReadonlyArray<In> | undefined = undefined
1274 if (!predicate(state)) {
1275 return Effect.succeed([state] as const)
1276 }
1277 return upstream.pipe(
1278 Effect.flatMap((arr) => {
1279 for (let i = 0; i < arr.length; i++) {
1280 state = f(state, arr[i])
1281 if (!predicate(state)) {
1282 if ((i + 1) < arr.length) {
1283 leftover = arr.slice(i + 1) as any
1284 }
1285 return Cause.done()
1286 }
1287 }
1288 return Effect.void
1289 }),
1290 Effect.forever({ disableYield: true }),
1291 Pull.catchDone(() => Effect.succeed([state, leftover] as const))
1292 )
1293 })
1294
1295/**
1296 * A sink that effectfully reduces input elements from the provided `initial`
1297 * state with `f` while the specified `predicate` returns `true`.
1298 *
1299 * @category folding
1300 * @since 4.0.0
1301 */
1302export const reduceWhileEffect = <S, In, E, R>(
1303 initial: LazyArg<S>,
1304 predicate: Predicate<S>,
1305 f: (s: S, input: In) => Effect.Effect<S, E, R>
1306): Sink<S, In, In, E, R> =>
1307 fromTransform((upstream) => {
1308 let state = initial()
1309 let leftover: NonEmptyReadonlyArray<In> | undefined = undefined
1310 if (!predicate(state)) {
1311 return Effect.succeed([state] as const)
1312 }
1313 return upstream.pipe(
1314 Effect.flatMap((arr) => {
1315 let i = 0
1316 return Effect.whileLoop({
1317 while: () => i < arr.length,
1318 body: constant(Effect.flatMap(Effect.suspend(() => f(state, arr[i++])), (s) => {
1319 state = s
1320 if (!predicate(state)) {
1321 if (i < arr.length) {
1322 leftover = arr.slice(i) as any
1323 }
1324 return Cause.done()
1325 }
1326 return Effect.void
1327 })),
1328 step: constVoid
1329 })
1330 }),
1331 Effect.forever({ disableYield: true }),
1332 Pull.catchDone(() => Effect.succeed([state, leftover] as const))
1333 )
1334 })
1335
1336/**
1337 * A sink that reduces non-empty input arrays from the provided `initial` state
1338 * with `f` while the specified `predicate` returns `true`.
1339 *
1340 * @category folding
1341 * @since 4.0.0
1342 */
1343export const reduceWhileArray = <S, In>(
1344 initial: LazyArg<S>,
1345 contFn: Predicate<S>,
1346 f: (s: S, input: NonEmptyReadonlyArray<In>) => S
1347): Sink<S, In> =>
1348 fromTransform((upstream) => {
1349 let state = initial()
1350 if (!contFn(state)) {
1351 return Effect.succeed([state] as const)
1352 }
1353 return upstream.pipe(
1354 Effect.flatMap((arr) => {
1355 state = f(state, arr)
1356 if (!contFn(state)) {
1357 return Cause.done()
1358 }
1359 return Effect.void
1360 }),
1361 Effect.forever({ disableYield: true }),
1362 Pull.catchDone(() => Effect.succeed([state] as const))
1363 )
1364 })
1365
1366/**
1367 * A sink that effectfully reduces non-empty input arrays from the provided
1368 * `initial` state with `f` while the specified `predicate` returns `true`.
1369 *
1370 * @category folding
1371 * @since 4.0.0
1372 */
1373export const reduceWhileArrayEffect = <S, In, E, R>(
1374 initial: LazyArg<S>,
1375 predicate: Predicate<S>,
1376 f: (s: S, input: NonEmptyReadonlyArray<In>) => Effect.Effect<S, E, R>
1377): Sink<S, In, never, E, R> =>
1378 fromTransform((upstream) => {
1379 let state = initial()
1380 if (!predicate(state)) {
1381 return Effect.succeed([state] as const)
1382 }
1383 return upstream.pipe(
1384 Effect.flatMap((arr) => f(state, arr)),
1385 Effect.flatMap((s) => {
1386 state = s
1387 if (!predicate(state)) {
1388 return Cause.done()
1389 }
1390 return Effect.void
1391 }),
1392 Effect.forever({ disableYield: true }),
1393 Pull.catchDone(() => Effect.succeed([state] as const))
1394 )
1395 })
1396
1397/**
1398 * A sink that reduces its inputs using the provided function `f` starting from
1399 * the provided `initial` state.
1400 *
1401 * @category folding
1402 * @since 4.0.0
1403 */
1404export const reduce = <S, In>(initial: LazyArg<S>, f: (s: S, input: In) => S): Sink<S, In> =>
1405 reduceArray(initial, (s, arr) => {
1406 for (let i = 0; i < arr.length; i++) {
1407 s = f(s, arr[i])
1408 }
1409 return s
1410 })
1411
1412/**
1413 * A sink that reduces its inputs using the provided function `f` starting from
1414 * the specified `initial` state.
1415 *
1416 * @category folding
1417 * @since 4.0.0
1418 */
1419export const reduceArray = <S, In>(
1420 initial: LazyArg<S>,
1421 f: (s: S, input: NonEmptyReadonlyArray<In>) => S
1422): Sink<S, In> =>
1423 fromTransform((upstream) => {
1424 let state = initial()
1425 return upstream.pipe(
1426 Effect.flatMap((arr) => {
1427 state = f(state, arr)
1428 return Effect.void
1429 }),
1430 Effect.forever({ disableYield: true }),
1431 Pull.catchDone(() => Effect.succeed([state] as const))
1432 )
1433 })
1434
1435/**
1436 * A sink that reduces its inputs using the provided effectful function `f`
1437 * starting from the specified `initial` state.
1438 *
1439 * @category folding
1440 * @since 4.0.0
1441 */
1442export const reduceEffect = <S, In, E, R>(
1443 initial: LazyArg<S>,
1444 f: (s: S, input: In) => Effect.Effect<S, E, R>
1445): Sink<S, In, never, E, R> => reduceWhileEffect(initial, constTrue, f) as any
1446
1447const head_ = reduceWhile(Option.none<unknown>, Option.isNone, (_, in_) => Option.some(in_))
1448
1449/**
1450 * Creates a sink containing the first value.
1451 *
1452 * **Details**
1453 *
1454 * Returns `Option.some(first)` for non-empty input, or `Option.none` when the
1455 * upstream ends without input. The first element is consumed; later elements
1456 * from the same pulled array are emitted as leftovers.
1457 *
1458 * @category constructors
1459 * @since 2.0.0
1460 */
1461export const head = <In>(): Sink<Option.Option<In>, In, In> => head_ as any
1462
1463const last_ = reduceArray(Option.none<unknown>, (_, arr) => Arr.last(arr))
1464
1465/**
1466 * Creates a sink containing the last value.
1467 *
1468 * **When to use**
1469 *
1470 * Use when you need to consume all upstream input and keep only the final
1471 * element.
1472 *
1473 * **Details**
1474 *
1475 * Returns `Option.some(last)` with the final input value, or `Option.none` when
1476 * the upstream ends without input.
1477 *
1478 * **Gotchas**
1479 *
1480 * This sink produces a result only when the upstream ends, so it does not
1481 * complete for a stream that does not end.
1482 *
1483 * @see {@link head} for taking the first input value instead
1484 *
1485 * @category constructors
1486 * @since 2.0.0
1487 */
1488export const last = <In>(): Sink<Option.Option<In>, In> => last_ as any
1489
1490/**
1491 * Creates a sink containing the first value matched by a synchronous predicate.
1492 *
1493 * **When to use**
1494 *
1495 * Use to scan stream input until the first matching element is found and return
1496 * that element as an `Option`.
1497 *
1498 * **Details**
1499 *
1500 * Returns `Option.none` if the upstream stream ends before a match is found.
1501 * Refinement predicates narrow the returned value type. The matching input is
1502 * consumed; any later elements from the same pulled array are returned as
1503 * leftovers.
1504 *
1505 * @see {@link findEffect} for an effectful predicate that can fail or require services
1506 *
1507 * @category constructors
1508 * @since 4.0.0
1509 */
1510export const find: {
1511 <In, Out extends In>(refinement: Refinement<In, Out>): Sink<Option.Option<Out>, In, In>
1512 <In>(predicate: Predicate<In>): Sink<Option.Option<In>, In, In>
1513} = <In>(predicate: Predicate<In>): Sink<Option.Option<In>, In, In> =>
1514 reduceWhile(
1515 Option.none<In>,
1516 Option.isNone,
1517 (acc, in_) => predicate(in_) ? Option.some(in_) : acc
1518 )
1519
1520/**
1521 * Creates a sink containing the first value matched by an effectful predicate.
1522 *
1523 * **When to use**
1524 *
1525 * Use when you need to run effects, fail, or use services while searching for
1526 * the first matching input.
1527 *
1528 * **Details**
1529 *
1530 * Returns `Option.some` with the first input whose predicate result is `true`,
1531 * or `Option.none` if the upstream stream ends first. If the predicate effect
1532 * fails, the sink fails with the same error.
1533 *
1534 * @see {@link find} for the synchronous predicate variant
1535 *
1536 * @category constructors
1537 * @since 2.0.0
1538 */
1539export const findEffect = <In, E, R>(
1540 predicate: (input: In) => Effect.Effect<boolean, E, R>
1541): Sink<Option.Option<In>, In, In, E, R> =>
1542 reduceWhileEffect(
1543 Option.none<In>,
1544 Option.isNone,
1545 (acc, in_) => Effect.map(predicate(in_), (b) => b ? Option.some(in_) : acc)
1546 )
1547
1548/**
1549 * Creates a sink which sums up its inputs.
1550 *
1551 * @category constructors
1552 * @since 2.0.0
1553 */
1554export const sum: Sink<number, number> = reduceArray(() => 0, (s, arr) => {
1555 for (let i = 0; i < arr.length; i++) {
1556 s += arr[i]
1557 }
1558 return s
1559})
1560
1561/**
1562 * A sink that counts the number of elements fed to it.
1563 *
1564 * **When to use**
1565 *
1566 * Use to consume input and return only the number of elements received.
1567 *
1568 * @category constructors
1569 * @since 2.0.0
1570 */
1571export const count: Sink<number, unknown> = reduceArray(() => 0, (s, arr) => s + arr.length)
1572
1573/**
1574 * Accumulates incoming elements into an array.
1575 *
1576 * **When to use**
1577 *
1578 * Use when you need a sink result containing all upstream input elements.
1579 *
1580 * @see {@link take} for collecting only a fixed number of input elements
1581 *
1582 * @category constructors
1583 * @since 4.0.0
1584 */
1585export const collect = <In>(): Sink<Array<In>, In> =>
1586 reduceArray(Arr.empty<In>, (s, arr) => {
1587 s.push(...arr)
1588 return s
1589 })
1590
1591/**
1592 * Collects the longest input prefix whose elements satisfy the predicate or
1593 * refinement.
1594 *
1595 * **Details**
1596 *
1597 * The first failing input is consumed and excluded from the result. Any later
1598 * elements from the same pulled array are returned as leftovers.
1599 *
1600 * @category constructors
1601 * @since 4.0.0
1602 */
1603export const takeWhile: {
1604 <In, Out extends In>(refinement: Refinement<In, Out>): Sink<Array<Out>, In, In>
1605 <In>(predicate: Predicate<In>): Sink<Array<In>, In, In>
1606} = <In>(predicate: Predicate<In>): Sink<Array<In>, In, In> =>
1607 fromTransform((upstream) => {
1608 const out = Arr.empty<In>()
1609 return upstream.pipe(
1610 Effect.flatMap((arr) => {
1611 for (let i = 0; i < arr.length; i++) {
1612 if (!predicate(arr[i])) {
1613 const leftover: Arr.NonEmptyReadonlyArray<In> | undefined = (i + 1) < arr.length
1614 ? arr.slice(i + 1) as any
1615 : undefined
1616 return Cause.done([out, leftover] as const)
1617 }
1618 out.push(arr[i])
1619 }
1620 return Effect.void
1621 }),
1622 Effect.forever({ disableYield: true }),
1623 Pull.catchDone((end) => Effect.succeed<End<Array<In>, In>>(end ?? [out]))
1624 )
1625 })
1626
1627/**
1628 * Applies a `Filter` to input elements while it succeeds, collecting each
1629 * successful output.
1630 *
1631 * **Details**
1632 *
1633 * The first input for which the filter fails is consumed and excluded from the
1634 * result. Any later elements from the same pulled array are returned as
1635 * leftovers.
1636 *
1637 * @category constructors
1638 * @since 4.0.0
1639 */
1640export const takeWhileFilter = <In, Out, X>(
1641 filter: Filter.Filter<In, Out, X>
1642): Sink<Array<Out>, In, In> =>
1643 fromTransform((upstream) => {
1644 const out = Arr.empty<Out>()
1645 return upstream.pipe(
1646 Effect.flatMap((arr) => {
1647 for (let i = 0; i < arr.length; i++) {
1648 const result = filter(arr[i])
1649 if (Result.isFailure(result)) {
1650 const leftover: Arr.NonEmptyReadonlyArray<In> | undefined = (i + 1) < arr.length
1651 ? arr.slice(i + 1) as any
1652 : undefined
1653 return Cause.done([out, leftover] as const)
1654 }
1655 out.push(result.success)
1656 }
1657 return Effect.void
1658 }),
1659 Effect.forever({ disableYield: true }),
1660 Pull.catchDone((end) => Effect.succeed<End<Array<Out>, In>>(end ?? [out]))
1661 )
1662 })
1663
1664/**
1665 * Collects input elements effectfully while the predicate succeeds.
1666 *
1667 * **Details**
1668 *
1669 * The first input for which the predicate returns `false` is consumed and
1670 * excluded from the result. Any later elements from the same pulled array are
1671 * returned as leftovers.
1672 *
1673 * @category constructors
1674 * @since 4.0.0
1675 */
1676export const takeWhileEffect: {
1677 <In, E, R>(predicate: (input: In) => Effect.Effect<boolean, E, R>): Sink<Array<In>, In, In, E, R>
1678} = <In, E, R>(
1679 predicate: (input: In) => Effect.Effect<boolean, E, R>
1680): Sink<Array<In>, In, In, E, R> =>
1681 fromTransform((upstream) => {
1682 const out = Arr.empty<In>()
1683 let leftover: Arr.NonEmptyReadonlyArray<In> | undefined = undefined
1684 return upstream.pipe(
1685 Effect.flatMap((arr) => {
1686 let i = 0
1687 return Effect.whileLoop({
1688 while: () => i < arr.length,
1689 body: constant(Effect.flatMap(
1690 Effect.suspend(() => {
1691 const input = arr[i++]
1692 return Effect.map(predicate(input), (passes) => [input, passes] as const)
1693 }),
1694 ([input, passes]) => {
1695 if (!passes) {
1696 if (i < arr.length) {
1697 leftover = arr.slice(i) as any
1698 }
1699 return Cause.done()
1700 }
1701 out.push(input)
1702 return Effect.void
1703 }
1704 )),
1705 step: constVoid
1706 })
1707 }),
1708 Effect.forever({ disableYield: true }),
1709 Pull.catchDone(() => Effect.succeed([out, leftover] as const))
1710 )
1711 })
1712
1713/**
1714 * Applies a `FilterEffect` to input elements effectfully while it succeeds,
1715 * collecting each successful output.
1716 *
1717 * **Details**
1718 *
1719 * The first input for which the filter fails is consumed and excluded from the
1720 * result. Any later elements from the same pulled array are returned as
1721 * leftovers.
1722 *
1723 * @category constructors
1724 * @since 4.0.0
1725 */
1726export const takeWhileFilterEffect = <In, Out, X, E, R>(
1727 filter: Filter.FilterEffect<In, Out, X, E, R>
1728): Sink<Array<Out>, In, In, E, R> =>
1729 fromTransform((upstream) => {
1730 const out = Arr.empty<Out>()
1731 let leftover: Arr.NonEmptyReadonlyArray<In> | undefined = undefined
1732 return upstream.pipe(
1733 Effect.flatMap((arr) => {
1734 let i = 0
1735 return Effect.whileLoop({
1736 while: () => i < arr.length,
1737 body: constant(Effect.flatMap(Effect.suspend(() => filter(arr[i++])), (result) => {
1738 if (Result.isFailure(result)) {
1739 if (i < arr.length) {
1740 leftover = arr.slice(i) as any
1741 }
1742 return Cause.done()
1743 }
1744 out.push(result.success)
1745 return Effect.void
1746 })),
1747 step: constVoid
1748 })
1749 }),
1750 Effect.forever({ disableYield: true }),
1751 Pull.catchDone(() => Effect.succeed([out, leftover] as const))
1752 )
1753 })
1754
1755/**
1756 * Collects input elements until the predicate returns `true`, including the
1757 * matching element in the result.
1758 *
1759 * @category constructors
1760 * @since 4.0.0
1761 */
1762export const takeUntil = <In>(predicate: Predicate<In>): Sink<Array<In>, In, In> =>
1763 suspend(() => {
1764 let done = false
1765 return takeWhile((i) => {
1766 if (done) return false
1767 done = predicate(i)
1768 return true
1769 })
1770 })
1771
1772/**
1773 * Collects input elements effectfully until the predicate returns `true`,
1774 * including the matching element in the result.
1775 *
1776 * **Details**
1777 *
1778 * If the predicate effect fails, the sink fails with the same error.
1779 *
1780 * @category constructors
1781 * @since 4.0.0
1782 */
1783export const takeUntilEffect = <In, E, R>(
1784 predicate: (input: In) => Effect.Effect<boolean, E, R>
1785): Sink<Array<In>, In, In, E, R> =>
1786 suspend(() => {
1787 let done = false
1788 return takeWhileEffect((input) => {
1789 if (done) {
1790 return Effect.succeed(false)
1791 }
1792 return Effect.map(predicate(input), (b) => {
1793 done = b
1794 return true
1795 })
1796 })
1797 })
1798
1799/**
1800 * A sink that executes the provided effectful function for every item fed
1801 * to it.
1802 *
1803 * **Example** (Running effects for each item)
1804 *
1805 * ```ts import.meta.vitest
1806 * import { Effect, Sink, Stream } from "effect"
1807 *
1808 * const processed: Array<number> = []
1809 * const sink = Sink.forEach((item: number) => Effect.sync(() => processed.push(item)))
1810 *
1811 * // Use it with a stream
1812 * const stream = Stream.make(1, 2, 3)
1813 * await Effect.runPromise(Stream.run(stream, sink))
1814 * processed // => [1, 2, 3]
1815 * ```
1816 *
1817 * @category constructors
1818 * @since 2.0.0
1819 */
1820export const forEach = <In, X, E, R>(
1821 f: (input: In) => Effect.Effect<X, E, R>
1822): Sink<void, In, never, E, R> => forEachArray(Effect.forEach((_) => f(_), { discard: true }))
1823
1824/**
1825 * A sink that executes the provided effectful function for every Chunk fed
1826 * to it.
1827 *
1828 * **Example** (Running effects for each chunk)
1829 *
1830 * ```ts import.meta.vitest
1831 * import { Effect, Sink, Stream } from "effect"
1832 *
1833 * const processed: Array<Array<number>> = []
1834 * const sink = Sink.forEachArray((chunk: ReadonlyArray<number>) => Effect.sync(() => processed.push([...chunk])))
1835 *
1836 * // Use it with a stream
1837 * const stream = Stream.make(1, 2, 3, 4, 5)
1838 * await Effect.runPromise(Stream.run(stream, sink))
1839 * processed // => [[1, 2, 3, 4, 5]]
1840 * ```
1841 *
1842 * @category constructors
1843 * @since 4.0.0
1844 */
1845export const forEachArray = <In, X, E, R>(
1846 f: (input: NonEmptyReadonlyArray<In>) => Effect.Effect<X, E, R>
1847): Sink<void, In, never, E, R> =>
1848 fromTransform((upstream) =>
1849 upstream.pipe(
1850 Effect.flatMap(f),
1851 Effect.forever({ disableYield: true }),
1852 Pull.catchDone(() => endVoid)
1853 )
1854 )
1855
1856/**
1857 * Runs an effectful function for each input element while it returns `true`.
1858 *
1859 * **Details**
1860 *
1861 * The sink stops consuming input when the function returns `false` or when the
1862 * upstream stream ends, and completes with `void`.
1863 *
1864 * @category constructors
1865 * @since 2.0.0
1866 */
1867export const forEachWhile = <In, E, R>(
1868 f: (input: In) => Effect.Effect<boolean, E, R>
1869): Sink<void, In, never, E, R> =>
1870 forEachWhileArray(Effect.fnUntraced(function*(input) {
1871 for (let i = 0; i < input.length; i++) {
1872 const cont = yield* f(input[i])
1873 if (!cont) return false
1874 }
1875 return true
1876 }))
1877
1878/**
1879 * Runs an effectful function for each non-empty input array while it returns
1880 * `true`.
1881 *
1882 * **Details**
1883 *
1884 * The sink stops consuming input when the function returns `false` or when the
1885 * upstream stream ends, and completes with `void`.
1886 *
1887 * @category constructors
1888 * @since 4.0.0
1889 */
1890export const forEachWhileArray = <In, E, R>(
1891 f: (input: NonEmptyReadonlyArray<In>) => Effect.Effect<boolean, E, R>
1892): Sink<void, In, never, E, R> =>
1893 fromTransform((upstream) =>
1894 upstream.pipe(
1895 Effect.flatMap(f),
1896 Effect.flatMap((cont) => cont ? Effect.void : Cause.done()),
1897 Effect.forever({ disableYield: true }),
1898 Pull.catchDone(() => endVoid)
1899 )
1900 )
1901
1902/**
1903 * Creates a sink produced from a scoped effect.
1904 *
1905 * **Example** (Unwrapping a sink effect)
1906 *
1907 * ```ts import.meta.vitest
1908 * import { Effect, Sink, Stream } from "effect"
1909 *
1910 * // Create a sink from an effect that produces a sink
1911 * const processed: Array<number> = []
1912 * const sinkEffect = Effect.succeed(
1913 * Sink.forEach((item: number) => Effect.sync(() => processed.push(item)))
1914 * )
1915 * const sink = Sink.unwrap(sinkEffect)
1916 *
1917 * // Use it with a stream
1918 * const stream = Stream.make(1, 2, 3)
1919 * await Effect.runPromise(Stream.run(stream, sink))
1920 * processed // => [1, 2, 3]
1921 * ```
1922 *
1923 * @category constructors
1924 * @since 2.0.0
1925 */
1926export const unwrap = <A, In, L, E, R, R2>(
1927 effect: Effect.Effect<Sink<A, In, L, E, R2>, E, R>
1928): Sink<A, In, L, E, Exclude<R, Scope.Scope> | R2> => fromChannel(Channel.unwrap(Effect.map(effect, toChannel)))
1929
1930/**
1931 * Runs a summary effect when the sink starts and again when it completes.
1932 *
1933 * @category mapping
1934 * @since 2.0.0
1935 */
1936export const summarized: {
1937 <A2, E2, R2, A3>(
1938 summary: Effect.Effect<A2, E2, R2>,
1939 f: (start: A2, end: A2) => A3
1940 ): <A, In, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<[A, A3], In, L, E2 | E, R2 | R>
1941 <A, In, L, E, R, A2, E2, R2, A3>(
1942 self: Sink<A, In, L, E, R>,
1943 summary: Effect.Effect<A2, E2, R2>,
1944 f: (start: A2, end: A2) => A3
1945 ): Sink<[A, A3], In, L, E | E2, R | R2>
1946} = dual(3, <A, In, L, E, R, A2, E2, R2, A3>(
1947 self: Sink<A, In, L, E, R>,
1948 summary: Effect.Effect<A2, E2, R2>,
1949 f: (start: A2, end: A2) => A3
1950): Sink<[A, A3], In, L, E | E2, R | R2> =>
1951 fromTransform(Effect.fnUntraced(function*(upstream, scope) {
1952 const start = yield* summary
1953 const [done, leftover] = yield* self.transform(upstream, scope)
1954 const end = yield* summary
1955 return [[done, f(start, end)], leftover] as const
1956 })))
1957
1958/**
1959 * Returns the sink that executes this one and times its execution.
1960 *
1961 * @category mapping
1962 * @since 2.0.0
1963 */
1964export const withDuration = <A, In, L, E, R>(
1965 self: Sink<A, In, L, E, R>
1966): Sink<[A, Duration.Duration], In, L, E, R> =>
1967 summarized(self, Clock.monotonicTimeNanos, (start, end) => Duration.nanos(end - start))
1968
1969/**
1970 * A sink that drains all input and returns the elapsed duration.
1971 *
1972 * @category constructors
1973 * @since 2.0.0
1974 */
1975export const timed: Sink<Duration.Duration, unknown> = map(withDuration(drain), ([, duration]) => duration)
1976
1977/**
1978 * Provides a `Context` to this sink.
1979 *
1980 * **Details**
1981 *
1982 * Services contained in the provided context are removed from the sink's
1983 * service requirements.
1984 *
1985 * @category providing services
1986 * @since 2.0.0
1987 */
1988export const provideContext: {
1989 <Provided>(
1990 context: Context.Context<Provided>
1991 ): <A, In, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A, In, L, E, Exclude<R, Provided>>
1992 <A, In, L, E, R, Provided>(
1993 self: Sink<A, In, L, E, R>,
1994 context: Context.Context<Provided>
1995 ): Sink<A, In, L, E, Exclude<R, Provided>>
1996} = dual(2, <A, In, L, E, R, Provided>(
1997 self: Sink<A, In, L, E, R>,
1998 context: Context.Context<Provided>
1999): Sink<A, In, L, E, Exclude<R, Provided>> =>
2000 fromTransform((upstream, scope) =>
2001 self.transform(upstream, scope).pipe(
2002 Effect.provideContext(context)
2003 )
2004 ))
2005
2006/**
2007 * Provides a single service implementation to this sink.
2008 *
2009 * **Details**
2010 *
2011 * The service identified by `key` is removed from the sink's service
2012 * requirements.
2013 *
2014 * @category providing services
2015 * @since 4.0.0
2016 */
2017export const provideService: {
2018 <I, S>(
2019 key: Context.Key<I, S>,
2020 value: Types.NoInfer<S>
2021 ): <A, In, L, E, R>(self: Sink<A, In, L, E, R>) => Sink<A, In, L, E, Exclude<R, I>>
2022 <A, In, L, E, R, I, S>(
2023 self: Sink<A, In, L, E, R>,
2024 key: Context.Key<I, S>,
2025 value: Types.NoInfer<S>
2026 ): Sink<A, In, L, E, Exclude<R, I>>
2027} = dual(3, <A, In, L, E, R, I, S>(
2028 self: Sink<A, In, L, E, R>,
2029 key: Context.Key<I, S>,
2030 value: Types.NoInfer<S>
2031): Sink<A, In, L, E, Exclude<R, I>> =>
2032 fromTransform((upstream, scope) =>
2033 self.transform(upstream, scope).pipe(
2034 Effect.provideService(key, value)
2035 )
2036 ))
2037
2038/**
2039 * Runs a fallback sink if this sink fails with a typed error.
2040 *
2041 * **Details**
2042 *
2043 * The fallback is built from the error and continues consuming from the same
2044 * upstream stream. If the upstream stream had already ended, the fallback sees
2045 * the upstream end instead.
2046 *
2047 * @category error handling
2048 * @since 2.0.0
2049 */
2050export const orElse: {
2051 <E, A2, In2, L2, E2, R2>(
2052 f: (error: Types.NoInfer<E>) => Sink<A2, In2, L2, E2, R2>
2053 ): <A, In, L, R>(self: Sink<A, In, L, E, R>) => Sink<A2 | A, In & In2, L2 | L, E2 | E, R2 | R>
2054 <A, In, L, E, R, A2, In2, L2, E2, R2>(
2055 self: Sink<A, In, L, E, R>,
2056 f: (error: E) => Sink<A2, In2, L2, E2, R2>
2057 ): Sink<A | A2, In & In2, L | L2, E | E2, R | R2>
2058} = dual(2, <A, In, L, E, R, A2, In2, L2, E2, R2>(
2059 self: Sink<A, In, L, E, R>,
2060 f: (error: E) => Sink<A2, In2, L2, E2, R2>
2061): Sink<A | A2, In & In2, L | L2, E | E2, R | R2> =>
2062 fromTransform((upstream, scope) => {
2063 let upstreamDone = false
2064 const pull = Effect.catchCause(upstream, (cause) => {
2065 upstreamDone = true
2066 return Effect.failCause(cause)
2067 })
2068 return Effect.catch(
2069 self.transform(pull, scope) as Effect.Effect<End<A | A2, L | L2>, E, R>,
2070 (error) =>
2071 f(error).transform(
2072 Effect.suspend(() => {
2073 if (upstreamDone) {
2074 return Cause.done()
2075 }
2076 return upstream
2077 }),
2078 scope
2079 )
2080 )
2081 }))
2082
2083/**
2084 * Handles failures from this sink by inspecting the full `Cause`.
2085 *
2086 * **When to use**
2087 *
2088 * Use to recover from a sink failure based on the full `Cause` instead of only
2089 * the typed error value.
2090 *
2091 * **Details**
2092 *
2093 * When this sink fails, the handler effect is run and its success value
2094 * becomes the sink result. If the handler fails, the returned sink fails with
2095 * that error.
2096 *
2097 * @see {@link catch_ catch} for recovering from typed errors only
2098 * @see {@link orElse} for recovering by switching to another sink
2099 *
2100 * @category error handling
2101 * @since 4.0.0
2102 */
2103export const catchCause: {
2104 <E, A2, E2, R2>(
2105 f: (error: Cause.Cause<Types.NoInfer<E>>) => Effect.Effect<A2, E2, R2>
2106 ): <A, In, L, R>(self: Sink<A, In, L, E, R>) => Sink<A2 | A, In, L, E, R2 | R>
2107 <A, In, L, E, R, A2, E2, R2>(
2108 self: Sink<A, In, L, E, R>,
2109 f: (error: Cause.Cause<E>) => Effect.Effect<A2, E2, R2>
2110 ): Sink<A | A2, In, L, E2, R | R2>
2111} = dual(2, <A, In, L, E, R, A2, E2, R2>(
2112 self: Sink<A, In, L, E, R>,
2113 f: (error: Cause.Cause<E>) => Effect.Effect<A2, E2, R2>
2114): Sink<A | A2, In, L, E2, R | R2> =>
2115 transformEffect(
2116 self,
2117 Effect.catchCause((cause) => Effect.map(f(cause), (a2) => [a2 as A | A2] as const))
2118 ))
2119
2120const catch_: {
2121 <E, A2, E2, R2>(
2122 f: (error: Types.NoInfer<E>) => Effect.Effect<A2, E2, R2>
2123 ): <A, In, L, R>(self: Sink<A, In, L, E, R>) => Sink<A2 | A, In, L, E2, R2 | R>
2124 <A, In, L, E, R, A2, E2, R2>(
2125 self: Sink<A, In, L, E, R>,
2126 f: (error: E) => Effect.Effect<A2, E2, R2>
2127 ): Sink<A | A2, In, L, E2, R | R2>
2128} = dual(2, <A, In, L, E, R, A2, E2, R2>(
2129 self: Sink<A, In, L, E, R>,
2130 f: (error: E) => Effect.Effect<A2, E2, R2>
2131): Sink<A | A2, In, L, E2, R | R2> =>
2132 transformEffect(
2133 self,
2134 Effect.catch((error) => Effect.map(f(error), (a2) => [a2 as A | A2] as const))
2135 ))
2136
2137export {
2138 /**
2139 * Handles typed errors from this sink with an effectful fallback value.
2140 *
2141 * **When to use**
2142 *
2143 * Use to recover from a typed sink failure by producing the replacement
2144 * result with an `Effect`.
2145 *
2146 * @see {@link catchCause} for recovering from the full failure cause
2147 * @see {@link orElse} for recovering by switching to another sink
2148 *
2149 * @category error handling
2150 * @since 4.0.0
2151 */
2152 catch_ as catch
2153}
2154
2155/**
2156 * Runs an effect after this sink completes, fails, or is interrupted.
2157 *
2158 * **Details**
2159 *
2160 * The effect receives the sink's `Exit` for the result value. The original
2161 * sink result and leftovers are preserved unless the finalizer itself fails.
2162 *
2163 * @category resource management
2164 * @since 4.0.0
2165 */
2166export const onExit: {
2167 <A, E, X, E2, R2>(
2168 f: (exit: Exit.Exit<A, E>) => Effect.Effect<X, E2, R2>
2169 ): <In, L, R>(self: Sink<A, In, L, E, R>) => Sink<A, In, L, E | E2, R2 | R>
2170 <A, In, L, E, R, X, E2, R2>(
2171 self: Sink<A, In, L, E, R>,
2172 f: (exit: Exit.Exit<A, E>) => Effect.Effect<X, E2, R2>
2173 ): Sink<A, In, L, E | E2, R | R2>
2174} = dual(2, <A, In, L, E, R, X, E2, R2>(
2175 self: Sink<A, In, L, E, R>,
2176 f: (exit: Exit.Exit<A, E>) => Effect.Effect<X, E2, R2>
2177): Sink<A, In, L, E | E2, R | R2> =>
2178 transformEffect(
2179 self,
2180 Effect.onExit((exit) => f(Exit.map(exit, ([a]) => a)))
2181 ))
2182
2183/**
2184 * Runs a finalizer effect after this sink completes, fails, or is interrupted.
2185 *
2186 * **Details**
2187 *
2188 * The original sink result and leftovers are preserved unless the finalizer
2189 * itself fails.
2190 *
2191 * @category resource management
2192 * @since 2.0.0
2193 */
2194export const ensuring: {
2195 <X, E2, R2>(
2196 effect: Effect.Effect<X, E2, R2>
2197 ): <A, E, In, L, R>(self: Sink<A, In, L, E, R>) => Sink<A, In, L, E | E2, R2 | R>
2198 <A, In, L, E, R, X, E2, R2>(
2199 self: Sink<A, In, L, E, R>,
2200 effect: Effect.Effect<X, E2, R2>
2201 ): Sink<A, In, L, E | E2, R | R2>
2202} = dual(2, <A, In, L, E, R, X, E2, R2>(
2203 self: Sink<A, In, L, E, R>,
2204 effect: Effect.Effect<X, E2, R2>
2205): Sink<A, In, L, E | E2, R | R2> => onExit(self, () => effect))