Class WebSocketUpstreamConnection

src/vs/server/node/agentHostChannel.ts:99–219

	(endpoint) => new WebSocketUpstreamConnection(endpoint, logService);

class WebSocketUpstreamConnection extends Disposable implements IUpstreamConnection {
	private readonly _onFrame = this._register(new Emitter<string>());
	readonly onFrame: Event<string> = this._onFrame.event;

	private readonly _onClose = this._register(new Emitter<void>());
	readonly onClose: Event<void> = this._onClose.event;

	private _ws: wsTypes.WebSocket | undefined;
	private _connectPromise: Promise<void> | undefined;
	private _closeFired = false;

	constructor(
		private readonly _endpoint: IAgentHostUpstreamEndpoint,
		private readonly _logService: ILogService,
	) {
		super();
	}

	connect(): Promise<void> {
		if (this._store.isDisposed) {
			return Promise.reject(new Error('UpstreamConnection is disposed'));
		}
		return this._connectPromise ??= this._doConnect();
	}

	private async _doConnect(): Promise<void> {
		const ws = await loadWs();
		const url = this._buildUrl();
		const wsOptions = await this._buildWsOptions();

		this._logService.info(`[AgentHostChannel] Opening upstream to ${this._endpoint.socketPath ?? url}`);
		const socket = new ws.WebSocket(url, wsOptions);
		this._ws = socket;

		return new Promise<void>((resolve, reject) => {
			const onOpen = () => {
				cleanup();
				this._logService.trace('[AgentHostChannel] Upstream open');
				socket.on('message', (data: Buffer | string) => {
					const text = typeof data === 'string' ? data : data.toString('utf-8');
					this._onFrame.fire(text);
				});
				socket.on('close', () => this._fireClose());
				socket.on('error', err => {
					this._logService.warn('[AgentHostChannel] Upstream error', err);
					this._fireClose();
				});
				resolve();
			};

			const onError = (err: Error) => {
				cleanup();
				this._logService.warn('[AgentHostChannel] Upstream connection failed', err);
				this._fireClose();
				reject(err);
			};

			const onClose = () => {