| 1 | /** |
| 2 | * Low-level helpers for adapting push-based SQL row sources into Effect |
| 3 | * streams. |
| 4 | * |
| 5 | * SQL drivers often expose large query results through cursors, event emitters, |
| 6 | * or driver-specific streams that push rows as they arrive. This module |
| 7 | * provides the small interop layer used by SQL integrations to turn those |
| 8 | * producers into `Stream` values for `Statement.stream` and |
| 9 | * `Connection.executeStream`, so callers can process large result sets |
| 10 | * incrementally instead of materializing every row in memory. |
| 11 | * |
| 12 | * @stability unstable |
| 13 | * @since 4.0.0 |
| 14 | */ |
| 15 | import * as Cause from "../Cause.ts" |
| 16 | import * as Effect from "../Effect.ts" |
| 17 | import * as Queue from "../Queue.ts" |
| 18 | import type * as Scope from "../Scope.ts" |
| 19 | import * as Stream from "../Stream.ts" |
| 20 | |
| 21 | /** |
| 22 | * Creates a stream from a callback-style producer with pause and resume |
| 23 | * callbacks that are triggered when the internal queue applies backpressure. |
| 24 | * |
| 25 | * @stability unstable |
| 26 | * @category constructors |
| 27 | * @since 4.0.0 |
| 28 | */ |
| 29 | export const asyncPauseResume = <A, E = never, R = never>( |
| 30 | register: (emit: { |
| 31 | readonly single: (item: A) => void |
| 32 | readonly array: (arr: ReadonlyArray<A>) => void |
| 33 | readonly fail: (error: E) => void |
| 34 | readonly end: () => void |
| 35 | }) => Effect.Effect< |
| 36 | { |
| 37 | onPause(): void |
| 38 | onResume(): void |
| 39 | }, |
| 40 | E, |
| 41 | R | Scope.Scope |
| 42 | >, |
| 43 | bufferSize = 128 |
| 44 | ): Stream.Stream<A, E, R> => |
| 45 | Stream.callback<A, E, R>((queue) => |
| 46 | Effect.suspend(() => { |
| 47 | let cbs!: { |
| 48 | onPause(): void |
| 49 | onResume(): void |
| 50 | } |
| 51 | |
| 52 | let paused = false |
| 53 | const offer = (arr: ReadonlyArray<A>) => { |
| 54 | if (arr.length === 0) return |
| 55 | const isFull = Queue.isFullUnsafe(queue) |
| 56 | if (!isFull || (isFull && paused)) { |
| 57 | return Effect.runFork(Queue.offerAll(queue, arr)) |
| 58 | } |
| 59 | paused = true |
| 60 | cbs.onPause() |
| 61 | return Queue.offerAll(queue, arr).pipe( |
| 62 | Effect.tap(() => |
| 63 | Effect.sync(() => { |
| 64 | cbs.onResume() |
| 65 | paused = false |
| 66 | }) |
| 67 | ), |
| 68 | Effect.runFork |
| 69 | ) |
| 70 | } |
| 71 | |
| 72 | return Effect.map( |
| 73 | register({ |
| 74 | single: (item) => offer([item]), |
| 75 | array: (chunk) => offer(chunk), |
| 76 | fail: (error) => Queue.failCauseUnsafe(queue as any, Cause.fail(error)), |
| 77 | end: () => Queue.endUnsafe(queue as any) |
| 78 | }), |
| 79 | (_) => { |
| 80 | cbs = _ |
| 81 | } |
| 82 | ) |
| 83 | }), { bufferSize }) |