Advanced

Async operations

How rs-x unifies Promise, Observable, and Expression updates into one consistent observation and commit pipeline.

Detailed notes

There is no separate async API mode: Promise, Observable, and Expression values go through the same observation pipeline as other values.
Metadata is used at concrete decision points: ValueMetadata.isAsync(...) is checked in indexed observer factories to attach async observers, and ValueMetadata.needsProxy(...) is checked in IdentifierExpression to decide recursive proxying for member paths.
PENDING is internal runtime state used while async values are unresolved; it is not a user-facing expression value.
Promise observers cache resolved values and emit property changes when resolution arrives.
Observable observers cache emissions, dedupe consecutive equal values, and emit through state manager.
Expression observers forward expression.changed into the same property-change pipeline used by Promise/Observable.
State manager wraps async updates in startChangeCycle/endChangeCycle, and transaction commit is deferred until cycle depth returns to zero.

Step 1: classify async value types

Code links: ValueMetadata dispatcher · PromiseMetadata · ObservableMetadata · ExpressionMetadata registration · PromiseAccessor · ObservableAccessor

Metadata marks values as async/proxy-aware

@Injectable()
export class ExpressionMetadata implements IValueMetadata {
  public readonly priority = 300;

  public isAsync(): boolean {
    return true;
  }

  public needsProxy(): boolean {
    return true;
  }

  public applies(value: unknown): boolean {
    return value instanceof AbstractExpression;
  }
}

Accessor fallback uses internal PENDING

// Promise/Observable accessor pattern (simplified)
const cached = accessor.getResolvedValue(context, index);
return cached === undefined ? PENDING : cached;

Step 2: observe async sources

Code links: Promise observer implementation · Observable observer implementation · Expression observer proxy factory

Promise resolution updates cache + emits change

target.then((resolved) => {
  this._promiseAccessor.setValue(context, index, resolved);
  this.emitChange({
    chain: [{ context, index }],
    target: context,
    index,
    newValue: resolved,
  });
});

Observable next(value) follows same pipeline

this._subscription = observable.subscribe((value) => {
  if (this._equalityService.isEqual(this.value, value)) return;
  this._observableAccessor.setValue(context, index, value);
  this.emitChange({
    chain: [{ context, index }],
    target: context,
    index,
    newValue: value,
  });
});

Step 3: defer commit until data is ready

Code links: AbstractExpression (PENDING handling) · ExpressionChangeTransactionManager · StateManager change-cycle wiring

Internal PENDING pauses the current reevaluation pass

protected evaluateBottomToTop(...): boolean {
  const value = this.evaluate(sender, root);
  if (value === PENDING) {
    return false;
  }
  this._value = value;
  return this.parent ? this.parent.reevaluated(...) : true;
}

Commit runs after change-cycle depth returns to zero

onEndChangeCycle() {
  counter--;
  if (counter === 0 && !suspended) commit();
}