Function next

packages/effect/src/internal/stream.ts:6365–6415

(
    leftover: Option.Option<Chunk.Chunk<A>>,
    delimiterIndex: number
  )

  <A, E, R>(self: Stream.Stream<A, E, R>, delimiter: Chunk.Chunk<A>) => Stream.Stream<Chunk.Chunk<A>, E, R>
>(2, <A, E, R>(self: Stream.Stream<A, E, R>, delimiter: Chunk.Chunk<A>): Stream.Stream<Chunk.Chunk<A>, E, R> => {
  const next = (
    leftover: Option.Option<Chunk.Chunk<A>>,
    delimiterIndex: number
  ): Channel.Channel<Chunk.Chunk<Chunk.Chunk<A>>, Chunk.Chunk<A>, E, E, unknown, unknown, R> =>
    core.readWithCause({
      onInput: (inputChunk: Chunk.Chunk<A>) => {
        let buffer: Array<Chunk.Chunk<A>> | undefined
        const [carry, delimiterCursor] = pipe(
          inputChunk,
          Chunk.reduce(
            [pipe(leftover, Option.getOrElse(() => Chunk.empty<A>())), delimiterIndex] as const,
            ([carry, delimiterCursor], a) => {
              const concatenated = pipe(carry, Chunk.append(a))
              if (
                delimiterCursor < delimiter.length &&
                Equal.equals(a, pipe(delimiter, Chunk.unsafeGet(delimiterCursor)))
              ) {
                if (delimiterCursor + 1 === delimiter.length) {
                  if (buffer === undefined) {
                    buffer = []
                  }
                  buffer.push(pipe(concatenated, Chunk.take(concatenated.length - delimiter.length)))
                  return [Chunk.empty<A>(), 0] as const
                }
                return [concatenated, delimiterCursor + 1] as const
              }
              return [concatenated, Equal.equals(a, pipe(delimiter, Chunk.unsafeGet(0))) ? 1 : 0] as const
            }
          )
        )
        const output = buffer === undefined ? Chunk.empty<Chunk.Chunk<A>>() : Chunk.unsafeFromArray(buffer)
        return core.flatMap(
          core.write(output),
          () => next(Chunk.isNonEmpty(carry) ? Option.some(carry) : Option.none(), delimiterCursor)
        )
      },
      onFailure: (cause) =>
        Option.match(leftover, {
          onNone: () => core.failCause(cause),
          onSome: (chunk) =>
            channel.zipRight(
              core.write(Chunk.of(chunk)),
              core.failCause(cause)
            )
        }),
      onDone: (done) =>
        Option.match(leftover, {
          onNone: () => core.succeed(done),
          onSome: (chunk) => channel.zipRight(core.write(Chunk.of(chunk)), core.succeed(done))
        })
    })
  return new StreamImpl(pipe(toChannel(self), core.pipeTo(next(Option.none(), 0))))
})