| 1 | import { registerTerminalOutputSink } from "./terminalEvents"; |
| 2 | |
| 3 | export type TerminalSinkSubscription = { |
| 4 | setActive: (active: boolean) => void; |
| 5 | dispose: () => void; |
| 6 | }; |
| 7 | |
| 8 | export function registerTerminalSink( |
| 9 | id: string, |
| 10 | sink: (data: Uint8Array) => void, |
| 11 | initiallyActive = true, |
| 12 | ): TerminalSinkSubscription { |
| 13 | let active = initiallyActive; |
| 14 | let disposed = false; |
| 15 | let cursor = 0; |
| 16 | const [unregister, history] = registerTerminalOutputSink(id, (bytes, sequence) => { |
| 17 | if (!active || disposed) return; |
| 18 | if (cursor === sequence) { |
| 19 | sink(bytes); |
| 20 | cursor = sequence + 1; |
| 21 | return; |
| 22 | } |
| 23 | flush(); |
| 24 | }); |
| 25 | const initialHistory = history(); |
| 26 | cursor = initialHistory[1] - initialHistory[0].length; |
| 27 | const flush = () => { |
| 28 | if (!active || disposed) return; |
| 29 | const [chunks, nextSequence] = history(); |
| 30 | const firstSequence = nextSequence - chunks.length; |
| 31 | const first = Math.max(cursor, firstSequence); |
| 32 | for (let sequence = first; sequence < nextSequence; sequence += 1) { |
| 33 | const bytes = chunks[sequence - firstSequence]; |
| 34 | if (bytes) sink(bytes); |
| 35 | } |
| 36 | // Output larger than the retained limit is intentionally skipped; advance |
| 37 | // the cursor so the next live chunk can continue without replaying old data. |
| 38 | cursor = nextSequence; |
| 39 | }; |
| 40 | flush(); |
| 41 | return { |
| 42 | setActive(nextActive) { |
| 43 | if (disposed || active === nextActive) return; |
| 44 | active = nextActive; |
| 45 | if (active) flush(); |
| 46 | }, |
| 47 | dispose() { |
| 48 | disposed = true; |
| 49 | unregister(); |
| 50 | }, |
| 51 | }; |
| 52 | } |
| 53 |