Sink Concurrency
This section covers concurrent operations that allow multiple sinks to run simultaneously. These can be valuable for enhancing task performance when concurrent execution is desired.
To run two sinks concurrently and combine their results, use Sink.zip. This operation executes both sinks concurrently and combines their outcomes into a tuple.
Example (Running Two Sinks Concurrently and Combining Results)
1import { import SinkSink, import ConsoleConsole, import StreamStream, import ScheduleSchedule, import EffectEffect } from "effect"2
3const const stream: Stream.Stream<string, never, never>stream = import StreamStream.const make: <[string, string, string, string, string]>(as_0: string, as_1: string, as_2: string, as_3: string, as_4: string) => Stream.Stream<string, never, never>Creates a stream from an sequence of values.
make("1", "2", "3", "4", "5").(method) Pipeable.pipe<Stream.Stream<string, never, never>, Stream.Stream<string, never, never>>(this: Stream.Stream<...>, ab: (_: Stream.Stream<string, never, never>) => Stream.Stream<string, never, never>): Stream.Stream<...> (+21 overloads)pipe(4 import StreamStream.const schedule: <number, string, never, string>(schedule: Schedule.Schedule<number, string, never>) => <E, R>(self: Stream.Stream<string, E, R>) => Stream.Stream<string, E, R> (+1 overload)Schedules the output of the stream using the provided `schedule`.
schedule(import ScheduleSchedule.const spaced: (duration: DurationInput) => Schedule.Schedule<number>Returns a schedule that recurs continuously, each repetition spaced the
specified duration from the last run.
spaced("10 millis"))5)6
7const const sink1: Sink.Sink<number, string, never, never, never>sink1 = import SinkSink.const forEach: <string, void, never, never>(f: (input: string) => Effect.Effect<void, never, never>) => Sink.Sink<void, string, never, never, never>A sink that executes the provided effectful function for every element fed
to it.
forEach(((parameter) s: strings: string) =>8 import ConsoleConsole.const log: (...args: ReadonlyArray<any>) => Effect.Effect<void>log(`sink 1: ${(parameter) s: strings}`)9).(method) Pipeable.pipe<Sink.Sink<void, string, never, never, never>, Sink.Sink<number, string, never, never, never>>(this: Sink.Sink<...>, ab: (_: Sink.Sink<void, string, never, never, never>) => Sink.Sink<...>): Sink.Sink<...> (+21 overloads)pipe(import SinkSink.const as: <number>(a: number) => <A, In, L, E, R>(self: Sink.Sink<A, In, L, E, R>) => Sink.Sink<number, In, L, E, R> (+1 overload)Replaces this sink's result with the provided value.
as(1))10
11const const sink2: Sink.Sink<number, string, never, never, never>sink2 = import SinkSink.const forEach: <string, void, never, never>(f: (input: string) => Effect.Effect<void, never, never>) => Sink.Sink<void, string, never, never, never>A sink that executes the provided effectful function for every element fed
to it.
forEach(((parameter) s: strings: string) =>12 import ConsoleConsole.const log: (...args: ReadonlyArray<any>) => Effect.Effect<void>log(`sink 2: ${(parameter) s: strings}`)13).(method) Pipeable.pipe<Sink.Sink<void, string, never, never, never>, Sink.Sink<number, string, never, never, never>>(this: Sink.Sink<...>, ab: (_: Sink.Sink<void, string, never, never, never>) => Sink.Sink<...>): Sink.Sink<...> (+21 overloads)pipe(import SinkSink.const as: <number>(a: number) => <A, In, L, E, R>(self: Sink.Sink<A, In, L, E, R>) => Sink.Sink<number, In, L, E, R> (+1 overload)Replaces this sink's result with the provided value.
as(2))14
15// Combine the two sinks to run concurrently and collect results in a tuple16const const sink: Sink.Sink<[number, number], string, never, never, never>sink = import SinkSink.const zip: <number, string, never, never, never, number, string, never, never, never>(self: Sink.Sink<number, string, never, never, never>, that: Sink.Sink<number, string, never, never, never>, options?: {
readonly concurrent?: boolean | undefined;
} | undefined) => Sink.Sink<...> (+1 overload)Feeds inputs to this sink until it yields a result, then switches over to
the provided sink until it yields a result, finally combining the two
results into a tuple.
zip(const sink1: Sink.Sink<number, string, never, never, never>sink1, const sink2: Sink.Sink<number, string, never, never, never>sink2, { (property) concurrent?: boolean | undefinedconcurrent: true })17
18import EffectEffect.const runPromise: <[number, number], never>(effect: Effect.Effect<[number, number], never, never>, options?: {
readonly signal?: AbortSignal;
} | undefined) => Promise<[number, number]>Executes an effect and returns a `Promise` that resolves with the result.
Use `runPromise` when working with asynchronous effects and you need to integrate with code that uses Promises.
If the effect fails, the returned Promise will be rejected with the error.
runPromise(import StreamStream.const run: <string, never, never, [number, number], never, never>(self: Stream.Stream<string, never, never>, sink: Sink.Sink<[number, number], string, unknown, never, never>) => Effect.Effect<[...], never, never> (+1 overload)Runs the sink on the stream to produce either the sink's result or an error.
run(const stream: Stream.Stream<string, never, never>stream, const sink: Sink.Sink<[number, number], string, never, never, never>sink)).(method) Promise<[number, number]>.then<void, never>(onfulfilled?: ((value: [number, number]) => void | PromiseLike<void>) | null | undefined, onrejected?: ((reason: any) => PromiseLike<never>) | null | undefined): Promise<...>Attaches callbacks for the resolution and/or rejection of the Promise.
then(namespace console
var console: ConsoleThe `console` module provides a simple debugging console that is similar to the
JavaScript console mechanism provided by web browsers.
The module exports two specific components:
* A `Console` class with methods such as `console.log()`, `console.error()` and `console.warn()` that can be used to write to any Node.js stream.
* A global `console` instance configured to write to [`process.stdout`](https://nodejs.org/docs/latest-v22.x/api/process.html#processstdout) and
[`process.stderr`](https://nodejs.org/docs/latest-v22.x/api/process.html#processstderr). The global `console` can be used without importing the `node:console` module.
_**Warning**_: The global console object's methods are neither consistently
synchronous like the browser APIs they resemble, nor are they consistently
asynchronous like all other Node.js streams. See the [`note on process I/O`](https://nodejs.org/docs/latest-v22.x/api/process.html#a-note-on-process-io) for
more information.
Example using the global `console`:
```js
console.log('hello world');
// Prints: hello world, to stdout
console.log('hello %s', 'world');
// Prints: hello world, to stdout
console.error(new Error('Whoops, something bad happened'));
// Prints error message and stack trace to stderr:
// Error: Whoops, something bad happened
// at [eval]:5:15
// at Script.runInThisContext (node:vm:132:18)
// at Object.runInThisContext (node:vm:309:38)
// at node:internal/process/execution:77:19
// at [eval]-wrapper:6:22
// at evalScript (node:internal/process/execution:76:60)
// at node:internal/main/eval_string:23:3
const name = 'Will Robinson';
console.warn(`Danger ${name}! Danger!`);
// Prints: Danger Will Robinson! Danger!, to stderr
```
Example using the `Console` class:
```js
const out = getStreamSomehow();
const err = getStreamSomehow();
const myConsole = new console.Console(out, err);
myConsole.log('hello world');
// Prints: hello world, to out
myConsole.log('hello %s', 'world');
// Prints: hello world, to out
myConsole.error(new Error('Whoops, something bad happened'));
// Prints: [Error: Whoops, something bad happened], to err
const name = 'Will Robinson';
myConsole.warn(`Danger ${name}! Danger!`);
// Prints: Danger Will Robinson! Danger!, to err
```
console.(method) globalThis.Console.log(message?: any, ...optionalParams: any[]): voidPrints to `stdout` with newline. Multiple arguments can be passed, with the
first used as the primary message and all additional used as substitution
values similar to [`printf(3)`](http://man7.org/linux/man-pages/man3/printf.3.html)
(the arguments are all passed to [`util.format()`](https://nodejs.org/docs/latest-v22.x/api/util.html#utilformatformat-args)).
```js
const count = 5;
console.log('count: %d', count);
// Prints: count: 5, to stdout
console.log('count:', count);
// Prints: count: 5, to stdout
```
See [`util.format()`](https://nodejs.org/docs/latest-v22.x/api/util.html#utilformatformat-args) for more information.
log)19/*20Output:21sink 1: 122sink 2: 123sink 1: 224sink 2: 225sink 1: 326sink 2: 327sink 1: 428sink 2: 429sink 1: 530sink 2: 531[ 1, 2 ]32*/The Sink.race operation allows multiple sinks to compete for completion. The first sink to finish provides the result.
Example (Racing Two Sinks to Capture the First Result)
1import { import SinkSink, import ConsoleConsole, import StreamStream, import ScheduleSchedule, import EffectEffect } from "effect"2
3const const stream: Stream.Stream<string, never, never>stream = import StreamStream.const make: <[string, string, string, string, string]>(as_0: string, as_1: string, as_2: string, as_3: string, as_4: string) => Stream.Stream<string, never, never>Creates a stream from an sequence of values.
make("1", "2", "3", "4", "5").(method) Pipeable.pipe<Stream.Stream<string, never, never>, Stream.Stream<string, never, never>>(this: Stream.Stream<...>, ab: (_: Stream.Stream<string, never, never>) => Stream.Stream<string, never, never>): Stream.Stream<...> (+21 overloads)pipe(4 import StreamStream.const schedule: <number, string, never, string>(schedule: Schedule.Schedule<number, string, never>) => <E, R>(self: Stream.Stream<string, E, R>) => Stream.Stream<string, E, R> (+1 overload)Schedules the output of the stream using the provided `schedule`.
schedule(import ScheduleSchedule.const spaced: (duration: DurationInput) => Schedule.Schedule<number>Returns a schedule that recurs continuously, each repetition spaced the
specified duration from the last run.
spaced("10 millis"))5)6
7const const sink1: Sink.Sink<number, string, never, never, never>sink1 = import SinkSink.const forEach: <string, void, never, never>(f: (input: string) => Effect.Effect<void, never, never>) => Sink.Sink<void, string, never, never, never>A sink that executes the provided effectful function for every element fed
to it.
forEach(((parameter) s: strings: string) =>8 import ConsoleConsole.const log: (...args: ReadonlyArray<any>) => Effect.Effect<void>log(`sink 1: ${(parameter) s: strings}`)9).(method) Pipeable.pipe<Sink.Sink<void, string, never, never, never>, Sink.Sink<number, string, never, never, never>>(this: Sink.Sink<...>, ab: (_: Sink.Sink<void, string, never, never, never>) => Sink.Sink<...>): Sink.Sink<...> (+21 overloads)pipe(import SinkSink.const as: <number>(a: number) => <A, In, L, E, R>(self: Sink.Sink<A, In, L, E, R>) => Sink.Sink<number, In, L, E, R> (+1 overload)Replaces this sink's result with the provided value.
as(1))10
11const const sink2: Sink.Sink<number, string, never, never, never>sink2 = import SinkSink.const forEach: <string, void, never, never>(f: (input: string) => Effect.Effect<void, never, never>) => Sink.Sink<void, string, never, never, never>A sink that executes the provided effectful function for every element fed
to it.
forEach(((parameter) s: strings: string) =>12 import ConsoleConsole.const log: (...args: ReadonlyArray<any>) => Effect.Effect<void>log(`sink 2: ${(parameter) s: strings}`)13).(method) Pipeable.pipe<Sink.Sink<void, string, never, never, never>, Sink.Sink<number, string, never, never, never>>(this: Sink.Sink<...>, ab: (_: Sink.Sink<void, string, never, never, never>) => Sink.Sink<...>): Sink.Sink<...> (+21 overloads)pipe(import SinkSink.const as: <number>(a: number) => <A, In, L, E, R>(self: Sink.Sink<A, In, L, E, R>) => Sink.Sink<number, In, L, E, R> (+1 overload)Replaces this sink's result with the provided value.
as(2))14
15// Race the two sinks, the result will be from the first to complete16const const sink: Sink.Sink<number, string, never, never, never>sink = import SinkSink.const race: <number, string, never, never, never, number, string, never, never, never>(self: Sink.Sink<number, string, never, never, never>, that: Sink.Sink<number, string, never, never, never>) => Sink.Sink<...> (+1 overload)Runs both sinks in parallel on the input, , returning the result or the
error from the one that finishes first.
race(const sink1: Sink.Sink<number, string, never, never, never>sink1, const sink2: Sink.Sink<number, string, never, never, never>sink2)17
18import EffectEffect.const runPromise: <number, never>(effect: Effect.Effect<number, never, never>, options?: {
readonly signal?: AbortSignal;
} | undefined) => Promise<number>Executes an effect and returns a `Promise` that resolves with the result.
Use `runPromise` when working with asynchronous effects and you need to integrate with code that uses Promises.
If the effect fails, the returned Promise will be rejected with the error.
runPromise(import StreamStream.const run: <string, never, never, number, never, never>(self: Stream.Stream<string, never, never>, sink: Sink.Sink<number, string, unknown, never, never>) => Effect.Effect<number, never, never> (+1 overload)Runs the sink on the stream to produce either the sink's result or an error.
run(const stream: Stream.Stream<string, never, never>stream, const sink: Sink.Sink<number, string, never, never, never>sink)).(method) Promise<number>.then<void, never>(onfulfilled?: ((value: number) => void | PromiseLike<void>) | null | undefined, onrejected?: ((reason: any) => PromiseLike<never>) | null | undefined): Promise<...>Attaches callbacks for the resolution and/or rejection of the Promise.
then(namespace console
var console: ConsoleThe `console` module provides a simple debugging console that is similar to the
JavaScript console mechanism provided by web browsers.
The module exports two specific components:
* A `Console` class with methods such as `console.log()`, `console.error()` and `console.warn()` that can be used to write to any Node.js stream.
* A global `console` instance configured to write to [`process.stdout`](https://nodejs.org/docs/latest-v22.x/api/process.html#processstdout) and
[`process.stderr`](https://nodejs.org/docs/latest-v22.x/api/process.html#processstderr). The global `console` can be used without importing the `node:console` module.
_**Warning**_: The global console object's methods are neither consistently
synchronous like the browser APIs they resemble, nor are they consistently
asynchronous like all other Node.js streams. See the [`note on process I/O`](https://nodejs.org/docs/latest-v22.x/api/process.html#a-note-on-process-io) for
more information.
Example using the global `console`:
```js
console.log('hello world');
// Prints: hello world, to stdout
console.log('hello %s', 'world');
// Prints: hello world, to stdout
console.error(new Error('Whoops, something bad happened'));
// Prints error message and stack trace to stderr:
// Error: Whoops, something bad happened
// at [eval]:5:15
// at Script.runInThisContext (node:vm:132:18)
// at Object.runInThisContext (node:vm:309:38)
// at node:internal/process/execution:77:19
// at [eval]-wrapper:6:22
// at evalScript (node:internal/process/execution:76:60)
// at node:internal/main/eval_string:23:3
const name = 'Will Robinson';
console.warn(`Danger ${name}! Danger!`);
// Prints: Danger Will Robinson! Danger!, to stderr
```
Example using the `Console` class:
```js
const out = getStreamSomehow();
const err = getStreamSomehow();
const myConsole = new console.Console(out, err);
myConsole.log('hello world');
// Prints: hello world, to out
myConsole.log('hello %s', 'world');
// Prints: hello world, to out
myConsole.error(new Error('Whoops, something bad happened'));
// Prints: [Error: Whoops, something bad happened], to err
const name = 'Will Robinson';
myConsole.warn(`Danger ${name}! Danger!`);
// Prints: Danger Will Robinson! Danger!, to err
```
console.(method) globalThis.Console.log(message?: any, ...optionalParams: any[]): voidPrints to `stdout` with newline. Multiple arguments can be passed, with the
first used as the primary message and all additional used as substitution
values similar to [`printf(3)`](http://man7.org/linux/man-pages/man3/printf.3.html)
(the arguments are all passed to [`util.format()`](https://nodejs.org/docs/latest-v22.x/api/util.html#utilformatformat-args)).
```js
const count = 5;
console.log('count: %d', count);
// Prints: count: 5, to stdout
console.log('count:', count);
// Prints: count: 5, to stdout
```
See [`util.format()`](https://nodejs.org/docs/latest-v22.x/api/util.html#utilformatformat-args) for more information.
log)19/*20Output:21sink 1: 122sink 2: 123sink 1: 224sink 2: 225sink 1: 326sink 2: 327sink 1: 428sink 2: 429sink 1: 530sink 2: 531132*/