How should I filter Observables the "Right Way" with RxJs in Angular 17?

Question

With one Subscriber, an async list showing all items from Observable, what is the most efficient way to temporarily filter results based on user provided search criteria as a string?

The component imports TokenService and the template subscribes to tokens: tokenService.tokens | async

// ./token.service.ts

@Injectable({
  providedIn: 'root'
})
export class TokenService {

  private tokens$ = new BehaviorSubject<Token[]>([]);

  readonly tokens = this.tokens$.asObservable();

  constructor() {
    this.getTokens();
  }

  filterTokens(q?: string) { ... }

  ...
}

The output should display a list of items with names loosely matching the provided search criteria. Provided the letter w (lower case) as input "Work" and "New" would display from a set of ['Work', 'Test', 'New'].

Passing the newly filtered values to the BehaviorSubject obviously makes the values unrecoverable unless cached elsewhere.

  filterTokens(q?: string): void { 
    this.tokens$.next(
      this.tokens$.getValue().filter(t => t.issuer.toLowerCase().includes(q.toLowerCase()))
    );
  }

The following method returns "Work" and "New" when passed tokenService.filterTokens('r') | async:

  filterTokens(q?: string): Observable<Token[]> {
    return this.tokens.pipe(
      map(tt => tt.filter(t => t.issuer.toLowerCase().includes(q.toLowerCase())))
    );
  }

This however, is not dynamic based on user input.

Answer 1

The cleanest way to set up your filterTokens method as you have already done:

// service
filterTokens(q?: string): Observable<Token[]> {
   return this.tokens.pipe(
      map(tt => tt.filter(
         t => t.issuer.toLowerCase().includes(q.toLowerCase())
      ))
   );
}

And have the consumer (usually in the component), utilize a switchMap to call the function whenever the input value changes:

// component
filteredTokens$ = this.searchString$.pipe(
   switchMap(str => this.service.filterTokens(str))
);

If you're not familiar with switchMap, it simply takes the incoming value and maps it to a new observable. Then it will emit the emissions from this new observable. Whenever it receives a new value (search string), it will unsubscribe from the prior "filterTokens observable" and subscribe to the new one.

This means it will always emit the results from service.filterTokens() using the most recent value of searchString$.

---

Also, you can simplify your service further, by not using a BehaviorSubject at all. Just declare your tokens as an observable to the fetch the data, followed by shareReplay:

export class TokenService {

  tokens$: Observable<Token[]> = this.http.get('/url').pipe(shareReplay(1));

  filterTokens(searchString: string): Observable<Token[]> { 
    return this.tokens$.pipe(
      map(tokens => tokens.filter(issuerIncludes(searchString))
    );
  }

}

function issuerIncludes(searchString: string): (t:Token) => boolean {
    return t => t.issuer.toLowerCase().includes(searchString.toLowerCase()))));
}

Notice we declare tokens$ with the call to fetch the data then add switchMap(1), which means any subscribers will get the most recently emitted value upon subscription.

Notice here we don't need to call getTokens() in the constructor (which I'm assuming was doing some sort of http call, and explicitly subscribing). In general, you don't want to subscribe in your services, but rather expose observables that consumers subscribe to when needed. This keeps your data flow completely lazy, because the http call only gets made when there is a subscriber to one of the derived observables, not every time the service is injected.

Notice the filter function has been broken out into its own pure function outside the class to improve readability, but of course this is optional.

Answer 2

There isn't a "Right Way" to do this, there is only the "Way that works for your application". That being said, there are multiple ways to achieve this, one way is to have two Subjects, one for holding a list of tokens, the other for doing the search, then merging the two.

Working Example

Brief Overview of the Filter

Not a query

This is what happens when this.tokens is updated either by initialization or adding/removing a token to the array cache (this.tokens.next([...this.tokens.value, newToken])). The result will just be a Token[].

Doing a query

This is what happens when this.query.next() is called:

• from passes each array item to the pipe (like a for loop)
• filter tests each of the tokens to test the search string (fuzzySearch returns true/false if the token matches or not)
• toArray converts each observable that passed the filter test back to a Token[] that matched the search

The merge()

The merge takes the two Observables and runs each when something happens. The result is a single Observable from whichever ran. In this case it will be a Token[].

---

Implementation

Not sure what your Token class is, but here is what I did:

export class Token {
  constructor(private name: string) {}

  getValue = computed(() => this.name);
}

Next we have the Service:

@Injectable({ providedIn: 'root' })
export class TokenService {
  private tokens = new BehaviorSubject<Token[]>([]);
  private query = new Subject<string>();
  
  // Perform the filter
  private query$ = this.query.pipe(
    switchMap(q =>
      from(this.tokens.value).pipe(
        filter(token => this.fuzzySearch(token.getValue(), q)),
        toArray(),
      ),
    ),
  );

  // Watch for changes on each
  readonly tokens$ = merge(this.tokens, this.query$);

  // Trigger the search
  search(query: string) {
    this.query.next(query);
  }

  // Fuzzy Search Credit: https://stackoverflow.com/a/15252131/1778465
  fuzzySearch(word: string, q: string): bool {
    let hay = word.toLowerCase(), i = 0, n = -1, l;
    q = q.toLowerCase();
    for (; (l = q[i++]); ) if (!~(n = hay.indexOf(l, n + 1))) return false;
    return true;
  }
}

And lastly the Component:

@Component({
  selector: 'app-root',
  standalone: true,
  imports: [FormsModule, AsyncPipe],
  template: `
    <p>
      <input ngModel (ngModelChange)="tokenService.search($event)" />
    </p>

    @for (token of tokens$ | async; track token) {
      <p>{{ token.getValue() }}</p>
    }
  `,
})
export class App {
  tokens$ = this.tokenService.tokens$;

  constructor(protected tokenService: TokenService) {}
}