Hyperlinkv0.8.0-beta.28

Stream

Stream.slidingSizeconsteffect/Stream.ts:7135
(chunkSize: number, stepSize: number): <A, E, R>(
  self: Stream<A, E, R>
) => Stream<Arr.NonEmptyReadonlyArray<A>, E, R>
<A, E, R>(
  self: Stream<A, E, R>,
  chunkSize: number,
  stepSize: number
): Stream<Arr.NonEmptyReadonlyArray<A>, E, R>

Emits sliding windows of chunkSize elements, advancing by stepSize.

Example (Emitting sliding windows with a step size)

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

const program = Effect.gen(function*() {
  const chunks = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.slidingSize(3, 2),
    Stream.runCollect
  )
  yield* Console.log(chunks)
})

Effect.runPromise(program)
// Output: [ [ 1, 2, 3 ], [ 3, 4, 5 ] ]
grouping
Source effect/Stream.ts:713545 lines
export const slidingSize: {
  (chunkSize: number, stepSize: number): <A, E, R>(self: Stream<A, E, R>) => Stream<Arr.NonEmptyReadonlyArray<A>, E, R>
  <A, E, R>(self: Stream<A, E, R>, chunkSize: number, stepSize: number): Stream<Arr.NonEmptyReadonlyArray<A>, E, R>
} = dual(
  3,
  <A, E, R>(self: Stream<A, E, R>, chunkSize: number, stepSize: number): Stream<Arr.NonEmptyReadonlyArray<A>, E, R> =>
    transformPull(self, (upstream, _scope) =>
      Effect.sync(() => {
        let cause: Cause.Cause<E | Cause.Done> | null = null
        const list = MutableList.make<A>()
        let emitted = false
        const pull: Pull.Pull<
          Arr.NonEmptyReadonlyArray<Arr.NonEmptyReadonlyArray<A>>,
          E | Cause.Done
        > = Effect.matchCauseEffect(upstream, {
          onSuccess(arr) {
            MutableList.appendAllUnsafe(list, arr)
            if (list.length < chunkSize) return pull
            emitted = true
            const chunks = [] as any as Arr.NonEmptyArray<Arr.NonEmptyReadonlyArray<A>>
            while (list.length >= chunkSize) {
              if (chunkSize === stepSize) {
                chunks.push(MutableList.takeN(list, chunkSize) as any)
              } else {
                chunks.push(MutableList.toArrayN(list, chunkSize) as any)
                if (chunkSize === 1) {
                  MutableList.take(list)
                } else {
                  MutableList.takeNVoid(list, stepSize)
                }
              }
            }
            return Effect.succeed(chunks)
          },
          onFailure(cause_) {
            if (emitted) MutableList.takeNVoid(list, chunkSize - stepSize)
            if (list.length === 0) return Effect.failCause(cause_)
            cause = cause_
            return Effect.succeed(Arr.of(MutableList.takeAll(list) as any))
          }
        })

        return Effect.suspend(() => cause ? Effect.failCause(cause) : pull)
      }))
)
Referenced by 1 symbols