<A>(options?: { readonly strategy?: QueuingStrategy<A> | undefined }): <
E,
R
>(
self: Stream<A, E, R>
) => Effect.Effect<ReadableStream<A>, never, R>
<A, E, R>(
self: Stream<A, E, R>,
options?: { readonly strategy?: QueuingStrategy<A> | undefined }
): Effect.Effect<ReadableStream<A>, never, R>Creates an Effect that builds a ReadableStream from the stream.
When to use
Use when bridging to Web Streams from inside an Effect so the required
services can be captured from the current context.
Details
See https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream.
Example (Creating a ReadableStream effect)
import { Console, Effect, Stream } from "effect"
const stream = Stream.make(1, 2, 3, 4, 5)
const effect = Effect.gen(function*() {
const readableStream = yield* Stream.toReadableStreamEffect(stream)
yield* Console.log(readableStream instanceof ReadableStream) // true
})
Effect.runPromise(effect)destructors
Source effect/Stream.ts:1126921 lines
export const const toReadableStreamEffect: {
<A>(options?: {
readonly strategy?:
| QueuingStrategy<A>
| undefined
}): <E, R>(
self: Stream<A, E, R>
) => Effect.Effect<ReadableStream<A>, never, R>
<A, E, R>(
self: Stream<A, E, R>,
options?: {
readonly strategy?:
| QueuingStrategy<A>
| undefined
}
): Effect.Effect<ReadableStream<A>, never, R>
}
Creates an Effect that builds a ReadableStream from the stream.
When to use
Use when bridging to Web Streams from inside an Effect so the required
services can be captured from the current context.
Details
See https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream.
Example (Creating a ReadableStream effect)
import { Console, Effect, Stream } from "effect"
const stream = Stream.make(1, 2, 3, 4, 5)
const effect = Effect.gen(function*() {
const readableStream = yield* Stream.toReadableStreamEffect(stream)
yield* Console.log(readableStream instanceof ReadableStream) // true
})
Effect.runPromise(effect)
toReadableStreamEffect: {
<function (type parameter) A in <A>(options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): <E, R>(self: Stream<A, E, R>) => Effect.Effect<ReadableStream<A>, never, R>
options: | {
readonly strategy?:
| QueuingStrategy<A>
| undefined
}
| undefined
strategy?: QueuingStrategy<A> | undefinedstrategy?: interface QueuingStrategy<T = any>QueuingStrategy<function (type parameter) A in <A>(options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): <E, R>(self: Stream<A, E, R>) => Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) E in <E, R>(self: Stream<A, E, R>): Effect.Effect<ReadableStream<A>, never, R>E, function (type parameter) R in <E, R>(self: Stream<A, E, R>): Effect.Effect<ReadableStream<A>, never, 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>(options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): <E, R>(self: Stream<A, E, R>) => Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) E in <E, R>(self: Stream<A, E, R>): Effect.Effect<ReadableStream<A>, never, R>E, function (type parameter) R in <E, R>(self: Stream<A, E, R>): Effect.Effect<ReadableStream<A>, never, R>R>
) => import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface ReadableStream<R = any>The ReadableStream interface of the Streams API represents a readable stream of byte data.
ReadableStream<function (type parameter) A in <A>(options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): <E, R>(self: Stream<A, E, R>) => Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) R in <E, R>(self: Stream<A, E, R>): Effect.Effect<ReadableStream<A>, never, R>R>
<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, 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>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
options: | {
readonly strategy?:
| QueuingStrategy<A>
| undefined
}
| undefined
strategy?: QueuingStrategy<A> | undefinedstrategy?: interface QueuingStrategy<T = any>QueuingStrategy<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface ReadableStream<R = any>The ReadableStream interface of the Streams API represents a readable stream of byte data.
ReadableStream<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
import dualdual(
(args: anyargs) => const isStream: (
u: unknown
) => u is Stream<unknown, unknown, unknown>
Checks whether a value is a Stream.
Example (Checking whether a value is a Stream)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const stream = Stream.make(1, 2, 3)
const notStream = { data: [1, 2, 3] }
yield* Console.log(Stream.isStream(stream))
// true
yield* Console.log(Stream.isStream(notStream))
// false
})
Effect.runPromise(program)
isStream(args: anyargs[0]),
<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, 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>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
options: | {
readonly strategy?:
| QueuingStrategy<A>
| undefined
}
| undefined
strategy?: QueuingStrategy<A> | undefinedstrategy?: interface QueuingStrategy<T = any>QueuingStrategy<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface ReadableStream<R = any>The ReadableStream interface of the Streams API represents a readable stream of byte data.
ReadableStream<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
import EffectEffect.map(
import EffectEffect.context<function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options?: {
readonly strategy?: QueuingStrategy<A> | undefined;
}): Effect.Effect<ReadableStream<A>, never, R>
context: Context.Context<R>(parameter) context: {
mapUnsafe: ReadonlyMap<string, any>;
mutable: boolean;
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; <…;
toString: () => string;
toJSON: () => unknown;
}
const toReadableStreamWith: (<A, XR>(
context: Context.Context<XR>,
options?: {
readonly strategy?:
| QueuingStrategy<A>
| undefined
}
) => <E, R extends XR>(
self: Stream<A, E, R>
) => ReadableStream<A>) &
(<A, E, XR, R extends XR>(
self: Stream<A, E, R>,
context: Context.Context<XR>,
options?: {
readonly strategy?:
| QueuingStrategy<A>
| undefined
}
) => ReadableStream<A>)
Converts the stream to a ReadableStream using the provided services.
When to use
Use when bridging to Web Streams and you already have the Context required
to run the stream outside an Effect.
Details
See https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream.
Example (Converting to a ReadableStream with services)
import { Context, Stream } from "effect"
const stream = Stream.make(1, 2, 3, 4, 5)
const readableStream = Stream.toReadableStreamWith(stream, Context.empty())
toReadableStreamWith(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; <…;
}
context: Context.Context<R>(parameter) context: {
mapUnsafe: ReadonlyMap<string, any>;
mutable: boolean;
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; <…;
toString: () => string;
toJSON: () => unknown;
}
options: | {
readonly strategy?:
| QueuingStrategy<A>
| undefined
}
| undefined