How to build reactive Angular Components using Inputs as Observables

TL;DR

• Angular is integrated with RxJS, but not everywhere. Mixing imperative and reactive implementations might lead to bugs or unexpected behaviour.
• TypeScript decorators are powerful tools to extend the capabilities of our code.
• Redefining inputs as observables can improve your component's structure for better readability, robustness and flow control.
• This can be easily implemented via a decorator, which is presented in this blog post in two variants (@Observe & @AsObservable). The final implementation of these and other decorators is available in ngx-propserve.

Motivation

Angular is one of the most popular frameworks for building web applications. RxJS, a reactive extensions library, is heavily integrated within the framework to simplify asynchronous calls and callback-based logic, among others. However, not all the parts of the framework are integrated equally.

The API of an Angular component can be specified by using @Input and @Output decorators. On the one hand, Output properties make use of the EventEmitter class to emit events. EventEmitter is a generic class that extends Subject, an RxJS primitive, to multicast values to all its subscribers. On the other hand, Input properties are de-attached from RxJS, and Angular has well-defined lifecycle hooks to react to any changes in the component.

@Component({
  selector: "my-component",
  template: `<p>prop: {{ prop }}</p>`,
})
export class MyComponent implements OnChanges {
  @Input() prop!: number;
  @Output() propChange = new EventEmitter<number>();

  ngOnChanges(changes: SimpleChanges) {
    // I can react to changes of `prop`
  }
}

Our component above has an Input property called prop, which is rendered in the template. The OnChanges hook notifies us about value changes by triggering the ngOnChanges method.

Now let's imagine that we need to make an asynchronous operation whenever there are new values of prop. For example, an HTTP call to a known REST API:

@Component({
  selector: "my-component",
  template: `<p>result: {{ result$ | async }}</p>`,
})
export class MyComponent implements OnChanges {
  @Input() prop!: number;

  result$?: Observable<string>;

  constructor(private myService: MyService) {}

  ngOnChanges(changes: SimpleChanges) {
    if (changes["prop"]) {
      this.result$ = this.myService.getResult(this.prop);
    }
  }
}

As you can see in the template, the async pipe subscribes to the Observable variable result$ and returns its value so we can render it. Additionally, the async pipe handles as well result$ being undefined, new Observable instances whenever a change occurs, unsubscribing when the component is destroyed, among others. The async pipe is a very powerful piece of software! 🚀

For the second iteration, let's imagine we need to subscribe to our result$ internally. For example, we need to store an intermediate value:

@Component({
  selector: "my-component",
  template: `<p>result: {{ result$ | async }}</p>`,
})
export class MyComponent implements OnInit, OnChanges {
  @Input() prop!: number;

  result$?: Observable<string>;
  intermediateResult?: string;

  constructor(private myService: MyService) {}

  ngOnInit() {
    // unsubscribing not implemented for demonstration purposes
    this.result$?.subscribe((result) => {
      this.intermediateResult = transformResult(result);
    });
  }

  ngOnChanges(changes: SimpleChanges) {
    if (changes["prop"]) {
      this.result$ = this.myService.getResult(this.prop);
    }
  }
}

The main issue of this implementation is that our subscription to result$ won't automatically re-subscribe to new Observable instances, and therefore, intermediateResult won't get any new values. There are other minor issues, e.g., handling the initial undefined value of result$ and the need to implement custom debouncing in case prop changes very rapidly.

This type of error could be solved by integrating changes to Input properties with reactive observables. There is a long-running issue in the Angular Github repository regarding this topic. The Angular team has expressed the idea of investing in de-attaching RxJS entirely from the framework and providing an independent library to integrate both ecosystems. This decision would create an opportunity for improving the RxJS integration beyond the current state, but we'll see what the future holds!

A stronger integration with Input properties would allow us to organise all the component's streams in a more reactive fashion, like following the SIP Principle. Let's explore how we achieved this at LeanIX by implementing a new decorator.

React to changes with decorators

TypeScript decorators can be attached to a class declaration, method, accessor, property or parameter to extend their capabilities with extra functionality. Decorators are not foreign to Angular (@Input & @Output), so this concept is already well integrated into the ecosystem.

Observe

Implementation

The main goal is to be able to define component inputs as observables and be able to react to their changes by using RxJS. We will encapsulate this logic in a PropertyDecorator, so that it can be re-used across different components and has the same feel as the familiar @Input & @Output.

A PropertyDecorator is no more than a function with two parameters:

1. The prototype of the property's class.
2. The name of the property that the decorator is attached to.

Our @Observe decorator should listen for changes in a property and propagate such changes to a new Observable variable. The final implementation is as follows:

export function Observe<T>(observedKey: string): PropertyDecorator {
  // `target` defines the target class prototype that the property decorator
  // is attached to.
  return (target: any, key: string | symbol): void => {
    // Declare all the active subjects for a given target class instance.
    const subjects = new WeakMap<any, BehaviorSubject<T | undefined>>();

    // Return the associated subject for a given target class instance.
    // In case none is available yet, create one.
    const getSubject = (
      instance: any
    ): BehaviorSubject<T | undefined> | undefined => {
      if (!subjects.has(instance)) {
        subjects.set(instance, new BehaviorSubject<T | undefined>(undefined));
      }
      return subjects.get(instance);
    };

    // Transform the decorated property into an `Observable` that propagates
    // the changes of the internal subject.
    Object.defineProperty(target, key, {
      get(): Observable<T | undefined> | undefined {
        // `this` is the current instance of the class
        return getSubject(this);
      },
    });

    // Transform the definition of the observed property so that we can propagate
    // its value changes to the internal subject.
    Object.defineProperty(target, observedKey, {
      get(): T | undefined {
        return getSubject(this)?.getValue();
      },
      set(instanceNewValue: T): void {
        getSubject(this)?.next(instanceNewValue);
      },
    });
  };
}

Well, that's a lot to digest. Let's try to break this down:

The Observe decorator expects as an argument the key of the property that we want to observe.

export function Observe<T>(observedKey: string): PropertyDecorator;

Since we can have multiple class instances of the same component, we need to differentiate between those, so we don't mix up value changes coming from different instances. Therefore, a key-value map will keep track of all instances and orchestrate the associated listeners. By using a WeakMap, we ensure that whenever a component instance has been destroyed, its associated entry in the Observe decorator is also removed by the garbage collector.

const subjects = new WeakMap<any, BehaviorSubject<T | undefined>>();

Then, for the observed property, we change its definition to propagate value changes to the internal subject. The class instance is referenced inside defineProperty with this.

Object.defineProperty(target, observedKey, {
  get(): T | undefined {
    return getSubject(this)?.getValue();
  },
  set(instanceNewValue: T): void {
    getSubject(this)?.next(instanceNewValue);
  },
});

Finally, we can return the desired Observable stream from the internal subject and integrate it into our component.

Object.defineProperty(target, key, {
  get(): Observable<T | undefined> | undefined {
    return getSubject(this);
  },
});

Usage

Let's take a look at how this can be used in our component:

@Component({
  selector: "my-component",
  template: `<p>result: {{ result$ | async }}</p>`,
})
export class MyComponent {
  @Input() prop!: number;

  @Observe("prop") private prop$!: Observable<number>;

  result$ = this.prop$.pipe(
    switchMap((prop) => this.myService.getResult(prop)),
    // share the response across all subscribers to prevent
    // multiple HTTP requests
    share()
  );

  intermediateResult?: string;

  constructor(private myService: MyService) {
    // unsubscribing not implemented for demonstration purposes
    this.result$.subscribe((result) => {
      this.intermediateResult = transformResult(result);
    });
  }
}

Whenever prop changes, the prop$ stream propagates the new values to all the subscribers. Now we can also use all the RxJS operators to combine, transform and control the flow of the different streams as we'd like (e.g. applying a debounce time to prop changes or merging and combining value changes from different properties).

Also, @Observe allows us to better structure our stream by following the SIP principle, as well as replacing imperative code (ngOnChanges) with a more declarative and reactive flavour, as shown in the following example:

@Component({
  selector: "starship",
  template: `
    <p *ngFor="let result of results$ | async">result: {{ result }}</p>
  `,
})
export class StarshipComponent {
  @Input() term!: number;
  @Input() passengersCount!: number;

  // Source
  @Observe("term") private term$!: Observable<string>;
  @Observe("passengersCount") private passengersCount$!: Observable<number>;

  // Intermediate
  allResults$ = term$.pipe(
    debounceTime(200),
    switchMap(createQuery),
    switchMap(fetchResults)
  );

  // Presentational
  results$ = combineLatest(allResults$, passengersCount$).pipe(
    map(([allResults, passengersCount]) =>
      filterResults(allResults, passengersCount)
    )
  );
}

And magically, whatever changes occur in our input properties, they will be propagated to the UI ✨

AsObservable

Implementation

Following the same principle, we could implement another decorator that does not require us to define an extra property.

export function AsObservable<T>(): PropertyDecorator {
  return (target: any, key: string | symbol): void => {
    // Declare all the active subjects for a given target class instance
    const subjects = new WeakMap<any, BehaviorSubject<T | undefined>>();

    // Return the associated subject for a given target class instance.
    // In case none is available yet, create one.
    const getSubject = (
      instance: any
    ): BehaviorSubject<T | undefined> | undefined => {
      if (!subjects.has(instance)) {
        subjects.set(instance, new BehaviorSubject<T | undefined>(undefined));
      }
      return subjects.get(instance);
    };

    // Transform the property definition so that we can propagate the value
    // changes to the internal subject, and return its associated observable.
    Object.defineProperty(target, key, {
      get(): Observable<T | undefined> | undefined {
        return getSubject(this);
      },
      set(instanceNewValue: T) {
        getSubject(this)?.next(instanceNewValue);
      },
    });
  };
}

There are a lot of similarities with our previous implementation of @Observe. The only difference is that we would mutate the attached property during runtime and thus implicitly change its type definition.

Usage

Migrating the previous starship component to @AsObservable results into the following:

@Component({
  selector: "starship",
  template: `
    <p *ngFor="let result of results$ | async">result: {{ result }}</p>
  `,
})
export class StarshipComponent {
  // Source
  @Input("term")
  @AsObservable()
  term$!: Observable<string>;

  @Input("passengersCount")
  @AsObservable()
  passengersCount$!: Observable<number>;

  // Intermediate
  allResults$ = term$.pipe(
    debounceTime(200),
    switchMap(createQuery),
    switchMap(fetchResults)
  );

  // Presentational
  results$ = combineLatest(allResults$, passengersCount$).pipe(
    map(([allResults, passengersCount]) =>
      filterResults(allResults, passengersCount)
    )
  );
}

Now we have merged the Input declaration and the source stream into one single property, which reduces the code further. However, note that if we want to follow the $ convention for observables, we would need to make use of an alias, which is discouraged. Also, the Angular compiler might complain about the type being implicitly changed by the decorator.

Adoption

The @Observe decorator was introduced in our main monorepo around two years ago. Since then, other web applications across the engineering organization have adopted it. Currently, around 200 components across different domains, libraries and applications use it to organise their Input value changes.

In our codebase, we prefer @Observe over @AsObservable, since it de-attaches the actual Input definition from its associated source stream, making the transformation from input value to input stream more explicit and thus easier to follow.

Conclusions

• In this blog post, we've explored the use of TypeScript decorators to extend Angular's limitations regarding reactive Input properties.
• By defining a custom decorator, we can declare component Input properties as source streams and use the powerful capabilities of RxJS to implement components in a much more organised and reactive way.
• We've devised two implementations: @Observe & @AsObservable.
• @Observe has been broadly adopted across multiple teams, and it helps developers to improve their components' readability, maintainability and robustness.
• Have a look at ngx-propserve, a small repository where I've collected these and other variants.

Published by Aleix Casanovas Marques

Visit author page