Resource API Refactor

Now let’s take a look at how we might improve this application with resource and linkedSignal.

The interesting thing about this refactor is that we aren’t going to refactor to resource or rxResource everywhere in the application. We are going to hit scenarios that show the current limitations of resource.

This refactor is also much more straight forward than the refactor we did for Giflist. For this refactor I will help get things started, and then I will leave the rest of the refactor for you to do as an exercise. Remember that if you get stuck the source code is available.

The MessageService

Let’s start with the service we won’t be refactoring, and why. The reason we won’t be using resource or rxResource to load our messages is this:

  messages$ = this.getMessages().pipe(
    // restart stream when user reauthenticates
    retry({
      delay: () => this.authUser$.pipe(filter((user) => !!user)),
    }),
  );

Although the retry is a critical part for us as well, that’s not the specific reason why we can’t use rxResource here.

Our getMessages gives us a stream of messages coming from our backend. Whenever there is a new message, this emits.

The problem with rxResource is that, even though it allows us to use a stream to pull in data, it only actually uses the first value from that stream. We would get an initial set of messages, but they would never update.

We can see this if we inspect the internals of the rxResouce implementation:

export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T> {
  opts?.injector || assertInInjectionContext(rxResource);
  return resource<T, R>({
    ...opts,
    loader: (params) => {
      const cancelled = new Subject<void>();
      params.abortSignal.addEventListener('abort', () => cancelled.next());

      // Note: this is identical to `firstValueFrom` which we can't use,
      // because at the time of writing, `core` still supports rxjs 6.x.
      return new Promise<T>((resolve, reject) => {
        opts
          .loader(params)
          .pipe(take(1), takeUntil(cancelled))
          .subscribe({
            next: resolve,
            error: reject,
            complete: () => reject(new Error('Resource completed before producing a value')),
          });
      });
    },
  });
}

This is actually the entire implementation of rxResource, and you may notice that it is basically just a resource with some extra stuff set up automatically for us.

It will use the stream we pass in and basically convert it into a Promise that only returns the first value, we can see that here:

.pipe(take(1), takeUntil(cancelled))

It takes one value and unsubscribes from the stream, because it immediately resolves the promise with that first value: