Angular: the viewmodel of a component as an Observable

On this page · 14 sections

Aims of the technique

The primary objectives of this approach are:

eliminate the need for manual subscription cleanup.
enable the use of ChangeDetectionStrategy.onPush for better performance.
offer a solution that relies on no external libraries.
ensure the method is versatile enough for any scenario.
drive all viewmodel modifications and user interactions through observables.

Core concept

The foundation of this solution rests on these three pillars:

every component possesses a viewmodel, and any modification to it originates from a unified, combined observable.
this viewmodel observable is structured as a stream of pure functions that describe how the viewmodel should change.
the rxjs scan operator is used to apply and accumulate these mutations over time.

The first point ensures that only one subscription is needed, which the async pipe manages, thereby handling all unsubscriptions automatically.

Viewmodel mutation functions provide a clean mapping between any observable event and a corresponding change in the viewmodel's state.

The scan operator uses these mutation functions on the current state when any source observable emits a value, producing the new mutated state by applying the function to the previous one.

Illustrative example

Let's begin with a basic, if somewhat impractical, example to demonstrate the core mechanics—a simple counter controlled by buttons.

To start, we'll outline the viewmodel interface:

interface ICounterVm {
  counter: number;
}

Next, we create two subjects, one for each button's action.

// normally it could be done with a single subject, but for demonstration
// purposes, I'll use 2 subjects
public incrSubj = new Subject<number>;
public decrSubj = new Subject<number>;

The viewmodel observable is then defined, and in the component's constructor, we wire up the interactions:

public vm$ : Observable<ICounterVm>;
public incrSubj = new Subject<number>;
public decrSubj = new Subject<number>;
constructor() {

  // the subjects are mapped to an anonymous function that
  // - accepts as parameter the previous state of the viewmodel (vm:ICounterVm)
  // - and that returns the mutated viewmodel
  // they are the viewmodel mutation functions
  const incr$ = this.incrSubj.pipe(
    map( delta => (vm:ICounterVm) => ({...vm, counter:vm.counter+delta}) )
  );
  const decr$ = this.decrSubj.pipe(
    map( delta => (vm:ICounterVm) => ({...vm, counter:vm.counter-delta}))
  );

  // the viewmodel observable is a merge of all mutation observables (incr$ and decr$)
  // piped into a scan function
  // scan has two arguments
  // the first is the accumulator (the viewmodel) and the second the mutation function
  // the body of the scan operator executes the mutation function : mutationFn(prevVm) passing the previous state of the viewmodel.
  // this function returns the mutated viewmodel which is the new accumulated value of the vm$ observable
  this.vm$ = merge(of({counter:0}), incr$, decr$).pipe(
    scan( (prevVm:ICounterVm, mutationFn:(vm:ICounterVm)=>ICountVm)
      => mutationFn(prevVm)
    )
  )
}

Here is the corresponding view markup:

<!-- vm$ is subscribed by async pipe and exposes a variable named vm -->
<div *ngIf="vm$ | async as vm">
  Current counter : {{vm.counter}}
  <button type="button" (click)="decrSubj.next(1)">Decrement</button>
  <button type="button" (click)="incrSubj.next(1)">Increment</button>
</div>

Including viewmodel data in interactions

A common need is to pass a value from the current viewmodel to a subject, for instance, to show the details of a specific item selected from a list.

interface IPersonVm {
  persons:IPerson[];
  personDetail:IPersonDetail;
}

// this subject will be used to pass the person object
// when selecting a person from the list
// <div class="personrow" *ngFor="let person of vm.persons" (click)="personDetailSubj.next(person)"> ... </div>
public personDetailSubj = new Subject<IPerson>();
public vm$ : Observable<IPersonVm>;

constructor(private personService:PersonService) {

  // retrieving list of persons (could be a http request)
  const personList$ = this.personService.getPersons().pipe(
    map( persons => (vm:IPersonVm) => ({...vm, persons}) )
  );

  // select a person, get detail and set it on viewmodel
  const personDetail$ = this.personDetailSubj.pipe(
    mergeMap( person => this.personService.getPersonDetail(person.id) ),
    map( personDetail => (vm:IPersonVm) => ({...vm, personDetail }))
  );

  // in this example the initial viewmodel state is provided with the second
  // parameter of the scan function. Alternatively one could provide an initial
  // state with the rxjs of function
  const vm$ = merge(personList$, personDetail$).pipe(
    scan( (vm:IPersonVm, mutationFn:(vm:IPersonVm)=>IPersonVm)
       => mutationFn(vm), {persons:[], personDetail:null}
    )
  )

}

A few important notes about the initial state:

If the initial state is defined inline as an argument to scan, the viewmodel won't emit a value until at least one observable within the merge produces an event.
Should no observable emit (e.g., if they are all subjects and none are triggered), an initial value must be supplied using the rxjs of operator to seed the stream.
Beware that with the of operator, the scan function is bypassed. The vm$ observable will emit the provided value directly without any processing. This is a quirk of scan that has tripped me up on occasion.

I also use the spread operator {...vm, /\* changes here \*/} to create a new object rather than mutating the existing one. This follows the principle of immutability, which I aim to adhere to strictly, even if it might not always seem strictly necessary.

Modifying lists: add, update, delete

This pattern makes operations like adding, updating, and deleting items in a list straightforward. For the sake of clarity, I'll omit the server calls for now; just remember that the subjects are defined as properties of the component before the constructor runs.

/***  add example ***/
// add
public addSubj = new Subject<IPerson>();

// don't forget to add addPerson$ to the merge operator
const addPerson$ = this.addSubj.pipe(
  // spread operator is used on the existing persons list
  // to add the new person
  map( newPerson => (vm:IPersonVm) => ({
    ...vm,
    persons:[...vm.persons, newPerson ]
  }))
);

/*** delete example ***/
public deleteSubj = new Subject<IPerson>();

const deletePerson$ = this.deleteSubj.pipe(
  map( personToDelete => (vm:IPersonVm)=>({
    ...vm,
    persons:vm.persons.filter(p=>p!==personToDelete )
  }))
);

/*** update example ***/
public updateSubj = new Subject<IPerson>();

const updatePerson$ = this.updateSubj.pipe(
  map( personToUpdate => (vm:IPersonVm)=>{
    const indexOfPerson = vm.persons.findIndex(p=>p===personToUpdate);
    // spread operator to maintain immutability of the persons array
    const persons = [
      ...vm.persons.slice(0,indexOfPerson),
      personToUpdate,
      ...vm.persons.slice(indexOfPerson+ 1)
    ];
    return {...vm, persons};
  })
);

Performing server updates

This example illustrates how to push an update to a backend server:

public updateSubj = new Subject<IPerson>();

const updatePerson$ = this.updateSubj.pipe(
  mergeMap( personToUpdate => this.personService.update(personToUpdate) ),
  map( updatedPerson => {
    // this time we can not use the object equality,
    // because it will be a new object deserialized
    // from json of update response. In this case
    // I assume a person has an unqiue identifier called **id**
    const indexOfPerson = vm.persons.findIndex(p=>p.id === updatedPerson.id );
    const persons = [
      ...vm.persons.slice(0,indexOfPerson),
      personToUpdate,
      ...vm.persons.slice(indexOfPerson+ 1)
    ];
    return {...vm, persons};
  })
)

Handling routing parameters

It's common to rely on Angular's routing paramMap or queryParamMap for the initial viewmodel state. As these are observables, they can be seamlessly converted into a viewmodel mutation function and combined into the main stream.

 constructor(private route:ActivatedRoute) {
    const retrieveData$ = route.paramMap.pipe(
      map( paramMap => +this.paramMap.get('id') ),
      switchMap( id => this.personService.getPerson(id)),
      map( personDetail => (vm:IPersonVm)=> ({...vm, personDetail }))
    )
  };

  this.vm$ = merge(retrieveData$, /* other viewmodel mutations */);

Executing side effects

Should you need to perform side effects without altering the viewmodel, you can map an observable to a mutation function that executes the effect and then returns the viewmodel's current state.

const sideEffect$ = sideEffectSubj.pipe(
  map((_) => (vm: IViewModel) => {
    // execute side effect here
    return vm;
  })
);

Handling shared observables

Consider a scenario where a list must be reloaded under two conditions:

when the user explicitly clicks a reload button.
after an item is successfully deleted from the server.

The initial setup might look like this:

public reloadSubj = new Subject<boolean>();
public deleteSubj = new Subject<IItem>();

private delete$ = this.deleteSubj.pipe(
  mergeMap( item => this.itemService.delete(item.Id)),
  map( _ => (vm:IItemListVm) => vm)
);

private reload$ = this.reloadSubj.pipe(
  switchMap( _ => this.itemService.getItems() ),
  map( items => (vm:IItemListVm) => ({...vm, items}))
)

constructor() {
  vm$ = merge(this.reload$, this.delete$).pipe(
    scan( ... )
  )
}

To trigger a reload upon successful deletion, you can combine the observables:

private reload$ = merge(this.reloadSubj, this.delete$).pipe(
  swicthMap( _ => this.itemService.getItems() ),
  map( items => (vm:IItemListVm)=>({...vm, items}))
);

This introduces a problem: the delete$ observable is now subscribed to twice—once directly by merge and once indirectly via reload$. Consequently, the delete operation would execute twice. The share operator provides a simple fix for this duplication.

private delete$ = this.deleteSubj.pipe(
  mergeMap( item => this.itemService.delete(item.Id)),
  share(),
  map( _ => (vm:IItemListVm) => vm)
);

With share, an observable result is shared among future subscribers so that they don't receive past emissions. By using it on the delete$ observable, you prevent the server request from being repeated multiple times, despite the multiple subscriptions.

In contrast, shareReplay is better suited as a caching mechanism. It replays a specified number of previous emissions (determined by the bufferSize) to any new subscriber. This is useful when an observable fetches data from the server and you want multiple components to access that data without triggering a new fetch.

Implementing client-side filtering

One might be tempted to view filtering as another simple mutation, like this:

interface IPersonVm {
    persons:IPerson[];
  }

  class PersonListComponent {
    public vm$ : Observable<IPersonVm>;
    public filterSubj : BehaviorSubj<string>(null);

    constructor(private dataService:DataService) {
      this.vm$ = merge(this.retrievePerson$, this.filterPersons$).pipe(
        scan( (vm:IPersonVm, mutationFn:(vm:IPersonVm)=>IPersonVm)
            => mutationFn(vm), {persons:[], personDetail:null}
          )
      );
    }
    private retrievePersons$ = this.dataService.getPersons().pipe(
      map( persons => ({...vm, persons }) )
    );
    // attempt filtering as another mutation on the viewmodel
    private filterPersons$ = this.filterSubj.pipe(
      map( filterArg => ({
        ...vm,
        persons:persons.filter(p=>filterArg==null || p.name.includes(filterArg))
      }) )
    )
  }

This approach, however, fails on the second filter change. At that point, vm.persons is already a filtered subset, when what you actually want to filter is the original dataset. Thus, filtering must be kept outside the normal scan cycle that builds the viewmodel.

To address this, we apply the filter to the output of the viewmodel's mutation observable.

 constructor(private dataService) {

    const unfilteredVm$ =  merge(this.retrievePerson$).pipe(
        scan( (vm:IPersonVm, mutationFn:(vm:IPersonVm)=>IPersonVm)
            => mutationFn(vm), {persons:[], personDetail:null}
          )
    );

    // not anymore part of the viewmodel mutation scan cycle
    this.vm$ = combineLatest(unfilteredVm$, this.filterSubj).pipe(
      map( ([vm, filterArg]) => ({
          ...vm,
          persons:vm.persons.filter(filterArg==null || p.name.includes(filterArg))
      }))
    )

  }

This approach works, but it filters on every change, even those unrelated to the filter term. To avoid unnecessary re-filtering, we can refine the logic to only run the filter when filterSubj actually changes. Instead of storing a previous filter value on the component, we can use the scan operator again to track this state observably.

 this.vm$ = combineLatest(unfilterdVm$, this.filterSubj).pipe(
    scan( ([prevVm, prevFilterArg],[nextVm, nextFilterArg]) => {
      const shouldFilter = prevFilterArg!=nextFilterArg;
      const persons = shouldFilter
        ? nextVm.persons.filter(p=>p.name.includes(nextFilterArg))
        : prevVm.persons
      return [{...nextVm,persons},nextFilterArg];
    }),
    map( [vm,_] => vm)
  )

First, combineLatest emits a tuple [vm: IVm, filterArg: string] whenever either unfilteredVm$ or filterSubj emits a new value. This gives us both the new viewmodel and the current filter value.
We need the previous viewmodel (prevVm) because if the filter hasn't changed, we want to keep the already filtered list from before, rather than potentially re-filtering a new unfiltered list.
We need the previous filter value (prevFilter) to compare against the new one (filterArg) to determine if filtering is necessary.

The arguments to the scan function, using array destructuring, might look a bit unusual, but it's essentially just a special case of destructuring. In its undestructured form, the function would look like this:

 scan( (prev, next) => {
   const prevVm = prev[0];
   const prevFilterArg = prev[1];
   const nextVm = next[0];
   const nextFilterArg = next[1];
   ...
 })

Feel free to use whichever style you prefer for readability.

When the filter is deemed to have changed, we apply it to nextVm.persons, which holds the latest unfiltered data. If the filter remains unchanged, we reuse prevVm.persons, which is the list that was already filtered by the previous run.

The scan function must then output a tuple to be used by the next iteration, containing both the new viewmodel state and the latest filter value.

Finally, because the view expects an IPersonVm object, not a tuple, we apply a subsequent map operation to extract just the viewmodel.

Note that this is a simplified scenario. For instance, if list mutations (add, update, delete) also occur, the filter logic must be reapplied after those actions. The condition for when to filter can be generalized:

const shouldFilter =
  prevFilterArg != nextFilterArg || prevVm.persons !== nextVm.persons;

Debugging considerations

This approach makes debugging trickier. Setting breakpoints within an observable pipeline is more challenging. You'll often need to rely on console.log statements or the rxjs tap operator to inspect values and trace the application's flow.

Concluding remarks

Comparing this method to conventional state management approaches, the trade-offs are clear:

It generally involves writing more boilerplate code, especially for simpler components.
It doesn't appear to have limitations in terms of what can be accomplished. So far, it has handled every scenario I've come across.
It's not a beginner-friendly technique; a solid understanding of RxJS is essential.

Key observable patterns

This article has relied on a toolkit of observable operators and patterns:

merge to build the combined viewmodel observable from various sources.
scan for both accumulating viewmodel mutations and optimizing filter logic.
subjects to represent and emit user actions from the view.
mergeMap to map to a server request for mutation operations.
switchMap for server fetches that should be cancellable.
map to transform any stream, typically into a viewmodel mutation function.
share to prevent duplicate executions when an observable has multiple subscribers.
shareReplay to cache and share data across multiple components.

Other operators that are also useful but not detailed here include:

zip to combine, for example, a subject's current value with the result of a piped operation.
forkJoin to parallelize and await multiple HTTP requests.

Choosing a subject type

The choice between Subject, BehaviorSubject, and ReplaySubject depends on your needs for state and timing:

A plain Subject has no initial or current value; any late subscribers miss emissions that occurred before they subscribed.
A ReplaySubject doesn't require an initial value, but it retains and replays recent values to new subscribers.
A BehaviorSubject requires an initial value and automatically repays the current value to any new subscriber upon subscription.

Rekna

Writes about RxJS. Active 2020.

All 1 article →

← NewerWebpack 5 Module Federation: A game-changer in JavaScript architecture Older →Angular: Keeping it Fat, Dumb, and Happy