diff --git a/src/observable/ObservableValue.ts b/src/observable/ObservableValue.ts index b3ffa6ee..15d931ff 100644 --- a/src/observable/ObservableValue.ts +++ b/src/observable/ObservableValue.ts @@ -16,6 +16,7 @@ limitations under the License. import {AbortError} from "../utils/error"; import {BaseObservable} from "./BaseObservable"; +import type {SubscriptionHandle} from "./BaseObservable"; // like an EventEmitter, but doesn't have an event type export abstract class BaseObservableValue extends BaseObservable<(value: T) => void> { @@ -34,6 +35,10 @@ export abstract class BaseObservableValue extends BaseObservable<(value: T) = return new WaitForHandle(this, predicate); } } + + flatMap(mapper: (value: T) => (BaseObservableValue | undefined)): BaseObservableValue { + return new FlatMapObservableValue(this, mapper); + } } interface IWaitHandle { @@ -114,6 +119,61 @@ export class RetainedObservableValue extends ObservableValue { } } +export class FlatMapObservableValue extends BaseObservableValue { + private sourceSubscription?: SubscriptionHandle; + private targetSubscription?: SubscriptionHandle; + + constructor( + private readonly source: BaseObservableValue

, + private readonly mapper: (value: P) => (BaseObservableValue | undefined) + ) { + super(); + } + + onUnsubscribeLast() { + super.onUnsubscribeLast(); + this.sourceSubscription = this.sourceSubscription!(); + if (this.targetSubscription) { + this.targetSubscription = this.targetSubscription(); + } + } + + onSubscribeFirst() { + super.onSubscribeFirst(); + this.sourceSubscription = this.source.subscribe(() => { + this.updateTargetSubscription(); + this.emit(this.get()); + }); + this.updateTargetSubscription(); + } + + private updateTargetSubscription() { + const sourceValue = this.source.get(); + if (sourceValue) { + const target = this.mapper(sourceValue); + if (target) { + if (!this.targetSubscription) { + this.targetSubscription = target.subscribe(() => this.emit(this.get())); + } + return; + } + } + // if no sourceValue or target + if (this.targetSubscription) { + this.targetSubscription = this.targetSubscription(); + } + } + + get(): C | undefined { + const sourceValue = this.source.get(); + if (!sourceValue) { + return undefined; + } + const mapped = this.mapper(sourceValue); + return mapped?.get(); + } +} + export function tests() { return { "set emits an update": assert => { @@ -155,5 +215,34 @@ export function tests() { }); await assert.rejects(handle.promise, AbortError); }, + "flatMap.get": assert => { + const a = new ObservableValue}>(undefined); + const countProxy = a.flatMap(a => a.count); + assert.strictEqual(countProxy.get(), undefined); + const count = new ObservableValue(0); + a.set({count}); + assert.strictEqual(countProxy.get(), 0); + }, + "flatMap update from source": assert => { + const a = new ObservableValue}>(undefined); + const updates: (number | undefined)[] = []; + a.flatMap(a => a.count).subscribe(count => { + updates.push(count); + }); + const count = new ObservableValue(0); + a.set({count}); + assert.deepEqual(updates, [0]); + }, + "flatMap update from target": assert => { + const a = new ObservableValue}>(undefined); + const updates: (number | undefined)[] = []; + a.flatMap(a => a.count).subscribe(count => { + updates.push(count); + }); + const count = new ObservableValue(0); + a.set({count}); + count.set(5); + assert.deepEqual(updates, [0, 5]); + } } }