RxJS Combination Operators trong RxJS

27-12-2023
Toanngo92
0 Comments

`forkJoin()` trong RxJS

ForkJoin(): ForkJoin giống như Promise.all() trong JavaScript. Nó nhận vào một danh sách các Observables và đợi cho đến khi tất cả các Observable trong danh sách complete. Khi tất cả hoàn thành, nó sẽ emit giá trị của từng Observable theo dạng mảng hoặc đối tượng (tùy thuộc vào cách bạn truyền tham số).

forkJoin([of(1), of("hello"), of({ foo: "bar" })]).subscribe(observer);
// Output: [1, 'hello', {foo: 'bar'}]
// Output: 'complete'

Bạn cũng có thể truyền một đối tượng (Object) với các Observables:

forkJoin({ one: of(1), hello: of("hello"), foo: of({ foo: "bar" }) }).subscribe(
  observer
);
/**
 * Output:
 * {
 *   one: 1,
 *   hello: 'hello',
 *   foo: { foo: 'bar' }
 * }
 * Output: 'complete'
 */

Lưu ý: ForkJoin chỉ emit khi tất cả các Observables con hoàn thành. Nếu một trong số chúng không hoàn thành, ForkJoin sẽ không bao giờ emit.

Ngoài ra, nếu bất kỳ Observable con nào trong danh sách gặp lỗi, ForkJoin sẽ ném ra lỗi và các giá trị của các Observable khác đã hoàn thành cũng sẽ bị bỏ qua nếu không xử lý lỗi một cách thích hợp.

Một số use case

ForkJoin và Ứng dụng trong Angular:

Khi làm việc với Angular và cần lấy dữ liệu đồng thời từ nhiều dropdown hoặc select box, forkJoin() trở thành một công cụ hữu ích. Chẳng hạn, khi cần lấy dữ liệu cho các dropdown liên quan đến tài khoản, phòng ban và cửa hàng, forkJoin() giúp thực hiện các requests này cùng lúc.

forkJoin([
  this.apiService.getAccountDropdown(),
  this.apiService.getDepartmentDropdown(),
  this.apiService.getStoreDropdown(),
]).subscribe(observer);
// Output: [accountList, departmentList, storeList]
// Output: 'complete'

Nếu bạn muốn xử lý dữ liệu trả về từ các requests để chuyển đổi chúng thành một cấu trúc dễ sử dụng hơn, bạn có thể sử dụng tham số thứ hai trong forkJoin() là projectFunction. Điều này cho phép bạn thực hiện biến đổi dữ liệu trước khi nó được emit.

forkJoin(
  [
    this.apiService.getAccountDropdown(),
    this.apiService.getDepartmentDropdown(),
    this.apiService.getStoreDropdown(),
  ],
  (accountList, departmentList, storeList) => {
    return {
      accounts: accountList,
      departments: departmentList,
      stores: storeList,
    };
  }
).subscribe(observer);
// Output: { accounts: [...], departments: [...], stores: [...] }
// Output: 'complete'

Lưu ý: ForkJoin trong Angular rất hữu ích khi bạn cần gộp các kết quả từ nhiều requests thành một cấu trúc dễ quản lý hơn trước khi sử dụng chúng trong giao diện người dùng.

`combineLatest()` trong RxJS

CombineLatest Trong RxJS:

Sử Dụng Phổ Biến Trong Angular:

combineLatest() trong RxJS là một công cụ phổ biến trong Angular khi cần kết hợp dữ liệu từ nhiều nguồn khác nhau đồng thời.

combineLatest([
  interval(2000).pipe(map((x) => `First: ${x}`)),
  interval(1000).pipe(map((x) => `Second: ${x}`)),
  interval(3000).pipe(map((x) => `Third: ${x}`)),
]).subscribe(observer);

Khi một trong các Observable con trong combineLatest() phát ra một giá trị mới nhất, nó sẽ kết hợp giá trị này với các giá trị gần nhất từ các Observable con khác để emit một mảng mới.

Ví Dụ Về Hoạt Động:

Sau 3 giây, vì Observable thứ ba có thời gian dài nhất (3s), mảng được emit là [First 0, Second 2, Third 0].
Sau 1 giây tiếp theo, mảng sẽ là [First 1, Second 2, Third 0], sau đó [First 1, Second 3, Third 0], và tiếp tục như vậy.

Lưu Ý:

Sau khi các Observable con phát ra giá trị đầu tiên, combineLatest() sẽ kết hợp giá trị mới nhất từ Observable hiện tại và giá trị gần nhất từ các Observable khác để emit.
Điều này có thể dẫn đến việc một Observable bị nuốt mất giá trị nếu nó phát ra các giá trị nhanh hơn các Observable khác.
combineLatest() chỉ complete khi tất cả các Observable con complete.
Nếu một trong số các Observable con không bao giờ complete, combineLatest() sẽ không bao giờ complete.
Nếu một Observable con throw error, combineLatest() sẽ throw error và giá trị của các Observable con đã emit trước đó sẽ bị mất đi.

Với việc hiểu rõ về cách combineLatest() hoạt động, bạn có thể tận dụng nó để kết hợp dữ liệu từ các nguồn khác nhau một cách linh hoạt trong ứng dụng Angular.

Một số use case

Một Số Sử Dụng Phổ Biến:

Trong Angular, khi làm việc với Service và cần kết hợp nhiều trạng thái (state), combineLatest() từ RxJS là công cụ lựa chọn phổ biến. Với tính chất không kết thúc sau mỗi lần phát ra giá trị, nó hoạt động tốt khi kết hợp với AsyncPipe trong template.

this.vm$ = combineLatest([
  this.paginationService.currentPage$,
  this.paginationService.currentSize$,
  this.paginationService.totalCount$,
  this.paginationService.currentOffset$,
]).pipe(
  map((currentPage, currentSize, totalCount, currentOffset) => {
    return {
      currentPage,
      currentSize,
      totalCount,
      currentOffset,
    };
  })
);

Ví dụ trên liên quan đến việc xử lý PaginationComponent trong Angular. Khi updateSize() và updatePage() được gọi, currentPage$ và currentSize$ sẽ phát ra giá trị mới. Điều này khiến combineLatest() (biến vm$) phát ra giá trị mới, làm cập nhật template thông qua vm$ | async.

onSizeChanged(newSize: number) {
  this.paginationService.updateSize(newSize);
}

onPageChanged(newPage: number) {
  this.paginationService.updatePage(newPage);
}

Trong combineLatest(), nếu truyền vào một mảng Array<Observable>, bạn cũng có thể kết hợp với một hàm projectFunction. Hàm này sẽ được gọi với các giá trị của các Observable con và kết quả của nó sẽ là giá trị phát ra của combineLatest(). Việc này giống với tính nhất quán của forkJoin() và khả năng sử dụng projectFunction.

Dưới đây là ví dụ viết lại biến vm$ ở trên sử dụng projectFunction:

this.vm$ = combineLatest(
  [
    this.paginationService.currentPage$,
    this.paginationService.currentSize$,
    this.paginationService.totalCount$,
    this.paginationService.currentOffset$,
  ],
  (currentPage, currentSize, totalCount, currentOffset) => {
    return {
      currentPage,
      currentSize,
      totalCount,
      currentOffset,
    };
  }
);

Như vậy, combineLatest() là một công cụ mạnh mẽ để kết hợp và quản lý các trạng thái trong ứng dụng Angular, đặc biệt khi bạn cần theo dõi và kết hợp nhiều trạng thái đồng thời.

`zip()` trong RxJS

Sử Dụng Zip trong RxJS

Operator zip() trong RxJS nhận vào một danh sách các Observables hoặc Functions để biểu diễn các child Observable. Nó là một công cụ mạnh mẽ vì có khả năng gom tất cả các giá trị emit của các child Observable theo từng cặp.

Giả sử có ba Observable: obs1 phát ra giá trị 1 – 2 – 3, obs2 phát ra giá trị 4 – 5 – 6, và obs3 phát ra giá trị 7 – 8 – 9. Nếu sử dụng combineLatest(), các giá trị sẽ được kết hợp như sau:

combineLatest(of(1, 2, 3), of(4, 5, 6), of(7, 8, 9)).subscribe(observer);
// [1, 4, 7],
// [2, 4, 7],
// ...
// [3, 6, 9]

Tuy nhiên, khi sử dụng zip() với ví dụ trên:

zip(of(1, 2, 3), of(4, 5, 6), of(7, 8, 9)).subscribe(observer);
// [1, 4, 7];
// [2, 5, 8];
// [3, 6, 9];

zip() sẽ hoàn thành khi bất kỳ child Observable nào complete. Điều này có nghĩa là bạn sẽ nhận được kết quả ghép lại từ zip, và giá trị nào mà các Observables không phát ra trước khi complete sẽ bị bỏ qua.

zip(of(1, 2, 3, 99), of(4, 5, 6), of(7, 8, 9)).subscribe(observer);
// [1, 4, 7];
// [2, 5, 8];
// [3, 6, 9];
// Giá trị 99 của observable đầu tiên bị bỏ qua vì observable thứ 2 và thứ 3 đã complete trước.

Nếu bất kỳ Observable nào trong các child Observables emit lỗi, zip() sẽ throw error.

Nếu tham số cuối cùng của zip() là một Function, nó sẽ được xem như là một projectFunction. Cách hoạt động của nó tương tự như projectFunction của combineLatest() và forkJoin().

Một số use case

Sử Dụng Zip Trong RxJS

Operator zip() trong RxJS rất hữu ích trong một số tình huống cụ thể:

Gom các giá trị từ nhiều Observables

Nếu bạn cần lấy các giá trị cuối cùng từ nhiều Observables khác nhau, zip() là lựa chọn hoàn hảo. Ví dụ:

const age$ = of<number>(29, 28, 30);
const name$ = of<string>("Chau", "Trung", "Tiep");
const isAdmin$ = of<boolean>(true, false, true);

zip(age$, name$, isAdmin$).pipe(
  map(([age, name, isAdmin]) => ({ age, name, isAdmin }))
).subscribe(observer);

Kết quả sẽ gom các giá trị tương ứng từ mỗi Observable thành một object:

// { age: 29, name: 'Chau', isAdmin: true }
// { age: 28, name: 'Trung', isAdmin: false }
// { age: 30, name: 'Tiep', isAdmin: true }

Có thể sử dụng zip() cùng với projectFunction để có kết quả tương tự:

zip(age$, name$, isAdmin$, (age, name, isAdmin) => ({
  age,
  name,
  isAdmin,
})).subscribe(observer);

Kết hợp giá trị từ Observables tại các thời điểm khác nhau

Nếu bạn muốn lấy giá trị của chuột khi một sự kiện xảy ra sau sự kiện khác (ví dụ như khi mousedown và mouseup xảy ra), zip() cũng hữu ích:

const log = (event, val) => `${event}: ${JSON.stringify(val)}`;
const getCoords = map((e: MouseEvent) => ({ x: e.clientX, y: e.clientY }));

const documentEvent = (eventName) =>
  fromEvent(document, eventName).pipe(getCoords);

zip(documentEvent("mousedown"), documentEvent("mouseup")).subscribe((e) =>
  console.log(`${log("start", e[0])} ${log("end", e[1])}`)
);

Kết quả sẽ gộp giá trị tại mỗi thời điểm event xảy ra:

// start: {"x":291,"y":136} end: {"x":143,"y":168}
// start: {"x":33,"y":284} end: {"x":503,"y":74}

zip() cho phép bạn kết hợp giữa các giá trị từ các Observables khác nhau vào những thời điểm khác nhau, tạo ra một kết quả tổng hợp.

concat() trong RxJS

Operator concat() trong RxJS là một công cụ hữu ích để quản lý việc gộp các Observable lại với nhau một cách tuần tự:

Chờ Observable hiện tại complete: concat() theo dõi tuần tự các Observable được truyền vào. Nó chỉ chuyển sang Observable tiếp theo khi Observable hiện tại đã hoàn thành (complete).
Xử lý khi Observable phát ra giá trị hoặc complete: Nếu Observable hiện tại phát ra giá trị và complete, concat() sẽ gửi giá trị đó đi và chuyển sang Observable kế tiếp. Nếu Observable hiện tại complete mà không phát ra giá trị, concat() sẽ bỏ qua và chuyển sang Observable tiếp theo.
Xử lý lỗi: Nếu Observable gây ra lỗi, concat() sẽ kết thúc ngay lập tức và bỏ qua các Observable sau đó.

Ví dụ, nếu bạn có hai Observable, một đợi 1 giây rồi phát ra giá trị từ 4 đến 6 và một phát ra giá trị từ 1 đến 3, concat() sẽ chờ Observable đầu tiên hoàn thành trước khi bắt đầu theo sau bởi Observable thứ hai:

concat(of(4, 5, 6).pipe(delay(1000)), of(1, 2, 3)).subscribe(observer);
// output:
// sau 1s:
// 4-5-6-1-2-3
// output: 'complete'

Cũng có thể sử dụng concat() để lặp lại một Observable nhiều lần hoặc kết hợp với repeat():

const fiveSecondTimer = interval(1000).pipe(take(5));

// Lặp lại cùng một Observable
concat(fiveSecondTimer, fiveSecondTimer, fiveSecondTimer).subscribe(observer);
// output: 0,1,2,3,4 - 0,1,2,3,4 - 0,1,2,3,4
// output: 'complete'

// Sử dụng repeat()
concat(fiveSecondTimer.pipe(repeat(3))).subscribe(observer);
// output: 0,1,2,3,4 - 0,1,2,3,4 - 0,1,2,3,4
// output: 'complete'

Với concat(), việc quản lý Observable theo thứ tự và xử lý các giá trị emit từng bước một giúp bạn có thể kiểm soát dễ dàng hơn trong việc sử dụng chuỗi các Observable.

`merge()` trong RxJS

Operator merge() trong RxJS là một công cụ linh hoạt cho phép gộp các Observable lại với nhau mà không quan tâm đến thứ tự của chúng:

Không quan trọng thứ tự emit của Observable: merge() không giữ các Observable theo thứ tự như concat(). Thay vào đó, nó emit các giá trị ngay khi các Observable con emit, bất kể thứ tự chúng được truyền vào.
Quản lý số lượng Observable đồng thời: Bạn có thể xác định số lượng Observable mà merge() sẽ đồng thời subscribe (concurrently) thông qua tham số cuối cùng. Mặc định, nó sẽ subscribe đồng thời vào tất cả các Observable.
Xử lý giá trị và kết thúc: merge() sẽ emit các giá trị từ bất kỳ Observable nào emit, và nó sẽ kết thúc (complete) khi và chỉ khi tất cả các Observable con đã kết thúc.

Ví dụ, nếu bạn có hai Observable, một đợi 1 giây rồi phát ra giá trị từ 4 đến 6 và một phát ra giá trị từ 1 đến 3, merge() sẽ emit giá trị từ cả hai Observable ngay khi chúng phát ra giá trị:

merge(of(4, 5, 6).pipe(delay(1000)), of(1, 2, 3)).subscribe(observer);
// output:
// 1,2,3
// sau 1s: 4,5,6
// output: 'complete'

Với merge(), bạn cũng có thể quản lý số lượng Observable đồng thời bằng cách xác định concurrent (số lượng đồng thời) thông qua tham số cuối cùng:

merge(
  interval(1000).pipe(mapTo("first"), take(5)),
  interval(2000).pipe(mapTo("second"), take(3)),
  interval(3000).pipe(mapTo("third"), take(2)),
  2
).subscribe(observer);
// output:
// (sau 1s):
// first
// (sau 2s):
// first
// second
// (sau 3s):
// first
// (sau 4s):
// first
// second
// (sau 5s):
// first
// (sau 6s):
// second
// (sau 8s):
// third
// (sau 11s):
// third
// output: 'complete'

Ở ví dụ trên, merge() sẽ đồng thời subscribe vào hai Observable đầu tiên (first và second). Khi first hoàn thành, third mới được merge() subscribe tiếp theo. Điều này cho thấy concat() thực sự chính là merge() với concurrent được xác định là 1.

Một số use case

Trong Angular, merge() có thể được sử dụng khi bạn làm việc với FormGroup và muốn theo dõi thay đổi của từng FormControl.valueChanges để xử lý các hành động phù hợp. Bạn không cần phải quan tâm đến thứ tự nào FormControl sẽ thay đổi, chỉ cần xử lý khi mà giá trị của FormControl đó thay đổi.

Ví dụ, nếu bạn có một FormGroup chứa các trường như firstName và lastName, bạn có thể theo dõi các thay đổi của từng trường này:

const formControlValueChanges = Object.keys(this.formGroup.value).map((key) =>
  this.formGroup.get(key).valueChanges.pipe(map((value) => ({ key, value })))
); // Ví dụ từ {firstName: 'Chau', lastName: 'Tran'} -> [Observable<{key, value}>, Observable<{key, value}>]

merge(...formControlValueChanges).subscribe(({ key, value }) => {
  if (key === 'firstName') {
    // Xử lý khi giá trị của trường 'firstName' thay đổi
    // ...
  }
  if (key === 'lastName') {
    // Xử lý khi giá trị của trường 'lastName' thay đổi
    // ...
  }
});

Khi bất kỳ trường nào trong FormGroup thay đổi giá trị, merge() sẽ theo dõi tất cả các FormControl và thông báo về các thay đổi này. Điều này cho phép bạn xử lý mỗi thay đổi tùy thuộc vào trường cụ thể nào được thay đổi mà không cần phải quan tâm đến thứ tự của chúng.

race()

race() là một công cụ quan trọng trong RxJS với khả năng chọn lựa giữa các Observable để lấy giá trị nhanh nhất. Bạn có thể so sánh nó như cuộc đua giữa các Observable để xem cái nào nhanh nhất sẽ thắng.

Khi bạn sử dụng race(), nó sẽ quan sát tất cả các Observable và chỉ emit giá trị của Observable nhanh nhất – ngay khi nó emit ra giá trị đầu tiên. Nó sẽ tiếp tục quan sát cho đến khi một trong số các Observable hoàn thành (complete).

Nếu Observable nhanh nhất gặp lỗi thay vì emit giá trị, race() sẽ tức thì thông báo về lỗi đó mà không chờ các Observable khác.

Ví dụ:

race(
  interval(1000).pipe(mapTo("fast")),
  interval(2000).pipe(mapTo("medium")),
  interval(3000).pipe(mapTo("slow"))
).subscribe(observer);
// output: fast - 1s -> fast - 1s -> fast - 1s -> fast...

Ở đây, race() sẽ chọn Observable nhanh nhất là của interval(1000) và emit giá trị của nó mỗi giây. Các Observable khác không còn được quan tâm sau khi Observable interval(1000) đã bắt đầu emit giá trị.

Một số use case

Ở một ứng dụng, việc hiển thị các thông báo hoặc banner dựa trên hành động của người dùng là điều phổ biến. Ví dụ, sau khi người dùng gửi form, bạn muốn hiển thị một banner thông báo kết quả của việc gửi đó, và bạn muốn banner đó tự đóng sau một khoảng thời gian nhất định hoặc khi người dùng click để đóng hoặc khi họ chuyển sang một trang khác.

Để làm điều này, một trong những operator hữu ích là race(), vì bạn chỉ muốn đóng banner khi một trong ba điều kiện sau xảy ra:

Sau khi banner được hiển thị trong 10 giây.
Người dùng click để đóng banner.
Người dùng chuyển sang một trang khác.

Dưới đây là cách bạn có thể sử dụng race() cùng với các sự kiện tương ứng để quản lý việc đóng banner:

race(
  timer(10000), // Thời gian 10 giây
  this.userClick$, // Sự kiện click của người dùng
  this.componentDestroy$ // Sự kiện destroy component khi người dùng chuyển trang
)
  .pipe(takeUntil(this.componentDestroy$)) // Ngừng lắng nghe nếu component được destroy
  .subscribe(() => this.closeBanner());

`withLatestFrom()` trong RxJS

withLatestFrom() trong RxJS là một operator rất hữu ích. Nó nhận vào một Observable khác làm tham số và kết hợp giá trị gần nhất của Outer Observable với giá trị mới nhất của tham số Observable để tạo thành một mảng, sau đó emit mảng này.

Ví dụ, khi bạn sử dụng fromEvent(document, "click") và kết hợp với interval(1000) thông qua withLatestFrom(), bạn sẽ nhận được một mảng gồm thông tin về sự kiện click và giá trị gần nhất của interval tại thời điểm click đó.

fromEvent(document, "click")
  .pipe(withLatestFrom(interval(1000)))
  .subscribe(observer);

Kết quả sẽ như sau:

Khi bạn click trước khi qua đi 1s, nó sẽ chờ đến 1s sau click để tạo mảng [MouseEvent, 0].
Khi click sau 1s, nó sẽ tạo mảng mới [MouseEvent, 0].
Và khi bạn click vào khoảng thời gian 5.5s, nó sẽ tạo mảng [MouseEvent, 4], bởi vì giá trị gần nhất của interval(1000) tại thời điểm đó là 4 (0 – 1 – 2 – 3 -4).

Ngoài ra, withLatestFrom() cũng có thể nhận vào một tham số thứ hai là một projectFunction tùy chọn, hoạt động tương tự như những projectFunction khác mà bạn đã biết trong RxJS.

Một số use case

Bạn có thể sử dụng withLatestFrom() khi cần kết hợp kết quả từ một Observable với giá trị gần nhất từ một Observable khác. Điều này hữu ích khi bạn chỉ muốn lấy giá trị mới nhất của Observable thứ hai khi Observable chính của bạn emit.

Ví dụ, khi bạn gọi một API thông qua this.apiService.getSomething() và bạn muốn kết hợp kết quả từ API với thông tin về người dùng hiện tại (this.currentLoggedInUser$), bạn có thể sử dụng withLatestFrom() để làm điều này. Mỗi khi kết quả từ API được emit, nó sẽ được kết hợp với giá trị gần nhất của Observable về người dùng đang đăng nhập.

`startWith()` trong RxJS

startWith() là một operator đơn giản. Khi áp dụng startWith() cho một Observable, nó thêm giá trị được chỉ định vào đầu tiên của chuỗi emit trước khi bắt đầu với các giá trị từ Observable gốc.

Ví dụ, nếu bạn có một Observable emit chuỗi “world”, khi bạn sử dụng startWith("Hello"), nó sẽ đưa “Hello” vào đầu tiên trước khi bắt đầu emit chuỗi “world”.

of("world").pipe(startWith("Hello")).subscribe(observer);
// output:
// 'Hello'
// 'world'
// 'complete'

Như bạn có thể thấy, “Hello” được emit ngay từ đầu trước khi “world” từ Observable được emit.

Một số use case

startWith() là một cách tiện lợi để cung cấp giá trị ban đầu cho một Observable, đặc biệt là trong việc làm việc với API calls trong Angular.

Ví dụ, khi bạn muốn lấy danh sách các sách từ một API trong Angular, và bạn muốn có một giá trị ban đầu trước khi API hoàn thành, bạn có thể sử dụng startWith([]). Khi bạn đặt startWith([]) trong Observable của API, nó sẽ đưa ra một giá trị mảng rỗng ngay từ đầu.

this.books$ = this.apiService.getBooks().pipe(startWith([]));

Sử dụng *ngIf trong Angular để render dữ liệu khi nó đã có, bạn có thể dễ dàng kiểm tra sự tồn tại của giá trị từ Observable books$.

<ng-container *ngIf="books$ | async as books">
  <!-- Nếu giá trị của books$ có sẵn (bắt đầu với mảng rỗng do startWith([])), ng-container này sẽ render nội dung bên trong -->
</ng-container>

Như vậy, startWith() là một công cụ hữu ích để đưa ra giá trị ban đầu và giúp quản lý hiển thị dữ liệu trong Angular một cách linh hoạt.

`endWith()` trong RxJS

endWith() là một operator mạnh mẽ trong RxJS, nhưng hoạt động khác với startWith(). Nó giúp chúng ta thêm các giá trị vào cuối của một Observable, nhưng giá trị này chỉ được phát ra khi Outer Observable hoàn thành (complete).

Ví dụ, khi chúng ta có một chuỗi các thông điệp, và chúng ta muốn thêm một thông điệp “goodbye!” vào cuối khi chuỗi hoàn thành:

of("hi", "how are you?", "sorry, I have to go now")
  .pipe(endWith("goodbye!"))
  .subscribe(observer);

Khi chúng ta chạy Observable này, các thông điệp “hi”, “how are you?”, và “sorry, I have to go now” sẽ được phát ra theo thứ tự. Nhưng khi Observable kết thúc, thông điệp “goodbye!” sẽ được thêm vào cuối cùng.

Điều này có nghĩa là endWith() chỉ phát ra giá trị được cung cấp trong nó sau khi toàn bộ Observable hoàn thành.

`pairwise()` trong RxJS

pairwise() trong RxJS là một công cụ thú vị, đặc biệt hữu ích khi chúng ta muốn làm việc với các cặp giá trị gần nhau từ một chuỗi giá trị. Nó giúp chúng ta gom cặp giá trị hiện tại và giá trị trước đó của Observable thành một cặp giá trị và phát ra cặp này.

Ví dụ, nếu chúng ta có một chuỗi các số từ 1 đến 5 và muốn thực hiện phép cộng giữa các cặp số liên tiếp:

from([1, 2, 3, 4, 5])
  .pipe(
    pairwise(),
    map(([prev, cur]) => prev + cur)
  )
  .subscribe(observer);

Khi chúng ta áp dụng pairwise() vào chuỗi số này, nó sẽ tạo các cặp số liên tiếp. Sau đó, thông qua map(), chúng ta thực hiện phép cộng giữa các cặp số, và kết quả sẽ được phát ra là kết quả của phép cộng giữa các cặp số liên tiếp. Với chuỗi số từ 1 đến 5, chúng ta sẽ nhận được các kết quả là 3 (1 + 2), 5 (2 + 3), 7 (3 + 4), và 9 (4 + 5).