| 1 | import type * as Arr from "effect/Array" |
| 2 | import type * as Cause from "effect/Cause" |
| 3 | import * as Effect from "effect/Effect" |
| 4 | import * as Queue from "effect/Queue" |
| 5 | import * as Stream from "effect/Stream" |
| 6 | |
| 7 | export interface ChunkCoordination<A> { |
| 8 | readonly queue: Queue.Queue<Arr.NonEmptyReadonlyArray<A>, Cause.Done> |
| 9 | readonly stream: Stream.Stream<A> |
| 10 | readonly offer: Effect.Effect<void> |
| 11 | readonly proceed: Effect.Effect<void> |
| 12 | readonly awaitNext: Effect.Effect<void> |
| 13 | } |
| 14 | |
| 15 | export const chunkCoordination = <A>( |
| 16 | chunks: ReadonlyArray<Arr.NonEmptyReadonlyArray<A>> |
| 17 | ): Effect.Effect<ChunkCoordination<A>> => |
| 18 | Effect.gen(function*() { |
| 19 | let i = 0 |
| 20 | const queue = yield* Queue.unbounded<Arr.NonEmptyReadonlyArray<A>, Cause.Done>() |
| 21 | const ps = yield* Queue.unbounded<void>() |
| 22 | return { |
| 23 | queue, |
| 24 | stream: Stream.flattenArray(Stream.fromQueue(queue)), |
| 25 | offer: Effect.gen(function*() { |
| 26 | if (i < chunks.length) { |
| 27 | yield* Queue.offer(queue, chunks[i++]) |
| 28 | } |
| 29 | if (i >= chunks.length) { |
| 30 | yield* Queue.end(queue) |
| 31 | } |
| 32 | }), |
| 33 | proceed: Effect.asVoid(Queue.offer(ps, void 0)), |
| 34 | awaitNext: Queue.take(ps) |
| 35 | } |
| 36 | }) |