1. Prefixing and concatenating

1.1. startWith

Cần thêm một/nhiều phần tử trước khi Observable bắt đầu emit dữ liệu đi chúng ta sử dụng startWith

Ví dụ:

    let bag = DisposeBag()
    
    Observable.of("B", "C", "D", "E")
        .startWith("A")
        .subscribe(onNext: { value in
            print(value)
        })
        .disposed(by: bag)

Bạn sẽ thấy, với Observable phát ra các giá trị bắt đầu từ B. Thì để chèn thêm 1 phần tử ở trước ta dùng toán tử startWith và thêm A vào. Kết quả ra như sau:

A
B
C
D
E

1.2. Observable.concat

concat sẽ nối các phần tử của nhiều sequence observable lại với nhau.

Ví dụ:

  let bag = DisposeBag()
    
    let first = Observable.of(1, 2, 3)
    let second = Observable.of(4, 5, 6)
    let observable = Observable.concat([first, second])
    observable
        .subscribe(onNext: { value in
            print(value)
        })
        .disposed(by: bag)

Kết quả:

1
2
3
4
5
6

Toán tử concat dùng với 1 đối tượng Observable. Nó sẽ nối các phần tử của một đối tượng Observable khác vào.

Ví dụ:

    let bag = DisposeBag()
    
    let first = Observable.of("A", "B", "C")
    let second = Observable.of("D", "E", "F")
    
    let observable = first.concat(second)
    
    observable
        .subscribe(onNext: { value in
            print(value)
        })
        .disposed(by: bag)

Ta dùng first.concat với tham số là second. Kết quả in ra như sau:

A
B
C
D
E
F

Chú ý: Toán tử concat này sẽ đợi Observable gốc hoàn thành. Thì sau đó Observable thứ 2 sẽ tiếp tục được nối vào.

1.4. concatMap

Ở hình trên thì mô tả như sau:

• Hình tròn là 1 Observable
• Hình thoi là giá trị được phát ra cho Subscriber
• Luật trong toán tử là cứ 1 Observable sẽ phát ra 2 dữ liệu hình thoi

Ví dụ:

    let bag = DisposeBag()
    
    let cities = [ "Mien Bac" : Observable.of("Ha Noi", "Hai Phong"),
                   "Mien Trung" : Observable.of("Hue", "Da Nang"),
                   "Mien Nam" : Observable.of("Ho Chi Minh", "Can Tho")]
    
    let observable = Observable.of("Mien Bac", "Mien Trung", "Mien Nam")
        .concatMap { name in
            cities[name] ?? .empty()
        }
    
    observable
        .subscribe(onNext: { (value) in
            print(value)
        })
        .disposed(by: bag)

Kết quả ra như sau:

Ha Noi
Hai Phong
Hue
Da Nang
Ho Chi Minh
Can Tho

Trong đó

• cities là một Dictionary với key là String và value là 1 Observable.
• Tạo ra 1 Observable với kiểu dữ liệu cho phần tử là String.
• Dùng concatMap để biến đổi từ String thành String. Tuy nhiên có sự can thiệp là nối các chuỗi thuộc value của Dictionary trên lại với nhau.

Vì value là 1 Observable nên phải dùng concatMap chứ không phải là map, nếu bạn dùng map nó sẽ chỉ ra dạng:

RxSwift.(unknown context at $1048ac5cc).ObservableSequence<Swift.Array<Swift.String>>
RxSwift.(unknown context at $1048ac5cc).ObservableSequence<Swift.Array<Swift.String>>
RxSwift.(unknown context at $1048ac5cc).ObservableSequence<Swift.Array<Swift.String>>

2. Merging

Toán tử đầu tiên chính là merge. Cái tên cũng nói lên tất cả rồi. Và đặc điểm của toán tử này như sau:

• merge sẽ tạo ra 1 Observable mới, khi một Observable có các element kiểu là Observable
• Observable của merge sẽ kết thúc khi tất cả đều kết thúc
• Nó không quan tâm tới thứ tự các Observable được thêm vào. Nên nếu có bất cứ phần tử nào từ các Observable đó phát đi thì Subscriber cũng đều nhận được

Ví dụ:

   let bag = DisposeBag()
    
    let chu = PublishSubject<String>()
    let so = PublishSubject<String>()
    
    let source = Observable.of(chu.asObserver(), so.asObserver())
    
    let observable = source.merge()
    
    observable
        .subscribe(onNext: { (value) in
            print(value)
        })
        .disposed(by: bag)
    
    chu.onNext("Một")
    so.onNext("1")
    chu.onNext("Hai")
    so.onNext("2")
    chu.onNext("Ba")
    so.onCompleted()
    so.onNext("3")
    chu.onNext("Bốn")
    chu.onCompleted()

Kết quả như sau:

Một
1
Hai
2
Ba
Bốn

Trong đó:

• chu & so là 2 Subject, chịu trách nhiệm phát đi dữ liệu
• source được tạo ra từ 2 Observable chu & so . Nó là 1 Observable
• observable được tạo ra bằng toán tử merge
• Subscribe như bình thường
• Việc phát dữ liệu được thực hiện xen kẻ

3. Combining elements

3.1. combineLatest

Thông qua sơ đồ mô tả toán tử combineLatest , toán tử này sẽ phát đi những giá trị là sự kết hợp của các cặp giá trị mới nhất của từng Observable. Để hình dung cụ thể, ta qua từng bước ví dụ sau đây:

• Tạo các Observable, sẽ phát dữ liệu đi. Có thể không theo thứ tự.

 let chu = PublishSubject<String>()
  let so = PublishSubject<String>()

• Sử dụng combinedLatest với 2 Observable trên. Sau đó tiến hành subscribe như bình thường.

    let observable = Observable.combineLatest(chu, so)
    
    observable
        .subscribe(onNext: { (value) in
            print(value)
        })
    .disposed(by: bag)

• Giờ chúng ta phát dữ liệu đi một cách lần lượt và xem nó hoạt động như thế nào.

    chu.onNext("Một")
    chu.onNext("Hai")
    so.onNext("1")
    so.onNext("2")
    
    chu.onNext("Ba")
    so.onNext("3")
    chu.onNext("Bốn")
    so.onNext("4")
    so.onNext("5")
    so.onNext("6")

Kết quả:

("Hai", "1")
("Hai", "2")
("Ba", "2")
("Ba", "3")
("Bốn", "3")
("Bốn", "4")
("Bốn", "5")
("Bốn", "6")

Kết quả là sự kết hợp dữ liệu từ 2 dữ liệu được Observable chữ và số phát đi. Bạn sẽ thấy là không có ("Một", "1"). Vì lúc đó Observable so chưa phát ra gì cả. Khi so phát ra phần tử đầu tiên thì sẽ kết hợp với phần tử mới nhất của chu, đó là Hai.

Thử với complete:

    chu.onNext("Một")
    chu.onNext("Hai")
    so.onNext("1")
    so.onNext("2")
    
    chu.onNext("Ba")
    so.onNext("3")
    // completed
    chu.onCompleted()
    chu.onNext("Bốn")
    
    so.onNext("4")
    so.onNext("5")
    so.onNext("6")
    
    // completed
    so.onCompleted()

Kết quả:

("Hai", "1")
("Hai", "2")
("Ba", "2")
("Ba", "3")
("Ba", "4")
("Ba", "5")
("Ba", "6")

Không ảnh hưởng gì tới sự hoạt động của toán tử này. Nó sẽ vẫn lấy dữ liệu cuối cùng trước khi phát đi .onCompleted để kết hợp với các phần tử khác.

Chúng ta cũng có thể custom giá trị đầu ra với tham số combineLatest(_:_:resultSelector:)

    let observable = Observable.combineLatest(chu, so) { chu, so in
        "\(chu) : \(so)"
    }

Kết quả như sau:

Hai : 1
Hai : 2
Ba : 2
Ba : 3
Ba : 4
Ba : 5
Ba : 6

3.2. zip

Khi bạn quan tâm tới thứ tự kết hợp theo đúng thứ tự phát đi của từng Observable. Nhưng combinedLatest không đáp ứng được thì zip sẽ giúp bạn hoàn thành tâm nguyện này.

Ví dụ:

        let chu = PublishSubject<String>()
        let so = PublishSubject<String>()
        let observable = Observable.zip(chu, so) { chu, so in
            "\(chu) : \(so)"
        }
        
        observable
            .subscribe(onNext: { (value) in
                print(value)
            })
            .disposed(by: bag)
        
        chu.onNext("Một")
        chu.onNext("Hai")
        so.onNext("1")
        so.onNext("2")
        
        chu.onNext("Ba")
        chu.onNext("Bốn")
        chu.onNext("Năm")
        so.onNext("3")
        
        so.onNext("4")
        so.onNext("5")
        so.onNext("6")

Kết quả:

Một : 1
Hai : 2
Ba : 3
Bốn : 4
Năm : 5

4. Trigger

4.1. withLatestFrom

Ta đi trực tiếp vào ví dụ cho dễ hình dung:

        let button = PublishSubject<Int>()
        let textField = PublishSubject<String>()
        let observable = button.withLatestFrom(textField) { button, textfield in
            "\(button) : \(textfield)"
        }
        
        _ = observable
            .subscribe(onNext: { value in
                print(value)
            })
        textField.onNext("Đa")
        button.onNext(1)
        textField.onNext("Đà Na")
        button.onNext(2)
        textField.onNext("Đà Nẵng")
        textField.onCompleted()

Kết quả thực thi code như sau:

1 : Đa
2 : Đà Na

• button là một subject. Với Void thì chỉ phát ra sự kiện, chứ không có giá trị nào từ onNext
• textField là một subject, phát ra các String
• observable là sự kết hợp của button với textField thông qua toán tử withLatestFrom
• Mỗi lần button phát đi tín hiệu, thì kết quả sẽ nhận được là phần tử mới nhất từ textField

4.2. sample

Ta đi trực tiếp vào ví dụ cho dễ hình dung:

        let button = PublishSubject<Int>()
        let textField = PublishSubject<String>()
        let observable = button.sample(textField)
        
        _ = observable
            .subscribe(onNext: { value in
                print(value)
            })
        textField.onNext("Đa")
        button.onNext(1)
        textField.onNext("Đà Na")
        button.onNext(2)
        textField.onNext("Đà Nẵng")
        button.onNext(3)
        textField.onCompleted()

• button là một subject. Với Void thì chỉ phát ra sự kiện, chứ không có giá trị nào từ onNext
• textField là một subject, phát ra các String
• observable là sự kết hợp của button với textField thông qua toán tử sample
• Ngược lại với withLatestFrom, nó chỉ phát khi Observable textField phát ra, mỗi lần textField phát đi tín hiệu, thì kết quả sẽ nhận được là phần tử mới nhất từ button tính cả .onCompleted()
• Kết quả nhận được chỉ có giá trị của button,

5. Switches

5.1. amb

Đây là một toán tử khá là mơ hồ, cũng như cái tên của nó là ambiguity. Với các đặc tính sau:

• Nó sẽ tạo ra một Observable để giải quyết vấn đề quyết định nhận dữ liệu từ nguồn nào
• Trong khi cả 2 nguồn đều có thể phát dữ liệu. Thì nguồn nào phát trước, thì nó sẽ nhận dữ liệu từ nguồn đó.
• Nguồn phát sau sẽ bị âm thầm ngắt kết nối

Ví dụ:

let bag = DisposeBag()
let chu = PublishSubject<String>()
let so = PublishSubject<String>()
let observable = chu.amb(so)
observable
    .subscribe(onNext: { (value) in
        print(value)
    })
.disposed(by: bag)
so.onNext("1")
so.onNext("2")
so.onNext("3")
chu.onNext("Một")
chu.onNext("Hai")
chu.onNext("Ba")
so.onNext("4")
so.onNext("5")
so.onNext("6")
   
chu.onNext("Bốn")
chu.onNext("Năm")
chu.onNext("Sáu")

Kết quả:

1
2
3
4
5
6

Vì so đã phát trước, nên các dữ liệu từ chu sẽ không nhận được. Nếu bạn cho thêm chu phát onNext trước so thì sẽ thấy dữ liệu nhận được sẽ toàn là từ chu.

5.2. switchLatest

Ví dụ:

        let chu = PublishSubject<String>()
        let so = PublishSubject<String>()
        let dau = PublishSubject<String>()
        let observable = PublishSubject<Observable<String>>()
        observable
            .switchLatest()
            .subscribe(onNext: { (value) in
                print(value)
            }, onCompleted: {
                print("completed")
            })
        .disposed(by: bag)
        observable.onNext(so)
        so.onNext("1")
        so.onNext("2")
        so.onNext("3")
        observable.onNext(chu)
        chu.onNext("Một")
        chu.onNext("Hai")
        chu.onNext("Ba")
        so.onNext("4")
        so.onNext("5")
        so.onNext("6")
        observable.onNext(dau)
        dau.onNext("+")
        dau.onNext("-")
           
        observable.onNext(chu)
        chu.onNext("Bốn")
        chu.onNext("Năm")
        chu.onNext("Sáu")

Kết quả thực thi như sau:

1
2
3
Một
Hai
Ba
+
-
Bốn
Năm
Sáu

Ta có:

• 3 subject thay nhau phát dữ liệu
• observable với kiểu dữ liệu phát đi là Obsevable<String> , chính là kiểu của 3 subject trên Bạn sẽ thấy việc observable sẽ phát đi subject nào. Thì subscriber trên sẽ nhận được giá trị của subject đó. Hãy thay phiên nhau việc phát dữ liệu, nó sẽ giúp bạn hiểu rõ hơn về toán tử này Còn để kết thúc nó, thì phải phải .dispose subscription. Chứ không thể nào kết thúc nó được, mặc dù các subject có thể onCompleted hết tất cả nhưng nó vẫn không kết thúc.

6. Combining elements within a sequence

6.1. reduce

Ví dụ:

let source = Observable.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
let observable = source.reduce(0, accumulator: +)
Hoặc let observable = source.reduce(0) { $0 + $1 }
Hoặc 
let observable = source.reduce(0) { summary, newValue in
  return summary + newValue
}
_ = observable
    .subscribe(onNext: { value in
        print(value)
    })

Toán tử này khá quen thuộc trong swift rồi và Rx cũng tương tự thế nên mình sẽ không đề cập sâu nữa.

6.2. scan(_:accumulator:)

Về cấu trúc và cách viết thì tương tự reduce. Có chút khác biệt ở đây là, thay vì chờ Observable kết thúc và đưa ra kết quả cuối cùng. Thì scan nó sẽ tính toán và phát đi từng kết quả tinh toán được, sau khi có dữ liệu từ Observable phát ra. Không quan tâm Observable kết thúc mới thực hiện.

Ví dụ:

let source = Observable.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
//let observable = source.scan(0, accumulator: +)
//let observable = source.scan(0) { $0 + $1 }
let observable = source.scan(0) { summary, newValue in
  return summary + newValue
}
_ = observable
    .subscribe(onNext: { value in
        print(value)
    })

Kết quả:

1
3
6
10
15
21
28
36
45

Tham khảo: https://fxstudio.dev/rxswift-combining-operators/