debt - add Event#fromObservable to bridge between observables and emitters

src/vs/base/common/event.ts

@@ -8,6 +8,7 @@ import { onUnexpectedError } from 'vs/base/common/errors';
 import { once as onceFn } from 'vs/base/common/functional';
 import { combinedDisposable, Disposable, DisposableStore, IDisposable, SafeDisposable, toDisposable } from 'vs/base/common/lifecycle';
 import { LinkedList } from 'vs/base/common/linkedList';
+import { IObservable, IObserver } from 'vs/base/common/observable';
 import { StopWatch } from 'vs/base/common/stopwatch';
 
 
@@ -423,6 +424,55 @@ export namespace Event {
 			store?.dispose();
 		});
 	}
+
+	class EmitterObserver<T> implements IObserver {
+
+		readonly emitter: Emitter<T>;
+
+		private _counter = 0;
+		private _hasChanged = false;
+
+		constructor(readonly obs: IObservable<T, any>, store: DisposableStore | undefined) {
+			const options = {
+				onFirstListenerAdd: () => {
+					obs.addObserver(this);
+				},
+				onLastListenerRemove: () => {
+					obs.removeObserver(this);
+				}
+			};
+			if (!store) {
+				_addLeakageTraceLogic(options);
+			}
+			this.emitter = new Emitter<T>(options);
+			if (store) {
+				store.add(this.emitter);
+			}
+		}
+
+		beginUpdate<T>(_observable: IObservable<T, void>): void {
+			// console.assert(_observable === this.obs);
+			this._counter++;
+		}
+
+		handleChange<T, TChange>(_observable: IObservable<T, TChange>, _change: TChange): void {
+			this._hasChanged = true;
+		}
+
+		endUpdate<T>(_observable: IObservable<T, void>): void {
+			if (--this._counter === 0) {
+				if (this._hasChanged) {
+					this._hasChanged = false;
+					this.emitter.fire(this.obs.get());
+				}
+			}
+		}
+	}
+
+	export function fromObservable<T>(obs: IObservable<T, any>, store?: DisposableStore): Event<T> {
+		const observer = new EmitterObserver(obs, store);
+		return observer.emitter.event;
+	}
 }
 
 export interface EmitterOptions {

src/vs/base/test/common/event.test.ts

@@ -8,7 +8,8 @@ import { CancellationToken } from 'vs/base/common/cancellation';
 import { errorHandler, setUnexpectedErrorHandler } from 'vs/base/common/errors';
 import { AsyncEmitter, DebounceEmitter, Emitter, Event, EventBufferer, EventMultiplexer, IWaitUntil, MicrotaskEmitter, PauseableEmitter, Relay } from 'vs/base/common/event';
 import { DisposableStore, IDisposable, isDisposable, setDisposableTracker, toDisposable } from 'vs/base/common/lifecycle';
-import { DisposableTracker } from 'vs/base/test/common/utils';
+import { observableValue, transaction } from 'vs/base/common/observable';
+import { DisposableTracker, ensureNoDisposablesAreLeakedInTestSuite } from 'vs/base/test/common/utils';
 
 namespace Samples {
 
@@ -624,6 +625,33 @@ suite('PausableEmitter', function () {
 	});
 });
 
+suite('Event utils - ensureNoDisposablesAreLeakedInTestSuite', function () {
+	ensureNoDisposablesAreLeakedInTestSuite();
+
+	test('fromObservable', function () {
+
+		const obs = observableValue('test', 12);
+		const event = Event.fromObservable(obs);
+
+		const values: number[] = [];
+		const d = event(n => { values.push(n); });
+
+		obs.set(3, undefined);
+		obs.set(13, undefined);
+		obs.set(3, undefined);
+		obs.set(33, undefined);
+		obs.set(1, undefined);
+
+		transaction(tx => {
+			obs.set(334, tx);
+			obs.set(99, tx);
+		});
+
+		assert.deepStrictEqual(values, ([3, 13, 3, 33, 1, 99]));
+		d.dispose();
+	});
+});
+
 suite('Event utils', () => {
 
 	suite('EventBufferer', () => {

src/vs/workbench/contrib/mergeEditor/browser/mergeEditorInput.ts

@@ -17,12 +17,12 @@ import { IEditorIdentifier, IUntypedEditorInput } from 'vs/workbench/common/edit
 import { EditorInput } from 'vs/workbench/common/editor/editorInput';
 import { AbstractTextResourceEditorInput } from 'vs/workbench/common/editor/textResourceEditorInput';
 import { EditorWorkerServiceDiffComputer } from 'vs/workbench/contrib/mergeEditor/browser/model/diffComputer';
-import { autorun } from 'vs/workbench/contrib/audioCues/browser/observable';
 import { MergeEditorModel } from 'vs/workbench/contrib/mergeEditor/browser/model/mergeEditorModel';
 import { IEditorService } from 'vs/workbench/services/editor/common/editorService';
 import { AutoSaveMode, IFilesConfigurationService } from 'vs/workbench/services/filesConfiguration/common/filesConfigurationService';
 import { ILanguageSupport, ITextFileEditorModel, ITextFileService } from 'vs/workbench/services/textfile/common/textfiles';
 import { assertType } from 'vs/base/common/types';
+import { Event } from 'vs/base/common/event';
 
 export class MergeEditorInputData {
 	constructor(
@@ -123,10 +123,7 @@ export class MergeEditorInput extends AbstractTextResourceEditorInput implements
 			this._store.add(input2);
 			this._store.add(result);
 
-			this._store.add(autorun(reader => {
-				this._model?.hasUnhandledConflicts.read(reader);
-				this._onDidChangeDirty.fire(undefined);
-			}, 'drive::onDidChangeDirty'));
+			this._store.add(Event.fromObservable(this._model.hasUnhandledConflicts)(() => this._onDidChangeDirty.fire(undefined)));
 		}
 
 		this._ignoreUnhandledConflictsForDirtyState = undefined;