| 1 | import * as Effect from "effect/Effect"; |
| 2 | import * as Stream from "effect/Stream"; |
| 3 | |
| 4 | /** A request-owned prefix, consumed once before its RPC's live stream. */ |
| 5 | export function rpcInitialItems<A>(items: ReadonlyArray<A>): Stream.Stream<A> { |
| 6 | let pending: ReadonlyArray<A> | undefined = items; |
| 7 | return Stream.unwrap( |
| 8 | Effect.sync(() => { |
| 9 | const initial = pending; |
| 10 | // concat keeps its initial stream definition alive during the live tail. |
| 11 | // That definition must no longer own delivered snapshots or replay rows. |
| 12 | pending = undefined; |
| 13 | return initial === undefined ? Stream.empty : Stream.fromIterable(initial); |
| 14 | }), |
| 15 | ); |
| 16 | } |