DEV Community

Cover image for Create Your Own TanStackQuery In Angular 🤝 RxJs
Romain Geffrault
Romain Geffrault

Posted on

Create Your Own TanStackQuery In Angular 🤝 RxJs

#angular #typescript #rxjs

I have a project that is stuck on Angular v13 and cannot upgrade to newer versions, but I needed to implement a declarative server state tool that greatly simplifies the developer experience.

In this series of articles, I will share a complete declarative server state tool built entirely with RxJS and compatible with older Angular versions.

When talking about server state management, we often think about TanStack Query. We will not implement all of its functionalities, but you will have the right fundamentals to build them.

I do not adopt the same philosophy as TanStack Query; my utilities react declaratively while preserving type safety.

I worked extensively on my personal state management tool, which includes server state management, using only Signals.

I learned a lot about the problem and some declarative ways to manage it.

So I decided to create an RxJS tool that you can easily use in your project.

First, I'll introduce the fundamental building blocks: resource$ and resourceByGroup$.

Then, in subsequent articles, I'll cover more advanced utilities: query$, mutation$, and asyncMethod.

What resource$ will enable for server state management?

The Angular official resources are mainly used to track async data fetching.

It is pretty straightforward to implement that kind of behavior with RxJS.

But I chose to add more advanced features that are not exposed in Angular resources, because we will use it as our custom implementation to create a server state management tool.

Here is what my resource$ can do:

  • track async request status
  • catch errors
  • accept a local value
  • reload on demand

RxJS offers some behaviors that are relatively easy to implement and very convenient when dealing with server state:

  • repeat the stream when completed (more or less the equivalent of a reload)

  • declarative server state reactions are easier to write, which enables:

    • optimistic update / update ...
    • fallback on error...

One more thing: our resource$ utility can also be used for mutations (POST/PUT/PATCH/DELETE). However, I'll still provide a mutation$ implementation that makes your code more concise and composable.

resource$ usage and implementation

Here is my implementation, which also preserves the latest value by default.

This means that when a new request is made, the previous value remains available, which avoids screen flickering.

Usage:

  private readonly destroy$ = new Subject();
  ngOnDestroy() {
    this.destroy$.next();
    this.destroy$.complete();
  }


  protected readonly myResource$ = resource$({
    params$: this.id$,
    loader: ({ params: id }) =>
      this.checksService.getCheck$(id),
    destroy$: this.destroy$
  });

  // in the template:
  <ng-container *ngIf="myResource$ | async as myResource">
   hasValue: {{myResource.hasValue}}
   isLoading: {{myResource.isLoading}}
   status: {{myResource.status}}
   hasError: {{myResource.hasError}}
   error: {{myResource.error}}
   value: {{myResource.value}}
  </ng-container>

  // in ts logic, outside observable stream
  this.myResource$.snapshot();
Enter fullscreen mode Exit fullscreen mode

Code:

// polyfill to prevent type inference in generic types
export type NoInfer<T> = [T][T extends any ? 0 : never];

export function resource$<ResourceState, Params>({
  params$,
  loader,
  initialValue,
  initialParams,
  setValue$,
  destroy$,
  reload$,
}: {
  params$: Observable<Params>;
  loader: (param: { params: Params }) => Observable<ResourceState>;
  initialValue?: NoInfer<ResourceState>;
  initialParams?: NoInfer<Params>;
  setValue$?: (
    resourceSnapshot: () => ResourceLike<ResourceState, Params>
  ) => Observable<NoInfer<ResourceState> | undefined>;
  reload$?: Observable<unknown>;
  destroy$: Observable<unknown>;
}) {
  let resourceSnapshot: undefined | ResourceLike<ResourceState, Params> =
    undefined;
  let pramasSnapshot: undefined | Params = initialParams;
  const dataFromLoader$ = merge(
    params$,
    reload$?.pipe(
      filter((params) => !!params),
      map(() => pramasSnapshot as Params)
    ) ?? EMPTY
  ).pipe(
    tap((params) => (pramasSnapshot = params)),
    switchMap((params) =>
      loader({ params }).pipe(
        map(
          (value) =>
            ({
              hasValue: true,
              isLoading: false,
              status: "resolved",
              hasError: false,
              error: undefined,
              value,
              params,
            } as ResourceLike<ResourceState, Params>)
        ),
        startWith({
          hasValue: !!initialValue,
          isLoading: true,
          status: "loading",
          hasError: false,
          error: undefined,
          value: initialValue,
          params,
        } as ResourceLike<ResourceState, Params>),
        catchError((error) =>
          of({
            hasValue: false,
            isLoading: false,
            status: "error",
            hasError: true,
            error,
            value: undefined,
            params,
          } as ResourceLike<ResourceState, Params>)
        )
      )
    )
  );

  const localSet$ =
    setValue$?.(
      () => resourceSnapshot as ResourceLike<ResourceState, Params>
    ).pipe(
      map(
        (value) =>
          ({
            hasValue: !!value,
            isLoading: false,
            status: "local",
            hasError: true,
            error: undefined,
            value: value,
            params: undefined,
          } as ResourceLike<ResourceState, Params>)
      )
    ) ?? EMPTY;

  const stream$ = merge(dataFromLoader$, localSet$).pipe(
    // used to preserve the last value
    scan((acc, cur) => {
      if (!cur.hasValue && !cur.hasError) {
        return {
          ...cur,
          value: acc.value,
        };
      }
      return cur;
    }, idleResource<ResourceState, Params>(initialValue, initialParams)),
    startWith(
      idleResource(initialValue, initialParams) as ResourceLike<
        ResourceState,
        Params
      >
    ),
    tap((resource) => (resourceSnapshot = resource)),
    shareReplay({
      refCount: true,
      bufferSize: 1,
    })
  );
  stream$.pipe(takeUntil(destroy$)).subscribe();
  return Object.assign(stream$, {
    snapshot: () => resourceSnapshot,
    params: () => pramasSnapshot,
  });
}

export type ResourceLike<State, Params> = {
  readonly isLoading: boolean;
  readonly hasValue: boolean;
  readonly status: ResourceStatus;
  readonly hasError: boolean;
  readonly error: any | undefined;
  readonly value: State | undefined;
  readonly params: Params | undefined;
};

export const idleResource = <ResourceState, Params>(
  initialValue?: ResourceState,
  initialParams?: Params
): ResourceLike<ResourceState, Params> => ({
  hasValue: !!initialValue,
  isLoading: false,
  status: "idle",
  hasError: false,
  error: undefined,
  value: initialValue,
  params: initialParams,
});
type ResourceStatus =
  | "idle"
  | "error"
  | "loading"
  | "reloading"
  | "resolved"
  | "local";
Enter fullscreen mode Exit fullscreen mode

1- Yes, I have no access to takeUntilDestroyed, so I have to pass a destroy$ observable in order to unsubscribe properly
2- I chose to expose a snapshot, because I want the possibility to get the current state without subscribing (it just fits well in the current codebase)
3- You may customize this implementation to fit your needs.

This implementation exposes a declarative way to reload the async request (reload$).

  • You can bind it to an interval
  • It is also possible to reload after a source$ error (check the next demo)

Before presenting more elaborate use cases, let me address some important limitations.

This implementation may change as I work on it, by adding or changing some functionalities.

Like the official resource, our utility does not handle typed errors (they are all inferred as any).

So you may improve the way API errors are handled.

It is also possible to improve the fallback part (when applying an optimistic update from a mutation that ultimately fails).

Both the official Angular Resource and this resource$ do not allow parallel requests, which is very problematic. So let's create a resourceByGroup$ that enables parallel requests.

resourceByGroup$ usage and implementation

resourceByGroup$ is used to make parallel requests. It has the same options as resource$ with one additional required option: identifier.

The identifier is a callback function that is called each time param$ emits.

Its return value is used as a key for each parallel request.

Because resourceByGroup$ will return a Map-Like object:


{
    key1: ResourceLike<...>,
    key2: ResourceLike<...>,
}
Enter fullscreen mode Exit fullscreen mode

Each time the identifier callback returns a non-existing value, it will instantiate a new corresponding resource.

Usage:

  private readonly destroy$ = new Subject();
  ngOnDestroy() {
    this.destroy$.next();
    this.destroy$.complete();
  }


  protected readonly myResourceByGroup$ = resourceByGroup$({
    params$: this.id$,
    identifier: (id) => id,
    loader: ({ params: id }) =>
      this.checksService.getCheck$(id),
    destroy$: this.destroy$
  });

  // in the template:
   <div *ngFor="let resource of myResourceByGroup$ | async | keyvalue">
    key: {{ resource.key }}
    hasValue: {{resource.value.hasValue}}
    isLoading: {{resource.value.isLoading}}
    status: {{resource.value.status}}
    hasError: {{resource.value.hasError}}
    error: {{resource.value.error}}
    value: {{resource.value.value}}}
   </div>

   // or
    <div *ngFor="let item of myList">
    //...
    <ng-container
        *ngIf="
            myResourceByGroup$.select(item.id)
            | async as resourceItem
        "
    >
     // ...
    </ng-container>
   </div>
Enter fullscreen mode Exit fullscreen mode

Code:

// polyfill to prevent type inference in generic types
export type NoInfer<T> = [T][T extends any ? 0 : never];

export type ResourceByGroupRef$<GroupIdentifier, ResourceState, Params> =
  Observable<{
    [identifier in GroupIdentifier & string]: ResourceLike<
      ResourceState,
      Params
    >;
  }> & {
    select: (
      id: GroupIdentifier
    ) => Observable<ResourceLike<ResourceState, Params>>;
    selectField: <Field extends keyof ResourceLike<ResourceState, Params>>(
      id: GroupIdentifier,
      field: Field
    ) => Observable<ResourceLike<ResourceState, Params>[Field]>;
    change$: Observable<{
      identifier: GroupIdentifier;
      resource: ResourceLike<ResourceState, Params>;
    }>;
  };

type ResourceMappingByGroup<GroupIdentifier, ResourceState, Params> = {
  resourceByGroup: {
    [identifier in GroupIdentifier & string]: ResourceLike<
      ResourceState,
      Params
    >;
  };
  lastChange$: {
    identifier: GroupIdentifier;
    resource: ResourceLike<ResourceState, Params>;
  };
};

export function resourceByGroup$<ResourceState, Params, GroupIdentifier>({
  params$,
  identifier,
  loader,
  initialValue,
  destroy$,
}: {
  params$: Observable<Params>;
  identifier: (params: NoInfer<NonNullable<Params>>) => GroupIdentifier;
  loader: (param: { params: Params }) => Observable<ResourceState>;
  initialValue?: NoInfer<ResourceState>;
  destroy$: Observable<unknown>;
}): ResourceByGroupRef$<GroupIdentifier, ResourceState, Params> {
  const resourceByGroup$ = params$.pipe(
    groupBy(
      (params) => identifier(params as NonNullable<Params>),
      (params) => params
    ),
    mergeMap((paramsByGroup$) =>
      resource$({
        params$: paramsByGroup$,
        loader,
        initialValue,
        destroy$,
      }).pipe(
        map((resource) => ({
          resource,
          identifier: paramsByGroup$.key,
        }))
      )
    ),
    scan(
      (acc, { resource, identifier }) => {
        return {
          resourceByGroup: {
            ...acc.resourceByGroup,
            [identifier as GroupIdentifier & string]: resource,
          },
          lastChange$: {
            identifier,
            resource,
          },
        };
      },
      { resourceByGroup: {}, lastChange$: {} } as ResourceMappingByGroup<
        GroupIdentifier,
        ResourceState,
        Params
      >
    ),
    startWith({
      resourceByGroup: {},
      lastChange$: {},
    } as ResourceMappingByGroup<GroupIdentifier, ResourceState, Params>),
    shareReplay({ refCount: true, bufferSize: 1 })
  );

  resourceByGroup$.pipe(takeUntil(destroy$)).subscribe();

  return Object.assign(
    resourceByGroup$.pipe(map(({ resourceByGroup }) => resourceByGroup)),
    {
      select: (id: GroupIdentifier) =>
        resourceByGroup$.pipe(
          map(({ resourceByGroup }) => {
            const result =
              resourceByGroup[id as GroupIdentifier & string] ||
              idleResource(initialValue);
            return result;
          })
        ),
      selectField: (
        id: GroupIdentifier,
        field: keyof ResourceLike<ResourceState, Params>
      ) =>
        resourceByGroup$.pipe(
          map(({ resourceByGroup }) => {
            return (
              resourceByGroup[id as GroupIdentifier & string]?.[field] ||
              idleResource(initialValue)
            );
          })
        ),
      change$: resourceByGroup$.pipe(
        map(({ lastChange$ }) => lastChange$),
        filter((data) => !!data.resource)
      ),
    }
  ) as ResourceByGroupRef$<GroupIdentifier, ResourceState, Params>;
}
Enter fullscreen mode Exit fullscreen mode

Since I cannot use @let from Angular Control Flow, I added some utilities that I can use in my templates.
Some of them will also be used to simplify the upcoming server state utility functions.

Demo resource$ and resourceByGroup$

Here, I created a small example that displays a list of users.

  • You can update the user name optimistically
  • You can trigger an update error that will force the users list to refresh

Demo resource$ and resourceByGroup$

In the next article, I will share some StackBlitz demos that will help you understand how to use it.

resource$ and resourceByGroup$ can be used like that. The next utilities (query$, mutation$, and asyncMethod) will be used to improve the DX.

If you like this article, please leave a comment or a like. I don't have that much feedback yet.

A small overview of my state management tool

I would like to show you my experimental state management tool that handles server state.

ng-craft states management tool - Romain Geffrault

It will do almost the same as the previous example, except that it also handles query params state.

This tool depends only on Signals and RxJS is fully optional.

I wrote other articles that you might like:

Follow me on LinkedIn for an Angular Advent calendar.
👉 Romain Geffrault

Top comments (0)