Introducing the auto-signal pattern

YouTube

Signals are perfect for synchronous reactivity, and observables are perfect for asynchronous reactivity. They are complimentary, but also at odds with each other. In this article I will share a neat trick to get them to play well with each other.

The Problem

toSignal(source$) immediately subscribes to the source$ observable, and stays subscribed until it is destroyed. This is necessary because signals are expected to always contain a current value that can be read from at any time. However, it blocks the ability of RxJS streams to automatically

1. clean up
2. reset and re-fetch
3. cancel requests
4. defer work and requests

1. Automatic cleanup

Listening to data sources can be expensive. Firebase has different price levels depending on the number of simultaneous connections, and other data sources can cause memory leaks if not cleaned up. Apps that don't use RxJS effectively need manual cleanup code, which can easily be neglected in some scenarios.

If you need an event source on only some pages, observables can automatically clean up the connections or listeners when the user goes to other pages and all subscribers have unsubscribed.

But if you call toSignal in a shared service, the source observable will never clean up resources.

2. Reset and re-fetch

When you leave a page and come back later, seeing stale data there can be deceptive or even jarring. Apps that don't use RxJS effectively will have extra code that runs inside an OnDestroy to manually clear stale data. They also have to manually trigger a re-fetch.

Observables can automatically reset state when the user goes to other pages and all subscribers have unsubscribed, and automatically re-fetch data when the user returns to a page that needs it.

But if you call toSignal in a shared service, the source observable will never reset its state or re-fetch data:

3. Automatic cancel

When you leave a page, it becomes pointless and wasteful to continue fetching unnecessary data or creating expensive data sources. Without RxJS, it is up to developers to notice when certain data sources are large enough, or typical user connection speed is slow enough, that it noticeably deteriorates the user experience.

Observables can automatically cancel requests when the user goes to other pages and all subscribers have unsubscribed.

But if you call toSignal in a shared service, the source observable will continue on with unnecessary requests and download unnecessary data

4. Automatic defer

It is pointless and wasteful to fetch data or create expensive connections before they are necessary.

Observables can represent data sources but wait for subscribers instead of prematurely consuming them.

But if you call toSignal in a shared service, the source observable will consume the data source it represents as soon as that service is created:

Don't share signals?

If you want to share synchronously derived state, you either can't use signals, or you have to sacrifice automatic cleanup, resetting, refetching, cancellation and deferment. The more complex your application is, the more painful this tradeoff will be.

This is why, despite the huge benefits of signals, I have advised to never use signals in a shared service. It sucks to not have efficient derived state, but usually it sucks more to have lower code quality because RxJS can't automatically manage data dependencies.

Note: I'm really not asking for anything excessively luxurious here. These benefits are commonly achieved in React with hooks like useQuery. I believe Angular can have nice things too, if people want them.

Real-world example

One company I worked for had 2-3 routes using realtime data, but the users spent the majority of the time away from those routes, so it was important to close unused connections. The most complex feature required this series of asynchronous data dependencies:

1. Get Firebase token from server
2. Create Firebase connection
3. Listen to message data from Firebase
4. Use message data to fetch more details from server

RxJS automatically closed unused Firebase connections for us. We never worried about orchestrating when sibling routes might need the Firebase connection or not. We had a service providedIn: 'root' and the Firebase connections automatically closed when they were not needed by any component.

Realtime data is awesome for users, and RxJS makes it easy for developers. Once you get used to automatic cleanup, resetting, refetching, cancellation and deferment, you will never want to code any other way, even in simple apps. It is just too convenient to not have to use imperative duct tape everywhere.

The Solution

Rather than converting an observable to a signal with toSignal, we can keep the observables and signals separate, but build a connection between them that lasts exactly as long as we need it.

Let's say we have an observable of asynchronous data:

interval$ = interval(3000);

And we have a signal with synchronously derived state:

count = signal(0);
double = computed(() => this.count() * 2);

What we can do is make a function that creates a connection between interval$ and count, and we can represent this connection as an observable. When we subscribe to this observable, it will subscribe to interval$. When interval$ emits it will call count.set(value) with the value from the interval. It should do this only once, no matter how many subscribers it has, so we'll add a share() at the end. And we want it to reset to the initial state after all subscribers are gone, so we'll use a finalize for that. Also, we don't want self-completing observables to trigger this behavior, so we have to merge interval$ with NEVER to prevent the finalize from running too soon. Now this connection observable will only complete once all subscribers unsubscribe. Here's the implementation of the function that can create this connection observable:

import { WritableSignal } from '@angular/core';
import { Observable, tap, share, finalize, merge, NEVER } from 'rxjs';

/**
 * https://dev.to/mfp22/introducing-the-auto-signal-pattern-1a5h 
*/
export function connectSource<State>(
  state: WritableSignal<State>,
  source$: Observable<State>,
) {
  const initialState = state();
  return merge(source$, NEVER).pipe(
    tap(s => state.set(s)),
    finalize(() => state.set(initialState)),
    share(),
  );
}

And here's how to use it:

@Injectable({ providedIn: 'root' })
export class CountService {
  count = signal(0);
  double = computed(() => this.count() * 2);

  interval$ = interval(3000);

  connection$ = connectSource(this.count, this.interval$);
}

Now, whenever we want to use these count or double signals in a component, we need to make sure that connection$ has a subscription. We can create a function that does this automatically, by injecting a service, looking for a connection$ property and subscribing to it with takeUntilDestroy:

import { DestroyRef, inject, ProviderToken } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { Observable } from 'rxjs';

/**
 * https://dev.to/mfp22/introducing-the-auto-signal-pattern-1a5h 
*/
export function injectAutoSignal<
  T,
  Service extends { connection$: Observable<T> },
>(token: ProviderToken<Service>) {
  const service = inject(token);
  service.connection$.pipe(takeUntilDestroyed()).subscribe();
  return service;
}

And here's how to use that function:

@Component({
  // ...
  template: `
    <h1>Count is {{countService.count()}}</h1>
    <h1>Double is {{countService.double()}}</h1>
  `
})
export class CountComponent {
  countService = injectAutoSignal(CountService);
}

And we get almost everything we want: Derived state with signals, automatic cleanup, resetting, refetching and cancellation!

The one missing feature is automatic deferment. Our source observable is not completely deferred, because the component subscribes as soon as it is created and injects the service, rather than waiting for an actual subscription in the template. However, this is good enough for the vast majority of use cases.

If we want to use our derived state in another service, we just have to make sure we copy the connection$ observable to the new service so components can subscribe through the new service:

@Injectable({ providedIn: 'root' })
export class AnotherService {
  countService = inject(CountService);

  message = computed(() => `Count is ${this.countService.count()}`);

  connection$ = this.countService.connection$;
}

Definitely don't inject CountService using injectAutoSignal, because that would immediately subscribe to connection$, which would behave just like toSignal. Just use inject inside services.

You can also combine 2 connections using merge:

@Injectable({ providedIn: 'root' })

export class CountService1 extends CountService {}

@Injectable({ providedIn: 'root' })

export class CountService2 extends CountService {}

@Injectable({ providedIn: 'root' })

export class CombinedService {

  countService1 = inject(CountService1);

  countService2 = inject(CountService1);

combinedCount = computed(

    () => this.countService1.count() + this.countService2.count(),

  );

connection$ = merge(

    this.countService1.connection$,

    this.countService2.connection$,

  );

}

Conclusion

It looks like we have everything we want! We get the best of both worlds: Efficient synchronous reactivity with signals, and on-demand asynchronous reactivity from RxJS!

Hopefully this helps the Angular community write cleaner and safer code.

Please share this article with other Angular developers.