Advanced

Observable observation

How rs-x subscribes to observables, caches latest emission, and forwards emission changes through state tracking.

Detailed notes

ObservableObserverProxyPairFactory has priority 3 and applies via isObservable().
ObservableProxyFactory is keyed by observable instance and reuses one ObservableProxy per observable.
No proxy replacement object is returned; observer pair carries proxy: undefined.
ObservableProxy starts subscription on init() and keeps one subscription handle.
Emissions are deduplicated by previous value and ignored after dispose.
ObservableAccessor stores last value in IResolvedValueCache and supports Subject.next(...) for setValue.

Step 1: choose observable factory

Code links: ObservableObserverProxyPairFactory

Factory selection

Full implementation of ObservableObserverProxyPairFactory.

@Injectable()
export class ObservableObserverProxyPairFactory
  implements IObjectObserverProxyPairFactory<Observable<unknown>>
{
  public readonly priority = 3;

  constructor(
    @Inject(RsXStateManagerInjectionTokens.IObservableProxyFactory)
    private readonly _observableProxyFactory: IObservableProxyFactory,
  ) {}

  public create(
    owner: IDisposableOwner,
    objectObserverInfo: IProxyTarget<Observable<unknown>>,
  ): IObservableObserverProxyPair {
    return this._observableProxyFactory.create({
      observable: objectObserverInfo.target,
      owner,
    }).instance;
  }

  public applies(object: unknown): boolean {

Factory id reuse

ObservableProxyFactory key is the observable object itself.

public override getId(data: IObservableProxyData): Observable<unknown> {
  return data.observable;
}

protected override createId(data: IObservableProxyData): Observable<unknown> {
  return data.observable;
}

Step 2: subscribe once on init

Code links: ObservableProxyFactory + ObservableProxy

Deferred subscription model

Full implementation of ObservableProxy.

class ObservableProxy extends AbstractObserver<
  Observable<unknown>,
  undefined,
  undefined
> {
  private _observableSubscription: Subscription | undefined;
  private _oldValue: unknown;

  constructor(
    owner: IDisposableOwner,
    target: Observable<unknown>,
    private readonly _observableAccessor: IObservableAccessor,
  ) {
    super(owner, target, undefined, new ReplaySubject<IPropertyChange>(1));
  }

  public override init(): void {
    if (this._observableSubscription) {
      return;
    }
    this._observableSubscription = this.target.subscribe(
      this.emitObservableChange,

Observer pair shape

Full implementation of ObservableProxyFactory.

@Injectable()
export class ObservableProxyFactory
  extends KeyedInstanceFactory<
    Observable<unknown>,
    IObservableProxyData,
    IObservableObserverProxyPair
  >
  implements IObservableProxyFactory
{
  constructor(
    @Inject(RsXCoreInjectionTokens.IObservableAccessor)
    private readonly _observableAccessor: IObservableAccessor,
  ) {
    super();
  }

  public override getId(data: IObservableProxyData): Observable<unknown> {
    return data.observable;
  }

  protected override createId(data: IObservableProxyData): Observable<unknown> {
    return data.observable;

Step 3: cache and emit on each next()

Code links: ObservableProxy · ObservableAccessor

Emission dedupe and guard

The observer skips duplicate consecutive values and skips emissions after dispose.

private emitObservableChange = (newValue: unknown): void => {
  if (newValue === this._oldValue || this.isDisposed) {
    return;
  }

  this._oldValue = newValue;
  this._observableAccessor.setLastValue(this.target, newValue);

  this.emitChange({
    arguments: [],
    chain: [],
    target: this.target,
    index: this.id,
    newValue,
  });
};

Accessor cache implementation

ObservableAccessor writes values to resolved-value cache keyed by observable object.

public setLastValue(observable: LastValueObservable, value: unknown): void {
  this._resolvedValueCache.set(observable, value);
}

public clearLastValue(observable: LastValueObservable): void {
  this._resolvedValueCache.delete(observable);
}

Step 4: read/write behavior through ObservableAccessor

Code links: ObservableAccessor · ResolvedValueCache

Resolved read behavior

BehaviorSubject returns context.value directly; other observables use cache and fallback to PENDING.

public getResolvedValue(context: unknown, index: string): unknown {
  if (context instanceof BehaviorSubject) return context.value;

  const val = this.getIndexedValue(context, index);
  if ((val && typeof val === 'object') || typeof val === 'function') {
    return this._resolvedValueCache.get(val) ?? PENDING;
  }

  return PENDING;
}

Write behavior for Subject

setValue forwards to next(value) when the observed value is a Subject.

public setValue(context: unknown, index: string, value: unknown): void {
  const val = this.getIndexedValue(context, index);
  if (val instanceof Subject) {
    val.next(value);
    return;
  }
}

Step 5: cleanup and unsubscribe

Code links: ObservableProxyFactory + ObservableProxy

ObservableProxy dispose

Dispose unsubscribes and clears cache value for this observable.

protected override disposeInternal(): void {
  this._observableSubscription?.unsubscribe();
  this._observableAccessor.clearLastValue(this.target);
  this._observableSubscription = undefined;
  this._oldValue = null;
}

Factory release

On final release the observer is disposed.

protected override releaseInstance(
  observableObserverWithProxy: IObservableObserverProxyPair,
): void {
  observableObserverWithProxy.observer.dispose();
}