Function zipWithNext

packages/effect/src/internal/stream.ts:8611–8648

(
  self: Stream.Stream<A, E, R>
)

/** @internal */
export const zipWithNext = <A, E, R>(
  self: Stream.Stream<A, E, R>
): Stream.Stream<[A, Option.Option<A>], E, R> => {
  const process = (
    last: Option.Option<A>
  ): Channel.Channel<Chunk.Chunk<readonly [A, Option.Option<A>]>, Chunk.Chunk<A>, never, never, void, unknown> =>
    core.readWithCause({
      onInput: (input: Chunk.Chunk<A>) => {
        const [newLast, chunk] = Chunk.mapAccum(
          input,
          last,
          (prev, curr) => [Option.some(curr), pipe(prev, Option.map((a) => [a, curr] as const))] as const
        )
        const output = Chunk.filterMap(
          chunk,
          (option) =>
            Option.isSome(option) ?
              Option.some([option.value[0], Option.some(option.value[1])] as const) :
              Option.none()
        )
        return core.flatMap(
          core.write(output),
          () => process(newLast)
        )
      },
      onFailure: core.failCause,
      onDone: () =>
        Option.match(last, {
          onNone: () => core.void,
          onSome: (value) =>
            channel.zipRight(
              core.write(Chunk.of<readonly [A, Option.Option<A>]>([value, Option.none()])),
              core.void
            )
        })
    })
  return new StreamImpl(pipe(toChannel(self), channel.pipeToOrFail(process(Option.none()))))
}

/** @internal */
export const zipWithPrevious = <A, E, R>(