Lesson 7

Async with rxMethod

The Core Idea

Mental model: rxMethod is a long-lived RxJS pipeline that lives inside your store. You define it once with operators (debounce, switchMap, retry, etc.) and then call it with a signal, an observable, or a static value. When called with a signal, it automatically re-runs the pipeline whenever the signal changes - like a classic NgRx effect that reacts to state, but without actions or the actions/effects/reducer indirection.

Why Not Just async/await?

In Lesson 5 and Lesson 6, store methods used async/await with firstValueFrom. That works for fire-and-forget operations (load data, update state). But it falls short when you need:

Reactive re-execution - re-run the pipeline when a signal changes (e.g. search term, selected filter, route param)
Cancellation semantics - cancel the previous HTTP request when a new value arrives (switchMap)
Debouncing / throttling - wait for the user to stop typing before making a request
Retry / error recovery - retry failed requests without killing the subscription
Combining streams - wait for multiple async sources before proceeding

These are all things RxJS excels at. rxMethod lets you use the full RxJS operator toolkit inside a store method, while keeping the signal-based API on the outside.

Installation

npm install @ngrx/signals @ngrx/operators

rxMethod ships in @ngrx/signals/rxjs-interop. The companion tapResponse operator lives in @ngrx/operators - a separate package you'll almost always want alongside it.

Anatomy of rxMethod

An rxMethod has two parts: the definition (where you build the pipeline) and the invocation (where you feed it input).

import { rxMethod } from '@ngrx/signals/rxjs-interop';
import { pipe, switchMap, tap } from 'rxjs';
import { tapResponse } from '@ngrx/operators';

// Definition: build the pipeline
const loadPolicies = rxMethod<string>(
  pipe(
    tap(() => patchState(store, { isLoading: true })),
    switchMap((searchTerm) =>
      policyService.search(searchTerm).pipe(
        tapResponse({
          next: (policies) => patchState(store, setAllEntities(policies), { isLoading: false }),
          error: () => patchState(store, { isLoading: false, error: 'Search failed' }),
        })
      )
    )
  )
);

The generic type parameter (<string>) defines what the pipeline receives as input. The pipe(...) receives an Observable<string> as its source - every value you feed in arrives on that source observable.

Three Ways to Call rxMethod

Once defined, an rxMethod accepts three kinds of input:

// 1. Static value — emits once, immediately
loadPolicies('home insurance');

// 2. Signal (or computation function) — re-emits every time the signal changes
const searchTerm = signal('');
loadPolicies(searchTerm);
// later: searchTerm.set('auto') → pipeline re-runs with 'auto'

// 3. Observable — each emission feeds the pipeline
const search$ = fromEvent(input, 'input').pipe(map(e => e.target.value));
loadPolicies(search$);

This is the key design insight: the pipeline is defined once, but the input source is pluggable. The same rxMethod can be driven by a signal (reactive), an observable (imperative/event-driven), or a one-shot value (manual trigger).

Signal tracking: When you pass a signal to rxMethod, it creates an internal effect() that watches the signal. Every time the signal changes, the new value is pushed into the pipeline's source observable. This means your switchMap / debounceTime / etc. all work exactly as you'd expect - including cancelling in-flight requests on re-emission.

tapResponse: Safe Error Handling

tapResponse is a purpose-built RxJS operator from @ngrx/operators that does two things:

Routes success and error to separate callbacks (like tap + catchError)
Catches the error without killing the outer subscription

That second point is critical. In a long-lived pipeline, if an inner observable errors and you don't catch it, the entire rxMethod subscription dies — no more values will ever flow through. tapResponse prevents that by catching inside the inner observable.

import { tapResponse } from '@ngrx/operators';

// Full signature
tapResponse({
  next: (value) => { /* handle success */ },
  error: (error) => { /* handle error - pipeline stays alive */ },
  complete?: () => { /* optional: inner observable completed */ },
  finalize?: () => { /* optional: runs after success OR error */ },
});

Why not just catchError? You can use catchError inside the inner observable, but you must remember to return EMPTY (or another observable) to keep the outer alive. tapResponse encapsulates that pattern so you can't accidentally forget. It also provides a finalize callback that runs in both success and error paths - useful for clearing loading flags.

rxMethod Inside a SignalStore

The typical pattern: define rxMethod inside withMethods, then connect it to a signal in withHooks:

import { computed, inject } from '@angular/core';
import { signalStore, withState, withComputed, withMethods, withHooks, patchState } from '@ngrx/signals';
import { withEntities, setAllEntities } from '@ngrx/signals/entities';
import { rxMethod } from '@ngrx/signals/rxjs-interop';
import { tapResponse } from '@ngrx/operators';
import { debounceTime, distinctUntilChanged, pipe, switchMap, tap } from 'rxjs';

export interface Policy {
  id: string;
  holder: string;
  premium: number;
  status: 'active' | 'expired' | 'pending';
}

export const PolicySearchStore = signalStore(
  { providedIn: 'root' },
  withState({ query: '', isLoading: false, error: null as string | null }),
  withEntities<Policy>(),
  withComputed(({ entities }) => ({
    resultCount: computed(() => entities().length),
  })),
  withMethods((store, policyService = inject(PolicyService)) => ({

    setQuery(query: string): void {
      patchState(store, { query });
    },

    // rxMethod: reactive search pipeline
    searchPolicies: rxMethod<string>(
      pipe(
        debounceTime(300),
        distinctUntilChanged(),
        tap(() => patchState(store, { isLoading: true, error: null })),
        switchMap((query) =>
          policyService.search(query).pipe(
            tapResponse({
              next: (policies) => patchState(store, setAllEntities(policies)),
              error: (e: Error) => patchState(store, { error: e.message }),
              finalize: () => patchState(store, { isLoading: false }),
            })
          )
        )
      )
    ),
  })),
  withHooks({
    onInit(store) {
      // Connect the query signal to the search pipeline
      // Every time store.query() changes → pipeline re-runs (debounced)
      store.searchPolicies(store.query);
    },
  })
);

What's happening here:

setQuery updates the query state via patchState
searchPolicies is defined as an rxMethod<string> with debounce + switchMap
In onInit, we pass store.query (a signal) to searchPolicies - this creates a reactive binding
Every time query changes, the pipeline fires: debounces 300ms, skips duplicates, cancels in-flight requests, makes the API call

Note: We pass store.query (the signal reference), not store.query() (the current value). Passing the signal creates a live connection. Passing the value would be a one-shot static call.

Lifecycle and Cleanup

rxMethod ties its subscription lifetime to the injector that created it. When the store is destroyed (component unmounts for component-scoped stores, or app shuts down for root stores), the subscription is automatically unsubscribed. No manual cleanup needed.

When you pass a signal or observable as input, additional subscriptions are created for that specific invocation. These are also tied to the caller's injection context:

// In a component - subscription lives as long as the component
@Component({ ... })
export class PolicyListComponent {
  private store = inject(PolicySearchStore);
  private searchInput = signal('');

  constructor() {
    // This watcher is tied to the component's DestroyRef
    // Component destroyed → watcher cleaned up → no more emissions
    this.store.searchPolicies(this.searchInput);
  }
}

If you need to call an rxMethod with a signal outside an injection context (e.g. in a callback), pass an injector explicitly:

private injector = inject(Injector);

someCallback(): void {
  this.store.searchPolicies(this.searchInput, { injector: this.injector });
}

Manual Cleanup with destroy()

Every rxMethod invocation returns a RxMethodRef with a destroy() method. Use it when you need fine-grained control over when to stop listening:

const ref = store.searchPolicies(searchTerm$);

// Later: stop this particular subscription
ref.destroy();

// The rxMethod itself is still alive — you can call it again
store.searchPolicies(anotherSignal);

You can also call destroy() on the rxMethod function itself to kill the entire pipeline (all invocations):

store.searchPolicies.destroy(); // Kills the pipeline entirely

rxMethod<void> for Trigger-Only Pipelines

Not every pipeline needs input data. Use rxMethod<void> for effects that just need to be triggered:

withMethods((store, policyService = inject(PolicyService)) => ({
  loadAll: rxMethod<void>(
    pipe(
      tap(() => patchState(store, { isLoading: true })),
      exhaustMap(() =>
        policyService.getAll().pipe(
          tapResponse({
            next: (policies) => patchState(store, setAllEntities(policies)),
            error: (e: Error) => patchState(store, { error: e.message }),
            finalize: () => patchState(store, { isLoading: false }),
          })
        )
      )
    )
  ),
})),
withHooks({
  onInit(store) {
    store.loadAll(); // Fire once - no input needed
  },
})

Note the use of exhaustMap here instead of switchMap: if the user triggers a reload while one is in flight, exhaustMap ignores the new trigger until the current request completes. Choose your flattening operator based on the use case:

switchMap - cancel previous, use latest (search, autocomplete)
exhaustMap - ignore new until current finishes (form submit, initial load)
concatMap - queue and execute in order (ordered writes)
mergeMap - run all in parallel (independent fire-and-forget)

Reacting to Multiple Signals

What if your pipeline depends on multiple signals? You have two approaches:

Option 1: Computed signal as input

withState({ status: 'all' as string, page: 1 }),
withMethods((store, policyService = inject(PolicyService)) => ({
  loadFiltered: rxMethod<{ status: string; page: number }>(
    pipe(
      distinctUntilChanged((a, b) => a.status === b.status && a.page === b.page),
      switchMap(({ status, page }) =>
        policyService.getFiltered(status, page).pipe(
          tapResponse({
            next: (policies) => patchState(store, setAllEntities(policies)),
            error: (e: Error) => patchState(store, { error: e.message }),
          })
        )
      )
    )
  ),
})),
withHooks({
  onInit(store) {
    // Pass a computation function - re-evaluates when either signal changes
    store.loadFiltered(() => ({ status: store.status(), page: store.page() }));
  },
})

The computation function (() => ({ ... })) is tracked by an internal effect(). Whenever store.status() or store.page() changes, the function re-evaluates and the new object is pushed into the pipeline.

Option 2: Separate rxMethods per concern

Sometimes it's cleaner to have independent pipelines that each watch one signal, rather than combining everything into one mega-pipeline. This is a design judgment - there's no single right answer.

rxMethod Outside a Store

rxMethod isn't exclusive to SignalStore. You can use it directly in components or services - anywhere you have an injection context:

import { Component, inject, signal } from '@angular/core';
import { rxMethod } from '@ngrx/signals/rxjs-interop';
import { tapResponse } from '@ngrx/operators';
import { debounceTime, distinctUntilChanged, pipe, switchMap } from 'rxjs';

@Component({
  selector: 'app-search',
  template: `<input (input)="query.set($event.target.value)">`,
})
export class SearchComponent {
  private searchService = inject(SearchService);
  query = signal('');
  results = signal<Result[]>([]);

  private search = rxMethod<string>(
    pipe(
      debounceTime(300),
      distinctUntilChanged(),
      switchMap((q) =>
        this.searchService.search(q).pipe(
          tapResponse({
            next: (r) => this.results.set(r),
            error: console.error,
          })
        )
      )
    )
  );

  constructor() {
    this.search(this.query); // reactive binding
  }
}

This is essentially a lightweight alternative to a full store when all you need is one reactive pipeline in a component.

Classic NgRx Comparison

Classic NgRx (Effects)

// actions.ts
export const searchPolicies = createAction('[Policy] Search', props<{ query: string }>());
export const searchPoliciesSuccess = createAction('[Policy] Search Success', props<{ policies: Policy[] }>());
export const searchPoliciesFailure = createAction('[Policy] Search Failure', props<{ error: string }>());

// effects.ts
searchPolicies$ = createEffect(() =>
  this.actions$.pipe(
    ofType(searchPolicies),
    debounceTime(300),
    distinctUntilChanged((a, b) => a.query === b.query),
    switchMap(({ query }) =>
      this.policyService.search(query).pipe(
        map((policies) => searchPoliciesSuccess({ policies })),
        catchError((error) => of(searchPoliciesFailure({ error: error.message })))
      )
    )
  )
);

// reducer.ts
on(searchPolicies, (state) => ({ ...state, isLoading: true })),
on(searchPoliciesSuccess, (state, { policies }) => ({ ...state, policies, isLoading: false })),
on(searchPoliciesFailure, (state, { error }) => ({ ...state, error, isLoading: false })),

SignalStore (rxMethod)

// Everything in one file - no actions, no reducer cases, no effects class
searchPolicies: rxMethod<string>(
  pipe(
    debounceTime(300),
    distinctUntilChanged(),
    tap(() => patchState(store, { isLoading: true, error: null })),
    switchMap((query) =>
      policyService.search(query).pipe(
        tapResponse({
          next: (policies) => patchState(store, setAllEntities(policies)),
          error: (e: Error) => patchState(store, { error: e.message }),
          finalize: () => patchState(store, { isLoading: false }),
        })
      )
    )
  )
)

Same operators, same RxJS pipeline logic. The difference: no action dispatch indirection. The pipeline is co-located with the state it updates. Input comes directly from a signal rather than being serialised through the action stream.

When to Use rxMethod vs async/await

Scenario	Use	Why
One-shot load on init	`async/await`	No reactive input, no cancellation needed
Search with debounce	`rxMethod`	Needs debounceTime + switchMap
Re-fetch when filter signal changes	`rxMethod`	Reactive re-execution on signal change
Form submit (fire once)	`async/await`	Imperative, no stream semantics needed
Polling / WebSocket	`rxMethod`	Long-lived subscription with cleanup
Optimistic update + rollback	`rxMethod`	Needs error recovery without killing the pipeline
Sequential API calls (await result of first)	`async/await`	Clearer with imperative flow

The rule of thumb: if the method is called once and you just need the result, async/await is simpler. If the method reacts to changing input, needs cancellation semantics, or is a long-lived subscription, reach for rxMethod.

Full Example: Policy Dashboard with Reactive Search

import { computed, inject } from '@angular/core';
import { signalStore, withState, withComputed, withMethods, withHooks, patchState } from '@ngrx/signals';
import { withEntities, setAllEntities, removeEntity } from '@ngrx/signals/entities';
import { rxMethod } from '@ngrx/signals/rxjs-interop';
import { tapResponse } from '@ngrx/operators';
import { debounceTime, distinctUntilChanged, exhaustMap, pipe, switchMap, tap } from 'rxjs';

export const PolicyDashboardStore = signalStore(
  { providedIn: 'root' },
  withEntities<Policy>(),
  withState({
    query: '',
    statusFilter: 'all' as 'all' | 'active' | 'expired',
    isLoading: false,
    error: null as string | null,
  }),
  withComputed(({ entities, statusFilter }) => ({
    filteredPolicies: computed(() => {
      const status = statusFilter();
      return status === 'all' ? entities() : entities().filter(p => p.status === status);
    }),
  })),
  withMethods((store, policyService = inject(PolicyService)) => ({
    setQuery(query: string): void {
      patchState(store, { query });
    },
    setStatusFilter(status: 'all' | 'active' | 'expired'): void {
      patchState(store, { statusFilter: status });
    },

    // Reactive search: re-runs when query or statusFilter changes
    loadPolicies: rxMethod<{ query: string; status: string }>(
      pipe(
        debounceTime(300),
        distinctUntilChanged((a, b) => a.query === b.query && a.status === b.status),
        tap(() => patchState(store, { isLoading: true, error: null })),
        switchMap(({ query, status }) =>
          policyService.search(query, status).pipe(
            tapResponse({
              next: (policies) => patchState(store, setAllEntities(policies)),
              error: (e: Error) => patchState(store, { error: e.message }),
              finalize: () => patchState(store, { isLoading: false }),
            })
          )
        )
      )
    ),

    // Fire-and-forget delete (no reactive input)
    deletePolicy: rxMethod<string>(
      pipe(
        exhaustMap((id) =>
          policyService.delete(id).pipe(
            tapResponse({
              next: () => patchState(store, removeEntity(id)),
              error: (e: Error) => patchState(store, { error: e.message }),
            })
          )
        )
      )
    ),
  })),
  withHooks({
    onInit(store) {
      // Reactive binding: re-search whenever query or status changes
      store.loadPolicies(() => ({
        query: store.query(),
        status: store.statusFilter(),
      }));
    },
  })
);

This store combines both patterns: loadPolicies is a reactive pipeline driven by signals, while deletePolicy is called imperatively with a static ID. Both use tapResponse for safe error handling.

Quiz

What does `rxMethod` accept as its definition argument?

An async function that returns a Promise

A function that receives an Observable and returns an Observable (i.e. an RxJS pipe)

A callback that runs on every signal change

What happens when you pass a signal to an `rxMethod` invocation?

The signal's current value is read once and the pipeline runs once

The signal is converted to an observable using toObservable()

An internal effect() watches the signal and pushes each new value into the pipeline's source observable

Why use `tapResponse` instead of plain `catchError` in an rxMethod pipeline?

tapResponse is faster because it skips error serialisation

tapResponse catches the error without killing the outer subscription, keeping the long-lived pipeline alive

catchError doesn't work inside switchMap

Try It Yourself

Exercise 1: Polling Store

Build a store that polls an API every 10 seconds for updated policy data. Use rxMethod<void> with interval or timer. Include a way to stop polling:

// TODO: PollingStore with rxMethod that polls every 10s
// Hint: timer(0, 10_000) gives you an immediate emit + 10s interval

import { inject } from '@angular/core';
import { signalStore, withMethods, withHooks, patchState, withState } from '@ngrx/signals';
import { withEntities, setAllEntities } from '@ngrx/signals/entities';
import { rxMethod } from '@ngrx/signals/rxjs-interop';
import { tapResponse } from '@ngrx/operators';
import { pipe, switchMap, timer } from 'rxjs';

export const PollingStore = signalStore(
  withEntities<Policy>(),
  withState({ lastUpdated: null as string | null }),
  withMethods((store, policyService = inject(PolicyService)) => ({
    startPolling: rxMethod<void>(
      pipe(
        switchMap(() =>
          timer(0, 10_000).pipe(
            switchMap(() =>
              policyService.getAll().pipe(
                tapResponse({
                  next: (policies) => patchState(store,
                    setAllEntities(policies),
                    { lastUpdated: new Date().toISOString() }
                  ),
                  error: console.error,
                })
              )
            )
          )
        )
      )
    ),
  })),
  withHooks({
    onInit(store) {
      store.startPolling();
    },
    // Cleanup is automatic — when the store's injector is destroyed,
    // the rxMethod subscription (including the timer) is unsubscribed.
  })
);

timer(0, 10_000) emits immediately then every 10 seconds. The outer switchMap means calling startPolling() again would restart the timer. Unsubscription is automatic via DestroyRef.

Exercise 2: Autocomplete with Minimum Length

Create an rxMethod<string> that only fires the API call when the input is at least 3 characters. Debounce 400ms, cancel previous on new input:

// TODO: autocomplete rxMethod - minimum 3 chars, debounce 400ms, switchMap

autocomplete: rxMethod<string>(
  pipe(
    debounceTime(400),
    distinctUntilChanged(),
    filter((query) => query.length >= 3),
    tap(() => patchState(store, { isSearching: true })),
    switchMap((query) =>
      suggestService.getSuggestions(query).pipe(
        tapResponse({
          next: (suggestions) => patchState(store, { suggestions }),
          error: () => patchState(store, { suggestions: [] }),
          finalize: () => patchState(store, { isSearching: false }),
        })
      )
    )
  )
)

The filter operator gates the pipeline - values shorter than 3 characters never reach switchMap. Note: if the user clears the input to 2 chars, no request fires. You might also want to clear suggestions in that case via a separate tap before the filter.

Exercise 3: Optimistic Delete with Rollback

Implement a delete that immediately removes the entity from the store (optimistic), makes the API call, and rolls back if the API fails:

// TODO: optimisticDelete rxMethod<Policy> - remove immediately, rollback on error

optimisticDelete: rxMethod<Policy>(
  pipe(
    concatMap((policy) => {
      // Optimistic: remove immediately
      patchState(store, removeEntity(policy.id));

      return policyService.delete(policy.id).pipe(
        tapResponse({
          next: () => { /* already removed - nothing to do */ },
          error: () => {
            // Rollback: re-add the entity
            patchState(store, addEntity(policy));
            patchState(store, { error: 'Delete failed — restored policy' });
          },
        })
      );
    })
  )
)

concatMap ensures deletes are processed one at a time (no race conditions on rollback). The entity is saved in the closure so we can re-add it on failure.

Your Tangible Win

You now have two async patterns in your toolkit: async/await for simple fire-and-forget operations, and rxMethod for reactive pipelines that need the full power of RxJS operators. Combined with tapResponse for safe error handling, this replaces classic NgRx Effects with far less ceremony - same operators, no action/effect/reducer indirection.

Primary sources: NgRx RxJS Integration Guide - canonical documentation for rxMethod. NgRx Operators - tapResponse and mapResponse documentation.

Something unclear? Ask me follow-up questions - I'm your teacher and can clarify anything, adjust examples, or go deeper on any concept.