changes<T> static method

Stream<DVModelChange<T>> changes<T>({ bool allTenants = false, })

Typed change stream for T.

Delivery is filtered before the watcher sees anything: changes from another tenant are dropped unless allTenants is set, and a registered policy must accept the model.

Implementation

static Stream<DVModelChange<T>> changes<T>({bool allTenants = false}) {
  return _controllerFor(T).stream.where((Object? event) {
    if (event is! DVModelChange<T>) {
      // Transport-injected events arrive as DVModelChange<Object?>; match
      // on the model's runtime type instead.
      return event is DVModelChange<Object?> && event.model is T;
    }
    return true;
  }).map((Object? event) {
    final raw = event! as DVModelChange<Object?>;
    return event is DVModelChange<T>
        ? event
        : DVModelChange<T>(
            kind: raw.kind,
            model: raw.model as T,
            tenant: raw.tenant,
            at: raw.at,
          );
  }).where((DVModelChange<T> change) {
    if (!allTenants &&
        change.tenant != const DVTenants().currentTenant) {
      return false;
    }
    final policy = _policies[T];
    return policy == null || policy(change.model);
  });
}