| 74 | * @since 4.0.0 |
| 75 | */ |
| 76 | export const pullIntoWritable = <A, IE, E>(options: { |
| 77 | readonly pull: Pull.Pull<NonEmptyReadonlyArray<A>, IE, unknown> |
| 78 | readonly writable: Writable |
| 79 | readonly onError: (error: unknown) => E |
| 80 | readonly endOnDone?: boolean | undefined |
| 81 | readonly encoding?: BufferEncoding | undefined |
| 82 | }): Pull.Pull<never, IE | E, unknown> => |
| 83 | options.pull.pipe( |
| 84 | Effect.flatMap((chunk) => { |
| 85 | let i = 0 |
| 86 | return Effect.callback<void, E>(function loop(resume) { |
| 87 | for (; i < chunk.length;) { |
| 88 | const success = options.writable.write(chunk[i++], options.encoding as any) |
| 89 | if (!success) { |
| 90 | options.writable.once("drain", () => (loop as any)(resume)) |
| 91 | return |
| 92 | } |
| 93 | } |
| 94 | resume(Effect.void) |
| 95 | }) |
| 96 | }), |
| 97 | Effect.forever({ disableYield: true }), |
| 98 | Effect.raceFirst(Effect.callback<never, E>((resume) => { |
| 99 | const onError = (error: unknown) => resume(Effect.fail(options.onError(error))) |
| 100 | options.writable.once("error", onError) |
| 101 | return Effect.sync(() => { |
| 102 | options.writable.off("error", onError) |
| 103 | }) |
| 104 | })), |
| 105 | options.endOnDone !== false ? |
| 106 | Pull.catchDone((_) => { |
| 107 | if ("closed" in options.writable && options.writable.closed) { |
| 108 | return Cause.done(_) |
| 109 | } |
| 110 | return Effect.callback<never, E | Cause.Done<unknown>>((resume) => { |
| 111 | options.writable.once("finish", () => resume(Cause.done(_))) |
| 112 | options.writable.end() |
| 113 | }) |
| 114 | }) : |
| 115 | identity |
| 116 | ) |