Hyperlinkv0.8.0-beta.28

Stream

Stream.transduceconsteffect/Stream.ts:8531
<A2, A, E2, R2>(sink: Sink.Sink<A2, A, A, E2, R2>): <E, R>(
  self: Stream<A, E, R>
) => Stream<A2, E2 | E, R2 | R>
<A, E, R, A2, E2, R2>(
  self: Stream<A, E, R>,
  sink: Sink.Sink<A2, A, A, E2, R2>
): Stream<A2, E2 | E, R2 | R>

Applies a sink transducer to the stream and emits each sink result.

Example (Transducing with a sink)

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

const program = Effect.gen(function* () {
  const result = yield* Stream.make(1, 2, 3, 4).pipe(
    Stream.transduce(Sink.take(2)),
    Stream.runCollect
  )

  yield* Console.log(result)
  // Output: [ [ 1, 2 ], [ 3, 4 ] ]
})
Aggregation
Source effect/Stream.ts:853146 lines
export const transduce = dual<
  <A2, A, E2, R2>(
    sink: Sink.Sink<A2, A, A, E2, R2>
  ) => <E, R>(self: Stream<A, E, R>) => Stream<A2, E2 | E, R2 | R>,
  <A, E, R, A2, E2, R2>(
    self: Stream<A, E, R>,
    sink: Sink.Sink<A2, A, A, E2, R2>
  ) => Stream<A2, E2 | E, R2 | R>
>(
  2,
  <A, E, R, A2, E2, R2>(
    self: Stream<A, E, R>,
    sink: Sink.Sink<A2, A, A, E2, R2>
  ): Stream<A2, E2 | E, R2 | R> =>
    transformPull(self, (upstream, scope) =>
      Effect.sync(() => {
        let done: Exit.Exit<never, Cause.Done<void> | E> | undefined
        let leftover: Arr.NonEmptyReadonlyArray<A> | undefined
        const upstreamWithLeftover = Effect.suspend(() => {
          if (leftover !== undefined) {
            const chunk = leftover
            leftover = undefined
            return Effect.succeed(chunk)
          }
          return upstream
        }).pipe(
          Effect.catch((error) => {
            done = Exit.fail(error)
            return Cause.done()
          })
        )
        const pull = Effect.map(
          Effect.suspend(() => sink.transform(upstreamWithLeftover, scope)),
          ([value, leftover_]) => {
            leftover = leftover_
            return Arr.of(value)
          }
        )
        return Effect.suspend((): Pull.Pull<
          Arr.NonEmptyReadonlyArray<A2>,
          E | E2,
          void,
          R2
        > => done ? done : pull)
      }))
)