(chunkSize: number, duration: Duration.Input): <A, E, R>(
self: Stream<A, E, R>
) => Stream<Array<A>, E, R>
<A, E, R>(
self: Stream<A, E, R>,
chunkSize: number,
duration: Duration.Input
): Stream<Array<A>, E, R>Partitions the stream into arrays, emitting when the chunk size is reached or the duration passes.
Example (Grouping elements by size or time)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const values = yield* Stream.make(1, 2, 3).pipe(
Stream.groupedWithin(2, "5 seconds"),
Stream.runCollect
)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ [ 1, 2 ], [ 3 ] ]export const const groupedWithin: {
(chunkSize: number, duration: Duration.Input): <
A,
E,
R
>(
self: Stream<A, E, R>
) => Stream<Array<A>, E, R>
<A, E, R>(
self: Stream<A, E, R>,
chunkSize: number,
duration: Duration.Input
): Stream<Array<A>, E, R>
}
Partitions the stream into arrays, emitting when the chunk size is reached
or the duration passes.
Example (Grouping elements by size or time)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const values = yield* Stream.make(1, 2, 3).pipe(
Stream.groupedWithin(2, "5 seconds"),
Stream.runCollect
)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ [ 1, 2 ], [ 3 ] ]
groupedWithin: {
(
chunkSize: numberchunkSize: number,
duration: Duration.Inputduration: import DurationDuration.type Input =
| number
| bigint
| Duration.Duration
| readonly [seconds: number, nanos: number]
| `${number} nano`
| `${number} nanos`
| `${number} micro`
| `${number} micros`
| `${number} milli`
| `${number} millis`
| `${number} second`
| `${number} seconds`
| `${number} minute`
| `${number} minutes`
| `${number} hour`
| `${number} hours`
| `${number} day`
| `${number} days`
| `${number} week`
| `${number} weeks`
| "Infinity"
| "-Infinity"
| Duration.DurationObject
Valid input types that can be converted to a Duration.
When to use
Use when an API should accept any value that Effect can convert into a
Duration, including existing durations, millisecond numbers, nanosecond
bigints, high-resolution tuples, duration strings, infinity strings, or
duration objects.
Details
String inputs accept values like "10 seconds", "500 millis",
"Infinity", and "-Infinity". Finite fractional values that are
normalized to nanoseconds are rounded to the nearest nanosecond, with ties
away from zero.
Input
): <function (type parameter) A in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>R>(self: Stream<A, E, R>(parameter) self: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>R>) => interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<interface Array<T>Array<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>A>, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>R>
<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>R>(self: Stream<A, E, R>(parameter) self: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>R>, chunkSize: numberchunkSize: number, duration: Duration.Inputduration: import DurationDuration.type Input =
| number
| bigint
| Duration.Duration
| readonly [seconds: number, nanos: number]
| `${number} nano`
| `${number} nanos`
| `${number} micro`
| `${number} micros`
| `${number} milli`
| `${number} millis`
| `${number} second`
| `${number} seconds`
| `${number} minute`
| `${number} minutes`
| `${number} hour`
| `${number} hours`
| `${number} day`
| `${number} days`
| `${number} week`
| `${number} weeks`
| "Infinity"
| "-Infinity"
| Duration.DurationObject
Valid input types that can be converted to a Duration.
When to use
Use when an API should accept any value that Effect can convert into a
Duration, including existing durations, millisecond numbers, nanosecond
bigints, high-resolution tuples, duration strings, infinity strings, or
duration objects.
Details
String inputs accept values like "10 seconds", "500 millis",
"Infinity", and "-Infinity". Finite fractional values that are
normalized to nanoseconds are rounded to the nearest nanosecond, with ties
away from zero.
Input): interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<interface Array<T>Array<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>A>, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>R>
} = dual<(...args: Array<any>) => any, <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input) => Stream<Array<A>, E, R>>(arity: 3, body: <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input) => Stream<Array<A>, E, R>): ((...args: Array<any>) => any) & (<A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input) => Stream<Array<A>, E, R>) (+1 overload)Creates a function that can be called in data-first style or data-last
(pipe-friendly) style.
When to use
Use to expose one implementation through both direct and pipe-friendly
call styles.
Details
Pass either the arity of the uncurried function or a predicate that decides
whether the current call is data-first. Arity is the common case. Use a
predicate when optional arguments make arity ambiguous.
Example (Selecting data-first or data-last style by arity)
import { Function, pipe } from "effect"
const sum = Function.dual<
(that: number) => (self: number) => number,
(self: number, that: number) => number
>(2, (self, that) => self + that)
console.log(sum(2, 3)) // 5
console.log(pipe(2, sum(3))) // 5
Example (Defining overloads with call signatures)
import { Function, pipe } from "effect"
const sum: {
(that: number): (self: number) => number
(self: number, that: number): number
} = Function.dual(2, (self: number, that: number): number => self + that)
console.log(sum(2, 3)) // 5
console.log(pipe(2, sum(3))) // 5
Example (Selecting data-first or data-last style with a predicate)
import { Function, pipe } from "effect"
const sum = Function.dual<
(that: number) => (self: number) => number,
(self: number, that: number) => number
>(
(args) => args.length === 2,
(self, that) => self + that
)
console.log(sum(2, 3)) // 5
console.log(pipe(2, sum(3))) // 5
dual(3, <function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>R>(
self: Stream<A, E, R>(parameter) self: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>R>,
chunkSize: numberchunkSize: number,
duration: Duration.Inputduration: import DurationDuration.type Input =
| number
| bigint
| Duration.Duration
| readonly [seconds: number, nanos: number]
| `${number} nano`
| `${number} nanos`
| `${number} micro`
| `${number} micros`
| `${number} milli`
| `${number} millis`
| `${number} second`
| `${number} seconds`
| `${number} minute`
| `${number} minutes`
| `${number} hour`
| `${number} hours`
| `${number} day`
| `${number} days`
| `${number} week`
| `${number} weeks`
| "Infinity"
| "-Infinity"
| Duration.DurationObject
Valid input types that can be converted to a Duration.
When to use
Use when an API should accept any value that Effect can convert into a
Duration, including existing durations, millisecond numbers, nanosecond
bigints, high-resolution tuples, duration strings, infinity strings, or
duration objects.
Details
String inputs accept values like "10 seconds", "500 millis",
"Infinity", and "-Infinity". Finite fractional values that are
normalized to nanoseconds are rounded to the nearest nanosecond, with ties
away from zero.
Input
): interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<interface Array<T>Array<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>A>, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Duration.Input): Stream<Array<A>, E, R>R> =>
const aggregateWithin: {
<B, A, A2, E2, R2, C, E3, R3>(
sink: Sink.Sink<B, A | A2, A2, E2, R2>,
schedule: Schedule.Schedule<
C,
Option.Option<B>,
E3,
R3
>
): <E, R>(
self: Stream<A, E, R>
) => Stream<B, E2 | E | E3, R2 | R3 | R>
<A, E, R, B, A2, E2, R2, C, E3, R3>(
self: Stream<A, E, R>,
sink: Sink.Sink<B, A | A2, A2, E2, R2>,
schedule: Schedule.Schedule<
C,
Option.Option<B>,
E3,
R3
>
): Stream<B, E | E2 | E3, R | R2 | R3>
}
self: Stream<A, E, R>(parameter) self: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
import SinkSink.const take: <In>(
n: number
) => Sink<Array<In>, In, In>
Collects up to n input elements into an array.
Details
If n is less than or equal to zero, the sink completes with an empty array.
If more elements are pulled than needed, the remaining elements from the same
array are returned as leftovers.
take(chunkSize: numberchunkSize),
import ScheduleSchedule.const spaced: (
duration: Duration.Input
) => Schedule<number>
Returns a schedule that recurs continuously, each repetition spaced the
specified duration from the last run.
When to use
Use when each delay should start after the previous action completes.
Example (Repeating with fixed spacing)
import { Console, Effect, Schedule } from "effect"
// Basic spaced schedule - runs every 2 seconds
const everyTwoSeconds = Schedule.spaced("2 seconds")
// Heartbeat that runs indefinitely with fixed spacing
const heartbeat = Effect.gen(function*() {
yield* Console.log("Heartbeat")
}).pipe(
Effect.repeat(everyTwoSeconds)
)
// Limited repeat - run only 5 times with 1-second spacing
const limitedTask = Effect.gen(function*() {
yield* Console.log("Executing scheduled task...")
yield* Effect.sleep("500 millis") // simulate work
return "Task completed"
}).pipe(
Effect.repeat(
Schedule.spaced("1 second").pipe(Schedule.upTo({ times: 5 }))
)
)
// Simple spaced schedule with limited repetitions
const limitedSpaced = Schedule.max([
Schedule.spaced("100 millis"),
Schedule.recurs(5) // at most 5 times
])
const program = Effect.gen(function*() {
yield* Console.log("Starting spaced execution...")
yield* Effect.repeat(
Effect.succeed("work item"),
limitedSpaced
)
yield* Console.log("Completed executions")
})
spaced(duration: Duration.Inputduration)
))