Function chunkCoordination

packages/effect/test/utils/coordination.ts:16–60

(
  _chunks: Iterable<Chunk.Chunk<A>>
)

}

export const chunkCoordination = <A>(
  _chunks: Iterable<Chunk.Chunk<A>>
): Effect.Effect<ChunkCoordination<A>> =>
  Effect.gen(function*() {
    const chunks = Chunk.fromIterable(_chunks)
    const queue = yield* Queue.unbounded<Exit.Exit<Chunk.Chunk<A>, Option.Option<never>>>()
    const ps = yield* Queue.unbounded<void>()
    const ref = yield* Ref.make<Chunk.Chunk<Chunk.Chunk<Exit.Exit<Chunk.Chunk<A>, Option.Option<never>>>>>(
      pipe(
        chunks,
        Chunk.dropRight(1),
        Chunk.map((chunk) => Chunk.of(Exit.succeed(chunk))),
        Chunk.appendAll(
          pipe(
            Chunk.last(chunks),
            Option.map((chunk) =>
              Chunk.unsafeFromArray<Exit.Exit<Chunk.Chunk<A>, Option.Option<never>>>([
                Exit.succeed(chunk),
                Exit.fail(Option.none())
              ])
            ),
            Option.match({
              onNone: () => Chunk.empty<Chunk.Chunk<Exit.Exit<Chunk.Chunk<A>, Option.Option<never>>>>(),
              onSome: Chunk.of
            })
          )
        )
      )
    )
    return {
      queue,
      offer: pipe(
        Ref.modify(ref, (chunk) => {
          if (Chunk.isEmpty(chunk)) {
            return [Chunk.empty(), Chunk.empty()]
          }
          return [Chunk.unsafeHead(chunk), Chunk.drop(1)(chunk)]
        }),
        Effect.flatMap((chunks) => Queue.offerAll(queue, chunks)),
        Effect.asVoid
      ),
      proceed: pipe(Queue.offer(ps, void 0), Effect.asVoid),
      awaitNext: Queue.take(ps)
    }
  })