effect-io-ai

Package: effect
Module: Stream

Stream.withExecutionPlan

Applies an ExecutionPlan to a stream, retrying with step-provided resources until it succeeds or the plan is exhausted.

Details

By default, a failing step can fallback even after emitting elements; set preventFallbackOnPartialStream to fail instead of mixing partial output with a later fallback.

Example (Applying an execution plan)

import { Console, Context, Effect, ExecutionPlan, Layer, Stream } from "effect"

class Service extends Context.Service<Service>()("Service", {
  make: Effect.succeed({
    stream: Stream.fail("A") as Stream.Stream<number, string>
  })
}) {
  static Bad = Layer.succeed(Service, Service.of({ stream: Stream.fail("A") }))
  static Good = Layer.succeed(Service, Service.of({ stream: Stream.make(1, 2, 3) }))
}

const plan = ExecutionPlan.make(
  { provide: Service.Bad },
  { provide: Service.Good }
)

const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))

const program = Effect.gen(function*() {
  const items = yield* stream.pipe(Stream.withExecutionPlan(plan), Stream.runCollect)
  yield* Console.log(items)
})

Effect.runPromise(program)
// Output: [ 1, 2, 3 ]

Signature

declare const withExecutionPlan: { <Input, R2, Provides, PolicyE>(policy: ExecutionPlan.ExecutionPlan<{ provides: Provides; input: Input; error: PolicyE; requirements: R2; }>, options?: { readonly preventFallbackOnPartialStream?: boolean | undefined; }): <A, E extends Input, R>(self: Stream<A, E, R>) => Stream<A, E | PolicyE, R2 | Exclude<R, Provides>>; <A, E extends Input, R, R2, Input, Provides, PolicyE>(self: Stream<A, E, R>, policy: ExecutionPlan.ExecutionPlan<{ provides: Provides; input: Input; error: PolicyE; requirements: R2; }>, options?: { readonly preventFallbackOnPartialStream?: boolean | undefined; }): Stream<A, E | PolicyE, R2 | Exclude<R, Provides>>; }

Source

Since v3.16.0