RxJS Creation Operators là các công cụ mạnh mẽ trong RxJS giúp tạo Observable một cách dễ dàng và linh hoạt hơn so với việc tạo bằng tay. Trong bài trước, chúng ta đã thấy cách tạo một Observable thủ công bằng cách sử dụng constructor. Nhưng liệu có cách nào khác không? Đây chính là nơi mà các Operators trong RxJS sẽ giúp ích cho chúng ta.

Các Operators là những hàm pure cho phép chúng ta thực hiện các thao tác functional trên Observable một cách dễ dàng và linh hoạt.

Để bắt đầu, dưới đây là đoạn mã cho observer mà chúng ta sẽ sử dụng trong các ví dụ tiếp theo:

const observer = {
  next: (val) => console.log(val),
  error: (err) => console.log(err),
  complete: () => console.log("complete"),
};

Common Creation Operators

Hàm of()

Hàm of() trong RxJS giúp tạo Observable từ hầu hết mọi giá trị: từ các kiểu dữ liệu cơ bản như số, chuỗi, tới các cấu trúc phức tạp như mảng, đối tượng, hoặc cả hàm. Khi sử dụng of(), Observable sẽ emit các giá trị được truyền vào và sau đó hoàn thành ngay sau khi tất cả các giá trị đó đã được phát đi.

Ví dụ với giá trị cơ bản:

// output: 'hello'
// complete: 'complete'
of("hello").subscribe(observer);

Hoặc với mảng hoặc đối tượng:

// output: [1, 2, 3]
// complete: 'complete'
of([1, 2, 3]).subscribe(observer);

Bạn cũng có thể truyền nhiều giá trị vào of() để tạo một chuỗi giá trị sẽ được emit:

// output: 1, 2, 3, 'hello', 'world', {foo: 'bar'}, [4, 5, 6]
// complete: 'complete'
of(1, 2, 3, "hello", "world", { foo: "bar" }, [4, 5, 6]).subscribe(observer);

Khi bạn sử dụng of(), Observable sẽ phát đi các giá trị này tuần tự và hoàn thành sau khi phát hết tất cả các giá trị đã được truyền vào.

Hàm from()

Hàm from() trong RxJS tương tự như of(), nhưng khác biệt ở chỗ from() nhận vào một Iterable hoặc một Promise để tạo Observable.

Iterable là bất kỳ giá trị nào bạn có thể duyệt qua như mảng (array), map, set hoặc thậm chí là chuỗi (string). Khi duyệt qua chuỗi, bạn sẽ nhận được từng ký tự một.

• Ví dụ với Array:

// output: 1, 2, 3
// complete: 'complete'
from([1, 2, 3]).subscribe(observer);

Khi nhận vào một mảng, from() sẽ phát đi các giá trị trong mảng theo trình tự. Điều này tương đương với of(1, 2, 3).

• Hoặc với String:

// output: 'h', 'e', 'l', 'l', 'o', ' ', 'w', 'o', 'r', 'l', 'd'
// complete: 'complete'
from("hello world").subscribe(observer);

• from() cũng có thể nhận các cấu trúc dữ liệu như Map hoặc Set:

const map = new Map();
map.set(1, "hello");
map.set(2, "bye");
// output: [1, 'hello'], [2, 'bye']
// complete: 'complete'
from(map).subscribe(observer);

const set = new Set();
set.add(1);
set.add(2);
// output: 1, 2
// complete: 'complete'
from(set).subscribe(observer);

Ngoài ra, from() có khả năng chuyển đổi Promise thành Observable:

// output: 'hello world'
// complete: 'complete'
from(Promise.resolve("hello world")).subscribe(observer);

Với Promise, from() sẽ giải quyết (resolve) Promise và trả về giá trị đã được giải quyết. Điều này cho phép bạn chuyển đổi Promise thành Observable.

Hàm fromEvent()

Hàm fromEvent() trong RxJS cho phép chuyển đổi các sự kiện (Events) như click chuột hoặc nhập input thành Observable.

Ví dụ, khi chúng ta có các sự kiện DOM như click chuột hoặc bàn phím:

const btn = document.querySelector("#btn");
const input = document.querySelector("#input");

// output (ví dụ): MouseEvent {...}
// complete: không có gì log.
fromEvent(btn, "click").subscribe(observer);

// output (ví dụ): KeyboardEvent {...}
// complete: không có gì log.
fromEvent(input, "keydown").subscribe(observer);

Điều quan trọng cần lưu ý là fromEvent() tạo ra một Observable không tự động kết thúc (complete) sau khi các sự kiện xảy ra. Điều này là hoàn toàn hợp lý vì chúng ta thường muốn tiếp tục lắng nghe các sự kiện như click hay keydown cho đến khi chúng ta không cần thiết nữa. fromEvent() không thể tự động nhận biết được khi nào chúng ta không muốn lắng nghe các sự kiện nữa. Điều này cũng đồng nghĩa rằng, nếu không cần thiết nữa, bạn cần chủ động unsubscribe các Observable được tạo từ fromEvent() để tránh tràn bộ nhớ (memory leak).

Hàm fromEventPattern()

Hàm fromEventPattern() trong RxJS là một phiên bản nâng cao của fromEvent(). Dù cả hai đều tạo Observable từ sự kiện, nhưng cách sử dụng và cách xử lý loại sự kiện khác biệt nhau.

Ví dụ với sự kiện click:

// Sử dụng fromEvent() như đã biết
// output: MouseEvent {...}
fromEvent(btn, "click").subscribe(observer);

// Sử dụng fromEventPattern()
// output: MouseEvent {...}
fromEventPattern(
  (handler) => {
    btn.addEventListener("click", handler);
  },
  (handler) => {
    btn.removeEventListener("click", handler);
  }
).subscribe(observer);

Ví dụ khác: Lấy tọa độ của con trỏ khi click vào một phần tử:

// Sử dụng fromEvent() với xử lý map
// output: 10 10
fromEvent(btn, "click")
  .pipe(map((ev: MouseEvent) => ev.offsetX + " " + ev.offsetY))
  .subscribe(observer);

// Sử dụng fromEventPattern() với các hàm riêng
function addHandler(handler) {
  btn.addEventListener("click", handler);
}
function removeHandler(handler) {
  btn.removeEventListener("click", handler);
}
// output: 10 10
fromEventPattern(
  addHandler,
  removeHandler,
  (ev: MouseEvent) => ev.offsetX + " " + ev.offsetY
).subscribe(observer);

fromEventPattern() chấp nhận 3 tham số: addHandler, removeHandler, và projectFunction (optional). Nó cung cấp một cách để chuyển đổi các sự kiện từ API gốc sang Observable. Ví dụ, bạn có thể sử dụng nó để xử lý các sự kiện phức tạp hơn từ SignalR Hub hoặc các API khác. Trong ví dụ cuối cùng, chúng ta thấy cách sử dụng fromEventPattern() để kết nối với SignalR Hub thông qua websocket và chuyển đổi các sự kiện thành Observable.

Hàm interval()

Hàm interval() trong RxJS tạo ra một Observable để phát ra các số nguyên từ 0 theo một chu kỳ cố định, tương tự như setInterval().

// output: 0, 1, 2, 3, 4, ...
interval(1000) // phát ra giá trị mỗi giây
  .subscribe(observer);

Tương tự như fromEvent(), interval() cũng không kết thúc tự động, vì vậy bạn cần phải gọi unsubscribe để dọn dẹp khi không cần thiết nữa.

Hàm timer()

Hàm timer() trong RxJS có hai cách sử dụng:

1. Tạo một Observable sẽ phát ra giá trị sau một khoảng thời gian chờ xác định. Cách này sẽ kết thúc tự động khi phát ra giá trị.

// output: sau 1 giây -> 0
// complete: 'complete'
timer(1000).subscribe(observer);

2. Tạo một Observable sẽ phát ra giá trị sau một khoảng thời gian chờ xác định và sau đó sẽ phát ra giá trị với một chu kỳ cố định. Tương tự như interval(), nhưng timer() hỗ trợ delay trước khi bắt đầu phát ra giá trị. Cách này cũng không kết thúc tự động.

// output: sau 1 giây -> 0, 1, 2, 3, 4, 5 ...
timer(1000, 1000).subscribe(observer);

Hàm throwError()

Hàm throwError() trong RxJS tạo ra một Observable không phát ra giá trị, thay vào đó, ngay khi được subscribe, nó sẽ gây ra một lỗi ngay lập tức.

// error: 'an error'
throwError("an error").subscribe(observer);

throwError() thường được sử dụng để xử lý lỗi trong các Observable. Sau khi xử lý lỗi, khi chúng ta muốn tiếp tục thông báo lỗi cho một ErrorHandler hoặc một phần xử lý lỗi khác, chúng ta có thể sử dụng throwError. Trong các trường hợp làm việc với Observable và các operators yêu cầu một Observable (như switchMap, catchError), việc throwError trả về một Observable là cực kỳ hữu ích.

Hàm defer()

Hàm defer() trong RxJS có một đặc điểm đặc biệt: nó nhận vào một ObservableFactory và mỗi khi có một Subscriber mới, nó sẽ sử dụng ObservableFactory này để tạo ra một Observable mới. Điều này tạo ra sự khác biệt lớn so với các hàm tạo Observable khác như of().

// Sử dụng of()
const now$ = of(Math.random());
// Kết quả giống nhau cho cả 3 lần subscribe
now$.subscribe(observer);
now$.subscribe(observer);
now$.subscribe(observer);

Nhưng khi sử dụng defer(), mỗi lần subscribe sẽ tạo ra một giá trị mới.

const now$ = defer(() => of(Math.random()));
// Mỗi lần subscribe, giá trị được tạo mới
now$.subscribe(observer);
now$.subscribe(observer);
now$.subscribe(observer);

Việc này có ích như thế nào? Giả sử chúng ta cần thử lại một Observable sau mỗi lần lỗi xảy ra, và cần sử dụng một giá trị ngẫu nhiên để quyết định việc thử lại. Trong trường hợp này, defer() kết hợp với retry() sẽ giúp chúng ta rất hiệu quả.