| 102 | // Parse a JSONL byte stream from the wire back into exec events. |
| 103 | // Buffers partial lines so a chunk boundary can fall anywhere. |
| 104 | export function decodeExecEvents<E extends ExecEncoding>( |
| 105 | bytes: ReadableStream<Uint8Array>, |
| 106 | ): ReadableStream<WorkspaceExecEvent<E>> { |
| 107 | const decoder = new TextDecoder("utf-8"); |
| 108 | let buffer = ""; |
| 109 | const emitLine = ( |
| 110 | line: string, |
| 111 | controller: TransformStreamDefaultController<WorkspaceExecEvent<E>>, |
| 112 | ) => { |
| 113 | if (line.length === 0) return; |
| 114 | const frame = JSON.parse(line) as ExecFrame; |
| 115 | controller.enqueue(eventOf(frame) as WorkspaceExecEvent<E>); |
| 116 | }; |
| 117 | return bytes.pipeThrough( |
| 118 | new TransformStream<Uint8Array, WorkspaceExecEvent<E>>({ |
| 119 | transform(chunk, controller) { |
| 120 | buffer += decoder.decode(chunk, { stream: true }); |
| 121 | let newline = buffer.indexOf("\n"); |
| 122 | while (newline !== -1) { |
| 123 | emitLine(buffer.slice(0, newline), controller); |
| 124 | buffer = buffer.slice(newline + 1); |
| 125 | newline = buffer.indexOf("\n"); |
| 126 | } |
| 127 | }, |
| 128 | flush(controller) { |
| 129 | buffer += decoder.decode(); |
| 130 | // A well-formed stream ends each event with a newline, so the |
| 131 | // tail is normally empty. Emit any trailing line defensively. |
| 132 | emitLine(buffer, controller); |
| 133 | buffer = ""; |
| 134 | }, |
| 135 | }), |
| 136 | ); |
| 137 | } |