RxJS subscription in Angular component

Question

I don't completely understand where and how I need to declare observables / subjects in Angular component.

Currently I develop a website which interacts with MovieDB API and I have everything working, but at the same time I understand that my code is bad, because there aren`t cleanups of subscriptions after destroying the component, but to do these cleanups I need at least understand how to use RxJS correctly.

I think that my usage is incorrect, because I have new subscription on every interaction with the page. Also as I understand they need to be declared in constructor.

The idea of this page is that there is an input form where user types the query and checks the radio-button to choose what to search: 'tv' or 'movies'. When there are the results of searching, there appears a button to expand the results.

So, here is the code:

import {Component, OnDestroy} from '@angular/core';
import {ISearchParams} from '../../models/search-params.interface';
import {ShowService} from '../../services/show.service';
import {IResultsIds} from '../../models/results.interface';
import {distinctUntilChanged} from 'rxjs/operators';
import {Subscription} from 'rxjs';

@Component({
  selector: 'app-search',
  templateUrl: './search-tab.component.html',
  styleUrls: ['./search-tab.component.scss']
})
export class SearchTabComponent implements OnDestroy {

  searchParams!: ISearchParams;

  searchResults!: IResultsIds;

  searchSub!: Subscription;
  showMoreSub!: Subscription;

  constructor(private movieService: ShowService) {
  }

  search(searchParams: ISearchParams): void {
    this.searchParams = searchParams;

    this.searchSub = this.movieService.search(searchParams)
      .pipe(distinctUntilChanged())
      .subscribe(results => {
        this.searchResults = results;
      });
  }

  showMore(): void {
    if (!this.isFinished()) {
      this.searchParams.page++;

      this.showMoreSub = this.movieService.search(this.searchParams)
        .subscribe(results => {
          this.searchResults!.ids.push(...results.ids);
        });
    }
  }

  isFinished = () => this.searchParams.page >= this.searchResults!.total_pages;

  ngOnDestroy(): void {
    // this.searchSub.unsubscribe();
    // this.showMoreSub.unsubscribe();
  }
}

And the HTML:

<main class="main-content search-container">
  <app-search-form (searchParams)="search($event)"></app-search-form>
  <div class="results" *ngIf="searchResults">
    <app-show-description *ngFor="let id of searchResults.ids"
                          [showType]="searchParams.type"
                          [showId]="id"></app-show-description>
  </div>
  <button *ngIf="searchResults && !isFinished()"
          (click)="showMore()"
          class="load-btn more-btn">show more...</button>
</main>

I will be very grateful if you help me and tell where I made mistakes. On more time, everything works this way, but I want to understand usage of RxJS.

UPD @H3AR7B3A7 thank you very much, you have clearified my knowledge a bit! but now I have another misunderstanding: how to transform 2 array observables into one? I have googled but I can't find the solution to my problem: I have function to load more movies - it has to extend the array of distinctMovies$, which looks just like an array of numbers (ids), but I can't join two arrays, all I get is [...], [...] but not [......]

I tried this:

showMore(): void {
    this.searchParams.page++;

    this.distinctMovies$ = concat(this.distinctMovies$,
      this.movieService.search(this.searchParams)
        .pipe(pluck('ids')))
      .pipe(tap(console.log));
  }

Answer 1

It looks like you understand the basics to me.

Although, you don't need to declare anything in the constructor.

Also as I understand they need to be declared in constructor.

You usually only use the constructor to inject services like you are doing already.

You might want to use the ngOnInit() method to declare your initial state of the component:

export class SearchTabComponent implements OnInit {
  distinctMovies$!: Observable<Movie[]>

  ngOninit(): void {
    //...
  }
}

You can solve a lot of your problems by just never subscribing in your component code. This way you never have to unsubscribe in an OnDestroy either...

For example (in you onInit):

this.distinctMovies$ = this.movieService.search(searchParams).pipe(distinctUntilChanged())

And in the template just use an async pipe:

*ngFor = "let movie of distinctMovies$ | async"

Apart from not having to unsubscribe you can also use the OnPush ChangeDetectionStrategy by using async pipes instead of subscribing.

Answer 2

By looking at the first glance, there are a few things that you can improve here.

Regarding unsubscribing

There are a several patterns considered as a "good practise" for this. One of them is:

// Define destroy subject
readonly destroy$ = new Subject<any>();

// On each subscription add
.pipe(takeUntil(this.destroy$))

// And just emit value to destroy$ subject in onDestroy hook
// And all streams that had takeUntil will be ended
ngOnDestroy(): void {
  this.destroy$.next();
  this.destroy$.complete();
}

Regarding searching things

When you are dealing with async requests (such as API calls), you have to think what would happen if for example you click several times on Search button. There will be several API calls, right? but we are not 100% sure what would be order of the responses from the server.

If you click for example:

Click 1, Click 2, Click 3, Click 4... will produce APICall1, APICall2, APICall3, APICALL4... that would be in that order... But responses from the server can be in some other order (yeah thats possible), also if you click several times without delay, then your server will get many requests at the moment but you will probably need only the last one.

So one of the common solutions is to have a stream that is listening to for example search changes:

searchTerm = new Subject();

ngOnInit() {

  this.searchTerms
    .pipe(
      distinctUntilChanged(),
      // Debounce for 250 miliseconds (avoid flooding backend server with too many requests in a short period of time)
      debounceTime(250),
      // Call api calls
      switchMap(terms => {
        return this.movieService.serach(terms);
      })),
      // Unsubsribe when onDestroy is triggered
      takeUntil(this.destroy$),
    .subscribe(results => {
         // Push/set results to this.searchResults
         if (this.addingMore) {
         //  push in this.searchResults
         } else {
            this.searchResults = results;
         }

         this.addingMore = false;
    });

}

search(searchTerms) {
  this.searchParams = searchParams;
  this.searchTerms.next(searchTerms);
}

showMore(): void {
  if (this.isFinished() return;

  this.searchParams.page++;

  // You can for example set a flag to know if its search or loadMore
  this.addingMore = true;
 
  this.searchTerms.next(searchTerms);
}