Stream.transformPullBracket — API

Stream.transformPullBracketconsteffect/Stream.ts:693

<A, E, R, B, E2, R2, EX, RX>(
  self: Stream<A, E, R>,
  f: (
    pull: Pull.Pull<Arr.NonEmptyReadonlyArray<A>, E, void, R>,
    scope: Scope.Scope,
    forkedScope: Scope.Scope
  ) => Effect.Effect<
    Pull.Pull<Arr.NonEmptyReadonlyArray<B>, E2, void, R2>,
    EX,
    RX
  >
): Stream<B, EX | Pull.ExcludeDone<E2>, R | R2 | RX>

Transforms a stream by effectfully transforming its pull effect.

Details

A forked scope is also provided to the transformation function, which is closed once the resulting stream has finished processing.

Example (Transforming a stream by effectfully transforming its pull effect)

import { Console, Effect, Scope, Stream } from "effect"

const stream = Stream.make(1, 2, 3)

const transformed = Stream.transformPullBracket(
  stream,
  (pull, _scope, forkedScope) =>
    Effect.gen(function*() {
      yield* Scope.addFinalizer(forkedScope, Console.log("Releasing scope"))
      return pull
    })
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(transformed)
  yield* Console.log(values)
})

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

constructors

Source effect/Stream.ts:69317 lines

export const transformPullBracket = <A, E, R, B, E2, R2, EX, RX>(
  self: Stream<A, E, R>,
  f: (
    pull: Pull.Pull<Arr.NonEmptyReadonlyArray<A>, E, void, R>,
    scope: Scope.Scope,
    forkedScope: Scope.Scope
  ) => Effect.Effect<
    Pull.Pull<Arr.NonEmptyReadonlyArray<B>, E2, void, R2>,
    EX,
    RX
  >
): Stream<B, EX | Pull.ExcludeDone<E2>, R | R2 | RX> =>
  fromChannel(
    Channel.fromTransformBracket((_, scope, forkedScope) =>
      Effect.flatMap(Channel.toPullScoped(self.channel, scope), (pull) => f(pull, scope, forkedScope))
    )
  )

Referenced by 1 symbols

Stream.tapSink