Migrate terminal shell integration to AsyncIterableProducer

src/vs/workbench/api/common/extHostTerminalShellIntegration.ts

@@ -12,7 +12,7 @@ import { IExtHostRpcService } from './extHostRpcService.js';
 import { IExtHostTerminalService } from './extHostTerminalService.js';
 import { Emitter, type Event } from '../../../base/common/event.js';
 import { URI } from '../../../base/common/uri.js';
-import { AsyncIterableObject, Barrier, type AsyncIterableEmitter } from '../../../base/common/async.js';
+import { AsyncIterableProducer, Barrier, DeferredPromise, type AsyncIterableEmitter } from '../../../base/common/async.js';
 
 export interface IExtHostTerminalShellIntegration extends ExtHostTerminalShellIntegrationShape {
 	readonly _serviceBrand: undefined;
@@ -400,7 +400,7 @@ class InternalTerminalShellExecution {
 	private _createDataStream(): AsyncIterable<string> {
 		if (!this._dataStream) {
 			if (this._isEnded) {
-				return AsyncIterableObject.EMPTY;
+				return AsyncIterableProducer.EMPTY;
 			}
 			this._dataStream = new ShellExecutionDataStream();
 		}
@@ -432,19 +432,21 @@ class InternalTerminalShellExecution {
 
 class ShellExecutionDataStream extends Disposable {
 	private _barrier: Barrier | undefined;
-	private _iterables: AsyncIterableObject<string>[] = [];
+	private _completionPromises: DeferredPromise<void>[] = [];
 	private _emitters: AsyncIterableEmitter<string>[] = [];
 
 	createIterable(): AsyncIterable<string> {
 		if (!this._barrier) {
 			this._barrier = new Barrier();
 		}
 		const barrier = this._barrier;
-		const iterable = new AsyncIterableObject<string>(async emitter => {
+		const completionPromise = new DeferredPromise<void>();
+		this._completionPromises.push(completionPromise);
+		const iterable = new AsyncIterableProducer<string>(async emitter => {
 			this._emitters.push(emitter);
 			await barrier.wait();
+			completionPromise.complete(undefined);
 		});
-		this._iterables.push(iterable);
 		return iterable;
 	}
 
@@ -459,7 +461,7 @@ class ShellExecutionDataStream extends Disposable {
 	}
 
 	async flush(): Promise<void> {
-		await Promise.all(this._iterables.map(e => e.toPromise()));
+		await Promise.all(this._completionPromises.map(p => p.p));
 	}
 }
 

src/vs/workbench/api/test/common/extHostTerminalShellIntegration.test.ts

@@ -64,7 +64,7 @@ suite('InternalTerminalShellIntegration', () => {
 	}
 
 	async function emitData(data: string): Promise<void> {
-		// AsyncIterableObjects are initialized in a microtask, this doesn't matter in practice
+		// AsyncIterableProducers are initialized in a microtask, this doesn't matter in practice
 		// since the events will always come through in different events.
 		await new Promise<void>(r => queueMicrotask(r));
 		si.emitData(data);