1. Subjects là gì?

• Observable là nguồn phát. Là trái tim của RxSwift. Chịu trách nhiệm phát đi các sự kiện/giá trị cho các đối tượng đăng ký tới.

• Observer là nơi nhận. Là đầu cuối trong cả chuỗi subscription. Chịu trách nhiệm xử lý dữ liệu nhận được. Cái mà kết hợp được cả 2 Observable & Observer là Subjects. Đây chính là phần cứu cánh cho chúng ta để giải quyết được nhiều bài toán khó hơn nữa. Subject trong RxSwift hoạt động như vừa là một Observable, vừa là một Observer. Khi một Subject nhận một .next event thì ngay lập tức nó sẽ phát ra các emit cho các subscriber của nó. Trong RxSwift, chúng ta có 4 loại Subject với các cách thức hoạt động khác nhau, bao gồm:

• PublishSubject: Khởi đầu empty và chỉ emit các element mới cho Subscriber của nó.

• BehaviorSubject: Khởi đầu với một giá trị khởi tạo và sẽ relay lại element cuối cùng của chuỗi cho Subscriber mới.

• ReplaySubject: Khởi tạo với một kích thước bộ đệm cố định, sau đó sẽ lưu trữ các element gần nhất vào bộ đệm này và relay lại các element chứa trong bộ đệm cho một Subscriber mới.

• AsyncSubject: Chỉ phát ra sự kiện .next cuối cùng trong chuỗi và chỉ khi subject nhận được .completed. Cái này ít được sử dụng.

• PublishRelay & BehaviorRelay : là các subject được bọc lại (wrap), nhưng chúng chỉ chấp nhận .next. Bạn không thể thêm các .error hay .completed. Vì vậy chúng thích hợp cho các sự kiện không bao giờ kết thúc (Phần này sẽ có một bài viết riêng).

2. Publish Subjects

2.1. Khái niệm Publish Subjects

Publish Subjects được sử dụng khi bạn chỉ muốn subscribers được thông báo về các sự kiện mới từ thời điểm bạn subscribe cho đến khi hủy subscribe hoặc Subject đã chấm dứt với sự kiện khi .completed hoặc .error. Đặc điểm:

• Chỉ phát đi giá trị mới nhất
• Các Subscriber sau chỉ nhận được giá trị khi Subject phát
• Không nhận được các giá trị trước khi subscribe
• Subject kết thúc khi phát đi .completed hoặc .error
• Subscription sẽ kết thúc khi nó .dispose()

2.2. Hoạt động

Create Publish Subjects

let subject = PublishSubject<String>()

Đối tượng subject thuộc class PublishSubject. Và bạn cần phải cung cấp kiểu dữ liệu cho các phần tử được Subject phát đi. Trong ví dụ, nó là String. Ngoài ra, thì không cần phải thêm bất kì tham số nữa trong lúc khởi tạo Subject.

Emit Data

subject.onNext("1")

Với .onNext là toán tử giúp bạn emit dữ liệu đi. Bạn có thể emit bất cứ khi nào và bất cử ở đâu, mà không phụ thuộc vào dữ liệu ban đầu. Đây là ưu điểm mà Subject hơn Observable.

Subscribe

 let subscription1 = subject .subscribe(onNext: { value in print("Sub 1: ", value) })

Bạn có thể thực thi đoạn code vừa hoàn thành thì sẽ thấy. subscription1 sẽ không nhận được gì. Nguyên nhân, là đã subscribe sau khi subject phát đi 1. Cùng thử với đoạn code đầy đủ:

 let subject = PublishSubject<String>() subject.onNext("1") let subscription1 = subject .subscribe(onNext: { value in print("Sub 1: ", value) }) subject.onNext("2") let subscription2 = subject .subscribe(onNext: { value in print("Sub 2: ", value) }) subject.onNext("3") subject.onNext("4") subject.onNext("5")

Khi thực thi, bạn sẽ thấy:

Giá trị 1 sẽ không nhận được, vì trước thời điểm phát đó, không có subscribe nào tới subject. subscription1 sẽ nhận được giá trị 2, còn subscription2 sẽ không nhận được 2 vì đã subscribe sau khi phát 2. Các giá trị liên tiếp sau thì cả 2 đều nhận được.

Terminate

Cuối cùng, là việc kết thúc subscription hoặc subject. Cùng thử nghiệm ví dụ sau:

 let disposeBag = DisposeBag() let subject = PublishSubject<String>() subject.onNext("1") // subscribe 1 let subscription1 = subject .subscribe(onNext: { value in print("Sub 1: ", value) }, onCompleted: { print("sub 1: completed") }) // emit subject.onNext("2") // subscribe 2 let subscription2 = subject .subscribe(onNext: { value in print("Sub 2: ", value) }, onCompleted: { print("sub 2: completed") }) // emit subject.onNext("3") subject.onNext("4") subject.onNext("5") // dispose subscription2 subscription2.dispose() // emit subject.onNext("6") subject.onNext("7") // completed subject.on(.completed) // emit subject.onNext("8") // subscribe 3 subject .subscribe { print("sub 3: ", $0.element ?? $0) } .disposed(by: disposeBag)

• subscription1 sẽ nhận giá trị từ 2, subscription2 sẽ nhận giá trị từ 3,
• Vì là subscription2 đã kết thúc bằng .dispose() chính nó, nên sẽ không nhận được 6 & 7.
• subject không thể phát ra bất cứ gì sau khi gọi .completed và các subscriber cũng không nhận được gì cả. Lúc này sẽ nhận được sự kiện completed mặc dù subject đã kết thúc.
• Khi subject đã .completed hoặc .error, thì các subscriber mới chỉ nhận được sự kiện đó.

Kết quả ra như sau:

Sub 1: 2
Sub 1: 3
Sub 2: 3
Sub 1: 4
Sub 2: 4
Sub 1: 5
Sub 2: 5
Sub 1: 6
Sub 1: 7
sub 1: completed
sub 3: completed

Tham khảo: https://fxstudio.dev/rxswift-publish-subjects/

3. Behavior Subjects

3.1. Khái niệm Behavior Subjects

Nó cũng là một loại Subject và cũng tương tự như Publish Subject. Tuy nhiên, có chỗ khác biệt là Behavior Subjects sẽ luôn cung cấp giá trị cuối cùng mà nó phát ra cho các subscriber khi đăng kí tới.

• Luôn luôn cung cấp giá trị cho phần tử của subject khi khởi tạo subject.
• Các subscriber khi subscribe tới subject, thì luôn nhận được giá trị mới nhất.

3.2. Hoạt động

Create Behavior Subjects

let subject = BehaviorSubject(value: "0")

Ta đã tạo 1 đối tượng là subject với kiểu BehaviorSubject. Bạn không cần phải khai báo kiểu dữ liệu cho phần tử của subject. Vì chúng sẽ tự động suy ra từ giá trị của tham số value mà bạn truyền vào. Trong trường hợp này chính là String.

Emit Data

Tương tự như Publish Subject

subject.onNext("1")

Subscribe

 subject .subscribe { print("sub1 ", $0) } .disposed(by: disposeBag) // emit subject.onNext("1")

Đoạn code subscribe trên nhằm kiểm tra thử việc chúng ta có nhận được giá trị ban đầu khi cấp cho Behavior Subject không. Thực thi đoạn code và xem kết quả nào.

sub1 next(0)
sub1 next(1)

Bạn sẽ thấy subscribe đầu tiên nhận hết các giá trị được phát ra từ subject. Và bạn tiếp tục thêm 1 subscriber nữa để xem sao.

subject .subscribe { print("sub2 ", $0) } .disposed(by: disposeBag)

Lúc này thì subscriber thứ 2 nhận được giá trị 1. Do lúc này 1 là mới nhất.

Terminate

Cùng thử đoạn code sau nhé:

 let subscription1 = subject .subscribe(onNext: { value in print("Sub 1: ", value) }, onCompleted: { print("sub 1: completed") }) subject.onNext("1") // emit subject.onNext("2") // subscribe 2 let subscription2 = subject .subscribe(onNext: { value in print("Sub 2: ", value) }, onCompleted: { print("sub 2: completed") }) // emit subject.onNext("3") subject.onNext("4") // dispose subscription2 subscription2.dispose() // emit subject.onNext("5") subject.onNext("6") subject .subscribe { print("sub 3: ", $0.element ?? $0) } .disposed(by: disposeBag) subject.onNext("7") subject.on(.completed) subject .subscribe { print("sub 4: ", $0.element ?? $0) } .disposed(by: disposeBag)

• subscription1 sẽ nhận giá trị từ 0, subscription2 sẽ nhận giá trị từ 2,
• Vì là subscription2 đã kết thúc bằng .dispose() chính nó, nên sẽ không nhận được 5 & 6.
• subject không thể phát ra bất cứ gì sau khi gọi .completed và các subscriber cũng không nhận được gì cả. Lúc này sẽ nhận được sự kiện completed mặc dù subject đã kết thúc.
• Khi subject đã .completed hoặc .error, thì các subscriber mới chỉ nhận được sự kiện đó.

Kết quả ra như sau:

Sub 1: 0
Sub 1: 1
Sub 1: 2
Sub 2: 2
Sub 1: 3
Sub 2: 3
Sub 1: 4
Sub 2: 4
Sub 1: 5
Sub 1: 6
sub 3: 6
Sub 1: 7
sub 3: 7
sub 1: completed
sub 3: completed
sub 4: completed

Tham khảo: https://fxstudio.dev/rxswift-behavior-subjects/

4. Replay Subjects

4.1. Khái niệm Replay Subjects

Đây cũng là một loại Subject. Đặc điểm của loại subject này, khi phát đi các giá trị thì đồng thời nó lưu lại các giá trị đó trong bộ đệm của mình. Và khi có một subscriber đăng kí tới, subject này sẽ phát đi các giá trị trong bộ đêm của nó cho subscriber đó.

• Khởi tạo bằng kích thước bộ đệm của subject
• Khi phát đi 1 phần tử thì đồng thời lưu trữ nó vào bộ đệm
• Khi có subscriber mới tới thì sẽ nhận được toàn bộ phần tử trong bộ đệm

4.2. Hoạt động

Create Behavior Subjects

let subject = ReplaySubject<String>.create(bufferSize: 2)

• Class sử dụng là ReplaySubject
• Kiểu giá trị được phát đi là String
• Bộ đệm lưu trữ tối đa là 2 phần tử

Ngoài ra, muốn bộ đệm lưu trữ tất cả các giá trị, thì bạn hãy khởi tạo với hàm sau:

let subject = ReplaySubject<String>.createUnbounded()

Emit Data

Tương tự như Publish Subject

 subject.onNext("1") subject.onNext("2") subject.onNext("3")

Subscribe

Giờ chúng ta tiến hành phát đi vài dữ liệu & subscribe lần đầu, để xem các giá trị nhận được là gì?

 subject .subscribe { print("🔵 ", $0) } .disposed(by: disposeBag)

Kết quả:

🔵 next(2)
🔵 next(3)

Vì với khai báo bufferSize = 2, nên bộ đệm của subject chỉ chứa được tối đa 2 phần tử. Do đó, không nhận được giá trị 1.

Chúng ta tiếp tục phát & subscribe lần 2 để xem ra sao.

 // emit subject.onNext("4") // subcribe 2 subject .subscribe { print("🔴 ", $0) } .disposed(by: disposeBag)

Subscriber 1 sẽ nhận được giá trị 4 và subscriber 2 chỉ nhận được 3 và 4:

🔵 next(2)
🔵 next(3)
🔵 next(4)
🔴 next(3)
🔴 next(4)

Terminate

Ta lại tiếp tục ví dụ với emit error và subscribe lần thứ 3:

 // error subject.onError(MyError.anError) // subcribe 3 subject .subscribe { print("🟠 ", $0) } .disposed(by: disposeBag)

Kết quả ra như sau:

🔵 next(2)
🔵 next(3)
🔵 next(4)
🔴 next(3)
🔴 next(4)
🔵 error(anError)
🔴 error(anError)
🟠 next(3)
🟠 next(4)
🟠 error(anError)

Bạn sẽ thấy subscriber thứ 3 sẽ nhận đầy đủ 2 giá trị trong bộ đêm và kèm theo giá trị error của subject. Ngoài ra, 2 subscriber trước đó vẫn nhận error.

Tiếp tục, với việc dispose luôn subject để xem như thế nào.

 // error subject.onError(MyError.anError) // dispose subject.dispose() // subcribe 3 subject .subscribe { print("🟠 ", $0) } .disposed(by: disposeBag)

Kết quả có chút thay đổi.

🔵 next(2)
🔵 next(3)
🔵 next(4)
🔴 next(3)
🔴 next(4)
🔵 error(anError)
🔴 error(anError)
🟠 error(Object `RxSwift.(unknown context at $12d143990).ReplayMany<Swift.String>` was already disposed.)

Đối tượng subscriber thứ 3 không nhận được các dữ liệu từ bộ đệm nữa. Bạn có thể test bằng đoạn code hoàn chỉnh dưới đây và quan sát kết quả:

 let disposeBag = DisposeBag() enum MyError: Error { case anError } let subject = ReplaySubject<String>.create(bufferSize: 2) // emit subject.onNext("1") subject.onNext("2") subject.onNext("3") // subcribe 1 subject .subscribe { print("🔵 ", $0) } .disposed(by: disposeBag) // emit subject.onNext("4") // subcribe 2 subject .subscribe { print("🔴 ", $0) } .disposed(by: disposeBag) // error subject.onError(MyError.anError) // dispose subject.dispose() // subcribe 3 subject .subscribe { print("🟠 ", $0) } .disposed(by: disposeBag)

🔵 next(2)
🔵 next(3)
🔵 next(4)
🔴 next(3)
🔴 next(4)
🔵 error(anError)
🔴 error(anError)
🟠 error(Object `RxSwift.(unknown context at $1027d3b90).ReplayMany<Swift.String>` was already disposed.)

• Ngay cả khi subject phát đi error hay completed thì các subscriber mới vẫn sẽ nhận được đầy đủ các giá trị trong bộ đệm và error hay completed cuối cùng đó.
• Khi sử dụng toán tử dispose() của subject thì toàn bộ mọi thứ sẽ được xoá hết. Nên các subscriber mới lúc đó sẽ không nhận được gì ngoài error.

Tham khảo: https://fxstudio.dev/rxswift-replay-subjects/