Scheduling Streams

schedule

When working with streams, you might need to introduce specific time intervals between each emission of stream elements. This can be achieved using the Stream.schedule combinator.
1
import { import Stream
Stream, import Schedule
Schedule, import Console
Console, import Effect
Effect } from "effect"
2

3
const const stream: Stream.Stream<number, never, never>
stream = import Stream
Stream.const make: <[number, number, number, number, number]>(as_0: number, as_1: number, as_2: number, as_3: number, as_4: number) => Stream.Stream<number, never, never>Creates a stream from an sequence of values.
make(1, 2, 3, 4, 5).(method) Pipeable.pipe<Stream.Stream<number, never, never>, Stream.Stream<number, never, never>, Stream.Stream<number, never, never>>(this: Stream.Stream<...>, ab: (_: Stream.Stream<number, never, never>) => Stream.Stream<...>, bc: (_: Stream.Stream<...>) => Stream.Stream<...>): Stream.Stream<...> (+21 overloads)
pipe(
4
  import Stream
Stream.const schedule: <number, number, never, number>(schedule: Schedule.Schedule<number, number, never>) => <E, R>(self: Stream.Stream<number, E, R>) => Stream.Stream<number, E, R> (+1 overload)Schedules the output of the stream using the provided `schedule`.
schedule(import Schedule
Schedule.const spaced: (duration: DurationInput) => Schedule.Schedule<number>Returns a schedule that recurs continuously, each repetition spaced the
specified duration from the last run.
spaced("1 second")),
5
  import Stream
Stream.const tap: <number, void, never, never>(f: (a: number) => Effect.Effect<void, never, never>) => <E, R>(self: Stream.Stream<number, E, R>) => Stream.Stream<number, E, R> (+1 overload)Adds an effect to consumption of every element of the stream.
tap(import Console
Console.const log: (...args: ReadonlyArray<any>) => Effect.Effect<void>
log)
6
)
7

8
import Effect
Effect.const runPromise: <Chunk<number>, never>(effect: Effect.Effect<Chunk<number>, never, never>, options?: {
    readonly signal?: AbortSignal;
} | undefined) => Promise<Chunk<number>>Runs an `Effect` workflow, returning a `Promise` which resolves with the
result of the workflow or rejects with an error.
runPromise(import Stream
Stream.const runCollect: <number, never, never>(self: Stream.Stream<number, never, never>) => Effect.Effect<Chunk<number>, never, never>Runs the stream and collects all of its elements to a chunk.
runCollect(const stream: Stream.Stream<number, never, never>
stream)).(method) Promise<Chunk<number>>.then<void, never>(onfulfilled?: ((value: Chunk<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)
9
/*
10
Output:
11
1
12
2
13
3
14
4
15
5
16
{
17
  _id: "Chunk",
18
  values: [ 1, 2, 3, 4, 5 ]
19
}
20
*/
In this example, we’ve used the Schedule.spaced("1 second") schedule to introduce a one-second gap between each emission in the stream.