Skip to content

Theoretical handling of stream non-availability or delayed availability #1435

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cte merged 1 commit into from
Mar 7, 2025
Merged

Theoretical handling of stream non-availability or delayed availability #1435

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 41 additions & 21 deletions src/integrations/terminal/TerminalManager.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,10 @@ declare module "vscode" {
// https://github.com/microsoft/vscode/blob/f0417069c62e20f3667506f4b7e53ca0004b4e3e/src/vscode-dts/vscode.d.ts#L10794
interface Window {
onDidStartTerminalShellExecution?: (
listener: (e: any) => any,
listener: (e: {
terminal: vscode.Terminal
execution: { read(): AsyncIterable<string>; commandLine: { value: string } }
}) => any,
thisArgs?: any,
disposables?: vscode.Disposable[],
) => vscode.Disposable
Expand Down Expand Up @@ -203,57 +206,77 @@ export class TerminalManager {
constructor() {
let startDisposable: vscode.Disposable | undefined
let endDisposable: vscode.Disposable | undefined

try {
// onDidStartTerminalShellExecution
startDisposable = (vscode.window as vscode.Window).onDidStartTerminalShellExecution?.(async (e) => {
// Get a handle to the stream as early as possible:
const stream = e?.execution.read()
const terminalInfo = TerminalRegistry.getTerminalInfoByTerminal(e.terminal)
if (stream && terminalInfo) {
const process = this.processes.get(terminalInfo.id)
if (process) {
terminalInfo.stream = stream
terminalInfo.running = true
terminalInfo.streamClosed = false
process.emit("stream_available", terminalInfo.id, stream)
}
} else {
console.error("[TerminalManager] Stream failed, not registered for terminal")
}

console.info("[TerminalManager] Shell execution started:", {
console.info("[TerminalManager] shell execution started", {
hasExecution: !!e?.execution,
hasStream: !!stream,
command: e?.execution?.commandLine?.value,
terminalId: terminalInfo?.id,
})

if (terminalInfo) {
const process = this.processes.get(terminalInfo.id)

if (process) {
if (stream) {
terminalInfo.stream = stream
terminalInfo.running = true
terminalInfo.streamClosed = false
console.log(`[TerminalManager] stream_available -> ${terminalInfo.id}`)
process.emit("stream_available", terminalInfo.id, stream)
} else {
process.emit("stream_unavailable", terminalInfo.id)
console.error(`[TerminalManager] stream_unavailable -> ${terminalInfo.id}`)
}
}
} else {
console.error("[TerminalManager] terminalInfo not available")
}
})

// onDidEndTerminalShellExecution
endDisposable = (vscode.window as vscode.Window).onDidEndTerminalShellExecution?.(async (e) => {
const exitDetails = this.interpretExitCode(e?.exitCode)
console.info("[TerminalManager] Shell execution ended:", {
...exitDetails,
})
console.info("[TerminalManager] Shell execution ended:", { ...exitDetails })
let emitted = false

// Signal completion to any waiting processes
// Signal completion to any waiting processes.
for (const id of this.terminalIds) {
const info = TerminalRegistry.getTerminal(id)

if (info && info.terminal === e.terminal) {
info.running = false
const process = this.processes.get(id)

if (process) {
console.log(`[TerminalManager] emitting shell_execution_complete -> ${id}`)
emitted = true
process.emit("shell_execution_complete", id, exitDetails)
}

break
}
}

if (!emitted) {
console.log(`[TerminalManager#onDidStartTerminalShellExecution] no terminal found`)
}
})
} catch (error) {
console.error("[TerminalManager] Error setting up shell execution handlers:", error)
console.error("[TerminalManager] failed to configure shell execution handlers", error)
}

if (startDisposable) {
this.disposables.push(startDisposable)
}

if (endDisposable) {
this.disposables.push(endDisposable)
}
Expand Down Expand Up @@ -366,9 +389,6 @@ export class TerminalManager {
}

disposeAll() {
// for (const info of this.terminals) {
// //info.terminal.dispose() // dont want to dispose terminals when task is aborted
// }
this.terminalIds.clear()
this.processes.clear()
this.disposables.forEach((disposable) => disposable.dispose())
Expand Down
45 changes: 38 additions & 7 deletions src/integrations/terminal/TerminalProcess.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ export interface TerminalProcessEvents {
*/
shell_execution_complete: [id: number, exitDetails: ExitCodeDetails]
stream_available: [id: number, stream: AsyncIterable<string>]
stream_unavailable: [id: number]
/**
* Emitted when an execution fails to emit a "line" event for a given period of time.
* @param id The terminal ID
Expand Down Expand Up @@ -85,15 +86,24 @@ export class TerminalProcess extends EventEmitter<TerminalProcessEvents> {
const terminalInfo = TerminalRegistry.getTerminalInfoByTerminal(terminal)

if (!terminalInfo) {
console.error("[TerminalProcess] Terminal not found in registry")
console.error("[TerminalProcess#run] terminal not found in registry")
this.emit("no_shell_integration")
this.emit("completed")
this.emit("continue")
return
}

// When executeCommand() is called, onDidStartTerminalShellExecution will fire in TerminalManager
// which creates a new stream via execution.read() and emits 'stream_available'
this.once("stream_unavailable", (id: number) => {
if (id === terminalInfo.id) {
console.error(`[TerminalProcess#run] stream_unavailable`)
this.emit("completed")
this.emit("continue")
}
})

// When `executeCommand()` is called, `onDidStartTerminalShellExecution`
// will fire in `TerminalManager` which creates a new stream via
// `execution.read()` and emits `stream_available`.
const streamAvailable = new Promise<AsyncIterable<string>>((resolve) => {
this.once("stream_available", (id: number, stream: AsyncIterable<string>) => {
if (id === terminalInfo.id) {
Expand All @@ -111,15 +121,35 @@ export class TerminalProcess extends EventEmitter<TerminalProcessEvents> {
})
})

// readLine needs to know if streamClosed, so store this for later
// `readLine()` needs to know if streamClosed, so store this for later.
// NOTE: This doesn't seem to be used anywhere.
this.terminalInfo = terminalInfo

// Execute command
// Execute command.
terminal.shellIntegration.executeCommand(command)
this.isHot = true

// Wait for stream to be available
const stream = await streamAvailable
// Wait for stream to be available.
// const stream = await streamAvailable

// Wait for stream to be available.
let stream: AsyncIterable<string>

try {
stream = await Promise.race([
streamAvailable,
new Promise<never>((_, reject) => {
setTimeout(
() => reject(new Error("Timeout waiting for terminal stream to become available")),
10_000,
)
}),
])
} catch (error) {
console.error(`[TerminalProcess#run] timed out waiting for stream`)
this.emit("stream_stalled", terminalInfo.id)
stream = await streamAvailable
}

let preOutput = ""
let commandOutputStarted = false
Expand Down Expand Up @@ -256,6 +286,7 @@ export class TerminalProcess extends EventEmitter<TerminalProcessEvents> {
}

public continue() {
console.log(`[TerminalProcess#continue] flushing all`)
this.flushAll()
this.isListening = false
this.removeAllListeners("line")
Expand Down