forkJoin -> Gộp các steam chạy đồng thời, và emit lại giá trị khi tất cả complete, nếu 1 stream bị lỗi thì nó sẽ ko emit, hoặc 1 stream ko complete thì forkJoin không emit giá trị nào cả (nghĩa là nó forkJoin sẽ ko completed).

combineLatest -> Khi 1 stream emit giá trị thì lúc đó nó sẽ lấy giá trị cuối cùng của stream đó và tất cả giá trị cũ của steam còn lại, ban đầu sẽ đợi all stream emit ít nhất 1 lần, sau đó sẽ tính toán những lần emit của các stream (mỗi lần emit là 1 lần trả về result) -> Khi tất cả đã từng 1 lần emit thì lần sau chỉ cần 1 thằng complete thì nó sẽ trả về, kể cả khi thằng kia chưa complete thì nó sẽ lấy giá trị cũ

→ Sẽ completed khi all stream bên trong completed

+ TH mà trong combineLatest có 1 stream ko emit gì, sẽ ko chạy

~ TH lúc đầu ko run vì đợi all complete 1 lần, do dùng this.searchInput.valueChanges -> start sẽ ko run

merge() -> Cho phép các stream chạy đồng thời, thằng nào complete thì nó log

concat() -> tại 1 thời điểm chỉ cho phép 1 thằng stream chạy, chạy xong từng steam nó sẽ emit dữ liệu

zip() -> gán Theo cặp, khi chạy thì nó sẽ ghép dữ liệu theo cặp, khi 1 stream chưa emit thì nó sẽ đợi

race(obs1, obs2).sub... → Chỉ emit và subscibe vào 1 trong 2, cái nào emit trc thì nó sẽ subscribe vào

*** debounceTime(time) | timer(time), delay(time)

+ debounceTime: trì hoãn giá trị emitted (phát ra) trong 1 time nhất định, nếu nguồn phát ra 1 giá trị mới, thì value cũ sẽ bị loại bỏ, time được đặt lại. -> giá trị trong 1 thời gian nhất định sẽ là mới nhất

+ timer (2 type)

-> Tạo an observable được đợi sau 1 khoảng time trước khi nó tạo ra số 0 ~ tạo ra độ trễ sau 1 time.

-> initialDelay::

Syntax ~ timer(due, scheduler)

-> due ~ Là khoảng time độ trễ tra khi emit | scheduler ~ Default is underfined. The scheduler to use to schedule the delay, default to

asyncScheduler⁠

-> return Observable<0> ~ sử dụng để phát ra noti sau khoảng time trì hoãn

Syntax ~ timer(startDue, intervalDuration, scheduler)

-> startDue ~ Là time chờ tra khi start khoảng time. | intervalDuration ~ là độ trễ giữa mỗi value emit trong time

-> return Observable<number>

*** Một số operators cần lưu ý

+ distinctUntilChanged: chỉ emit khi giá trị hiện tại khác với giá trị cuối cùng (trc đó)

-> Thường dùng trong valueChanges của formControl để có cho emit giá trị đi tiếp vào subscribe ko!

-> Nó dùng với phép so sánh nghiêm ngặt (===)

Automatically unsubscribe your RxJs Observables ❤️

+ unsubscribe when ngOnDestroy ~ Vì ngOnDestroy sẽ được gọi khi 1 thành phần được xoá khỏi DOM

~> Có nhiều solution resolve better::

+ Để ko thực hiện unsubscribe là ta ko subscribe nó ngay từ đầu ~> using the async pipe ~ automate subscribe / get data / unsubscribe

Syntax: < [dataSource]=“data$ | async”>

+ Đồng thời sử dụng pipe async và muốn transform data ~> gán cho 1 variable

Syntax: filterEntries$: Observable<any> = of().pipe(tap(data => //Do something transform))

+ Sử dụng subSink để gộp nhiều subscription

Syntax:

~ Create class adapter unsubscribe destroy (1)

~ Create new SubSink()

~ Use method ngOnDestroy

~ Component use -> extends (1), in constructor ~ use super(); & use subsink

Ex::

export class UnsubscribeDestroyAdapter implement OnDestroy {

sub = new SubSink();

ngOnDestroy(){ this.sub.unsubscribe();}

}

~> Use: export class comp extends UnsubscribeDestroyAdapter {

constructor(){

super();

this.sub.sink = someObservable.subscribe(…) ~ add single

This.sub.add(someObs.subscribe(), someObs.subscribe()); ~ add multiple

}

+ Có cách tương tự như trên kết hợp với takeUntil

Ex::

@Injectable()

export class TService extends Subject<boolean> implement OnDestroy {

ngOnDestroy(){

this.next();

this.complete();

}

Use ~>

Defined providers: [TService]

constructor(private tService: TService){}

observable.pipe(takeUntil(this.tService)).subscribe(…);

Break down my code::

(Tư tưởng là khi component destroy thì service này sẽ destroy, kết hợp với takeUntil ~ để không phải viết nhiều lần)

~ Tạo service @Injectable(),

~ Vì không có providerIn: ‘root’ nên là cần defined providers: [TService] trong Comp or Module để use

~ Class service extends Subject<boolean> ~ class service là 1 dạng subject

~ Implement OnDestroy để use method ngOnDestroy

~ Comp use: dùng với takeUntil()

~ Lúc này khi Comp destroy thì service sẽ destroy, và hàm ngOnDestroy của service sẽ run

~ Service được hiểu là subject và emit giá trị

~ TakeUntil sẽ nhận được giá trị bắn ra, và thực hiện đóng theo nguyên tắc của takeUntil

-> Cơ bản cách trên giống cách thông thường,

vẫn defined 1 Subject(), sau đó run hàm ngOnDestroy() và bắn vào subject 1 giá trị

+ Kết hợp với takeUntil() để huỷ stream đó

Ex::

destroy$ = new Subject();

observable.pipe( takeUntil(this.destroy$) ).subscribe(…)

ngOnDestroy(){

this.destroy$.next();

This.destroy$.complete();

}

—————————————————————————————————————

HIGHER ORDER OBSERVABLES (HOOs)❤️

Mỗi khi 1 observable trả về 1 obs thì t có cách sử dụng bên dưới.

Khi Subscribe-in-Subscribe thì khá tệ, dẫn đến việc hiện thị sai dữ liệu thì chúng ta có cách áp dụng cách dưới

VD: ta có 1 select ngân hàng, 1 select chi nhánh, thì mỗi lần mình chọn ngân hàng thì api call chi nhánh, cách xử lý đó là dùng switchMap, nó sẽ calcel api call chi nhánh trước đó. Done

mergeMap -> Nó nhảy vào bên trong, có bao nhiêu số obs thì nó chạy bấy nhiêu lần

mergeMap(truyền) -> chạy theo index lần nào

concatMap -> Giống mergeAll nhưng nó chạy theo thứ tự, nó chỉ xử lý chạy trong concatmap(()=> {})

-> VD: khi mà mình muốn upload ảnh lên trên form, thì submit sẽ call api create

-> from(abc).pipe(concatMap(() => interval(1000).pipe(take(1))))

.subscribe(

res => {console.log(res);},

error => console.log(error),

() => console.log('abc')

);

switchMap -> mỗi lần nó nhận 1 obs mới nó sẽ cancel obs hiện tại.

→ Khi switchMap nhận 1 obs mới thì nó sẽ cancel obs đang chạy đi và subcribe vào obs mới.

→ VD: this.queryInput.valueChanges

.pipe(

debounceTime(500),

switchMap((query) => this.apiService.filterData(query))

)

.subscribe((data) => {

/*...*/

});

—> Mỗi khi click sẽ subscribe emit(2) và nó sẽ unsubscribe emit(cũ)

fromEvent(document, 'click').pipe(

switchMap(() => interval(1000).pipe(take(10)))

);

—————————————————————————————————————

SUBJECT❤️

Là 1 dạng class bên trong của rxjs và cũng là 1 observer

Khi mà subject có giá trị thì nó next cho Những thằng observer khác

Nó cũng có thể bắn giá trị vào chính stream của chính nó như next() và nó tự subcriber chính nó (đây là cái hay)

Thường sẽ là private readonly subject (Chính vì subject tự truyền giáo trị cho chính nó thế nên là cần sét cho nó private rồi nếu muốn thằng khác dùng thì chỉ cần <this.subject.asObservable()> để gán nó là observable

*** Subject

Khi maf next() bắn dữ liệu mà thằng sau subcribe thì nó sẽ không có dữ liệu (Vì khi thằng subject bắn dữ liệu thì nó ko lưu bất cứ data nào cả trên stream của nó)

*** BehaviorSubject

Cần truyền giá trị ban đầu

Nó khác với Subject là: nó lưu trữ giá trị cuối cùng

*** ReplaySubject

Nó giống với BehaviorSubject

Truyền giá trị số lấy bao nhiêu giá trị cuối cùng

*** ShareReplay

Nó cache lại dữ liệu cuối cùng

Want to print your doc?
This is not the way.

Try clicking the ⋯ next to your doc name or using a keyboard shortcut (

CtrlP

) instead.