message: perf(streams): extract loop in chunkN
diff: @@ -15,33 +15,25 @@ class State<X> {
  constructor(public buffer: A.Chunk<X>, public done: boolean) {}
  }
  -/**
  - * Re-chunks the elements of the stream into chunks of
  - * `n` elements each.
  - * The last chunk might contain less than `n` elements
  - */
  -export function chunkN_<R, E, O>(self: Stream<R, E, O>, n: number): Stream<R, E, O> {
  - const emitOrAccumulate = (
  +function emitOrAccumulate<R, E, O>(
  buffer: A.Chunk<O>,
  done: boolean,
  ref: Ref.Ref<State<O>>,
  - pull: T.Effect<R, O.Option<E>, A.Chunk<O>>
  - ): T.Effect<R, O.Option<E>, A.Chunk<O>> => {
  + pull: T.Effect<R, O.Option<E>, A.Chunk<O>>,
  + n: number
  +): T.Effect<R, O.Option<E>, A.Chunk<O>> {
  if (A.size(buffer) < n) {
  if (done) {
  if (A.isEmpty(buffer)) {
  return Pull.end
  } else {
  - return T.zipRight_(
  - ref.set(new State(A.empty(), true)),
  - Pull.emitChunk(buffer)
  - )
  + return T.zipRight_(ref.set(new State(A.empty(), true)), Pull.emitChunk(buffer))
  }
  } else {
  return T.foldM_(
  pull,
  - O.fold(() => emitOrAccumulate(buffer, true, ref, pull), Pull.fail),
  - (ch) => emitOrAccumulate(A.concat_(buffer, ch), false, ref, pull)
  + O.fold(() => emitOrAccumulate(buffer, true, ref, pull, n), Pull.fail),
  + (ch) => emitOrAccumulate(A.concat_(buffer, ch), false, ref, pull, n)
  )
  }
  } else {
  @@ -51,6 +43,12 @@ export function chunkN_<R, E, O>(self: Stream<R, E, O>, n: number): Stream<R, E,
  }
  }
  +/**
  + * Re-chunks the elements of the stream into chunks of
  + * `n` elements each.
  + * The last chunk might contain less than `n` elements
  + */
  +export function chunkN_<R, E, O>(self: Stream<R, E, O>, n: number): Stream<R, E, O> {
  if (n < 1) {
  return halt(C.die(new C.IllegalArgumentException("chunkN: n must be at least 1")))
  } else {
  @@ -62,7 +60,7 @@ export function chunkN_<R, E, O>(self: Stream<R, E, O>, n: number): Stream<R, E,
  ),
  M.bind("p", () => self.proc),
  M.let("pull", ({ p, ref }) =>
  - T.chain_(ref.get, (s) => emitOrAccumulate(s.buffer, s.done, ref, p))
  + T.chain_(ref.get, (s) => emitOrAccumulate(s.buffer, s.done, ref, p, n))
  ),
  M.map(({ pull }) => pull)
  )
  