RxJS Error Handling Operators

Operator catchError trong RxJS là công cụ quan trọng giúp chúng ta xử lý lỗi trong Observable, tránh việc kết thúc sớm của chuỗi các sự kiện. Khi một lỗi xuất hiện trong Observable, catchError cho phép chúng ta xử lý lỗi này bằng cách trả về một Observable mới hoặc ném ra một lỗi khác để tiếp tục chuỗi các sự kiện.

Ví dụ, khi chúng ta có một chuỗi số từ 1 đến 5 và muốn kiểm tra các số trùng lặp trong một mảng đã được cache:

import { of, throwError } from "rxjs";
import { map, catchError } from "rxjs/operators";

const cached = [4, 5];
of(1, 2, 3, 4, 5)
  .pipe(
    map((n) => {
      if (cached.includes(n)) {
        throw new Error("Duplicated: " + n);
      }
      return n;
    }),
    catchError((err, caught) => of(err))
  )
  .subscribe(observer);

Trong ví dụ này, khi một số đã được cache xuất hiện trong chuỗi số emit, chúng ta ném ra một lỗi thông báo số đã trùng lặp. Tuy nhiên, chúng ta không kết thúc chuỗi sự kiện mà thay vào đó, thông qua catchError, chúng ta trả về một Observable mới (of(err)) để xử lý lỗi và tiếp tục chuỗi các sự kiện.

Ngoài ra, catchError cũng cho phép retry, tức là thử lại Observable ban đầu sau khi xử lý lỗi. Tuy nhiên, điều này cần được thực hiện cẩn thận để tránh lặp vô hạn. Việc kết hợp catchError với take có thể giới hạn số lần thử lại.

of(1, 2, 3, 4, 5)
  .pipe(
    map((n) => {
      if (cached.includes(n)) {
        throw new Error("Duplicated: " + n);
      }
      return n;
    }),
    catchError((err, caught) => caught), // Retry by returning the source Observable
    take(10) // Limit the retries to 10 times
  )
  .subscribe(observer);

Nhớ rằng bạn cũng có thể ném ra một lỗi mới từ trong catchError, để cho phép các operators sau có thể xử lý tiếp lỗi đó.

retry trong RxJS

Operator retry trong RxJS là một công cụ hữu ích cho việc xử lý lỗi trong Observable. Khi có lỗi xuất hiện trong source Observable, retry sẽ thực hiện việc resubscribe vào Observable ban đầu để thử lại chuỗi sự kiện. Điều này có thể giúp giảm thiểu việc kết thúc sớm của chuỗi các sự kiện do lỗi.

import { of } from "rxjs";
import { map, retry } from "rxjs/operators";

const cached = [4, 5];
of(1, 2, 3, 4, 5)
  .pipe(
    map((n) => {
      if (cached.includes(n)) {
        throw new Error("Duplicated: " + n);
      }
      return n;
    }),
    retry(3) // Retry up to 3 times
  )
  .subscribe(observer);

Trong ví dụ trên, khi một số đã được cache xuất hiện trong chuỗi số emit, chúng ta ném ra một lỗi thông báo số đã trùng lặp. Thông qua retry(3), chúng ta yêu cầu Observable thử lại tối đa 3 lần khi gặp lỗi. Kết quả là chuỗi sự kiện sẽ cố gắng thử lại và tiếp tục emit các giá trị sau mỗi lần thất bại.

Ngoài ra, chúng ta có thể sử dụng retryWhen để kiểm soát quá trình retry một cách linh hoạt hơn, cho phép chúng ta quyết định khi nào cần thực hiện retry.

Một ứng dụng khá thú vị là retryBackoff operator, nó sẽ gia tăng thời gian chờ sau mỗi lần thử lại. Điều này có thể hữu ích trong việc quản lý quá trình retry, giúp giảm áp lực lên server trong trường hợp xảy ra lỗi liên tục.

RxJS Error Conditional Operators

Operator defaultIfEmpty và throwIfEmpty trong RxJS giúp xử lý trường hợp source stream không phát ra bất kỳ giá trị nào (empty), chỉ kết thúc mà không có giá trị được phát ra.

Operator defaultIfEmpty cho phép chúng ta định nghĩa một giá trị mặc định để trả về nếu source stream là empty.

Operator throwIfEmpty cho phép chúng ta ném ra một lỗi được xác định nếu source stream là empty.

Ví dụ, chúng ta muốn thông báo lỗi nếu người dùng không thực hiện click vào trong vòng 1s:

import { fromEvent, timer } from "rxjs";
import { throwIfEmpty, takeUntil } from "rxjs/operators";

const click$ = fromEvent(document, "click");

click$
  .pipe(
    takeUntil(timer(1000)),
    throwIfEmpty(() => new Error("The document was not clicked within 1 second"))
  )
  .subscribe(observer);

Trong đoạn mã trên, chúng ta theo dõi sự kiện click từ document. Nếu không có sự kiện click nào trong vòng 1 giây (timer(1000)), throwIfEmpty sẽ ném ra một lỗi với thông báo “The document was not clicked within 1 second”. Nếu có sự kiện click xảy ra trước khi hết 1s, việc theo dõi sự kiện click sẽ kết thúc và không có lỗi nào được ném ra.

every trong RxJS

Operator every trong RxJS tạo ra một Observable phát ra giá trị true hoặc false tùy thuộc vào việc tất cả các phần tử trong nguồn đều thỏa mãn điều kiện được chỉ định.

Nó sẽ trả về true nếu tất cả các giá trị được phát ra từ nguồn đều thoả mãn điều kiện được xác định trong hàm predicate.

import { of } from "rxjs";
import { every } from "rxjs/operators";

of(1, 2, 3, 4, 5, 6)
  .pipe(every((x) => x < 5))
  .subscribe(observer);
/**
 * Output:
 * ------false|
 */

Nếu có bất kỳ giá trị nào không thỏa mãn điều kiện, nó sẽ phát ra false và kết thúc Observable. Lưu ý rằng nếu nguồn không hoàn thành (complete) thì sẽ không có giá trị nào được phát ra.

Nếu bạn muốn tìm một operator tương tự như some trong Array của JavaScript trong RxJS, bạn có thể sử dụng operator first kết hợp với hàm predicate. Ví dụ, giống như việc sử dụng some trong Router của Angular:

import { of } from "rxjs";
import { first, map } from "rxjs/operators";

of(1, 2, 3, 14, 5, 6)
  .pipe(
    first((x) => x > 10, false),
    map((v) => Boolean(v))
  )
  .subscribe(observer);
/**
 * Output:
 * ------true|
 */

Trong ví dụ này, operator first kết hợp với hàm predicate sẽ trả về true nếu có bất kỳ giá trị nào trong nguồn thoả mãn điều kiện. Nếu không có giá trị nào thỏa mãn, nó sẽ trả về giá trị mặc định false mà chúng ta đã xác định. Sau đó, chúng ta sử dụng map để chuyển đổi giá trị thành kiểu boolean trước khi subscribe.

if trong RxJS

Operator iif trong RxJS là một cách để quyết định vào thời điểm subscribe là Observable nào sẽ được sử dụng thực sự dựa trên điều kiện đã xác định.

Nó nhận vào một hàm điều kiện và hai Observables. Khi một Observable được trả về bởi operator này được subscribe, hàm điều kiện sẽ được gọi. Dựa trên giá trị boolean nó trả về tại thời điểm đó, người tiêu dùng sẽ subscribe vào Observable đầu tiên (nếu điều kiện là true) hoặc vào Observable thứ hai (nếu điều kiện là false). Hàm điều kiện cũng có thể không trả về gì cả – trong trường hợp đó, điều kiện sẽ được đánh giá là false và Observable thứ hai sẽ được subscribe.

Lưu ý rằng các Observables cho cả hai trường hợp (true và false) đều là tùy chọn. Nếu điều kiện trỏ đến một Observable mà không được xác định, luồng kết quả sẽ đơn giản chỉ hoàn thành ngay lập tức. Điều này cho phép bạn, thay vì kiểm soát Observable nào sẽ được subscribe, quyết định vào thời gian chạy liệu người tiêu dùng có truy cập vào Observable đã cho hay không.

Nếu bạn có logic phức tạp hơn yêu cầu quyết định giữa nhiều hơn hai Observables, defer có lẽ sẽ là lựa chọn tốt hơn. Trên thực tế, iif có thể dễ dàng được thực hiện với defer và tồn tại chỉ cho mục đích tiện lợi và đọc hiểu.

import { iif, of } from "rxjs";

let subscribeToFirst;
const firstOrSecond = iif(() => subscribeToFirst, of("first"), of("second"));

subscribeToFirst = true;
firstOrSecond.subscribe((value) => console.log(value));
// Logs:
// "first"

subscribeToFirst = false;
firstOrSecond.subscribe((value) => console.log(value));
// Logs:
// "second"

Trong ví dụ này, khi subscribeToFirst được thiết lập là true, Observable đầu tiên sẽ được subscribe và phát ra giá trị “first”. Khi subscribeToFirst được thiết lập là false, Observable thứ hai sẽ được subscribe và phát ra giá trị “second”.