Obsidian-Main/03. 專注Study/RxKotlin/20200207 - Study RxKotlin.md

# 剛開始
## 先說說Rx

Rx最早是Microsoft的某個實驗室為了解決asychronous、scalable還有一些app問題而提出來的libraray。大概在2009年的時候提出，叫做Reactive Extension for .NET(Rx). 一開始是以add-on的方式安裝在.NET 3.5上，到了.NET 4.0就變成了內建的library。也因為它open source的關係，讓其他語言得以將這套概念也移植過去，所以現在有RxJS, RxSwift, RxNET, RxScale, RxJava。這些library都致力於在它們的語言來實作出相同的「行為」，所以理論上iOS工程師可以和Web工程師用Rx來討論app的邏輯是沒有問題的。
Rx的官網：[http://reactivex.io/](http://reactivex.io/)，它的logo是一支電鰻（Electric eel）：
![[Rx_Logo_S.png]]

Rx Community
- http://android-united.community/
- https://kotlinlang.slack.com/

什麼是RxJava？
> RxJava is a library for composing asynchronous and event-based code using observable sequences and functional style operations, allowing for parameterized execution via schedulers.

> RxJava, in its essence, simplifies developing asynchronous programs by allowing your code to react to new data and process it in a sequential, isolated manner. In other words, RxJava lets  you observe sequence of asychronous events in an app and respond to each event accordingly. Examples are taps by a user on the screen and listening for results if asynchronous network calls.

## 再說RxJava
RxJava是一個實作Rx的framework。
RxJava與其他的Rx library提供了asynchronous與event-based的解決辦法

而Asychronous code跟Sychronous code的差異：
Sychronous code按照字面上的意思執行，每一次的結果都相同。
Asychronous code則是在必要的時候才被使用，每一次執行的「狀態」不盡相同。也就是沒辦法控制其順序與時間。
### Asychronous programming的詞彙
#### 1. State
- State指的是我們程式所儲存的資料與程式自身行為互動所產生的狀態。
- 
####  2. Imperative programming
- Imperative programming（指令式程式設計）是用一連串的命令或描述來改變程式的狀態。如下面的code：
    ```
    setupUI()
    bindClickListeners()
    createAdapter()
    listenForChanges()
    ```
    這些code可能有一些相關的邏輯，但是字面上看不出來，即使互相調換可能會造成錯誤，但也可能不會。
    
####  3. Side effect
- Side effect指的是「一段程式修改了它本身區域外的狀態」，譬如說，一個處理event的function它除了處理event之外，也改變的UI上所顯示的文字。
- Side effect並不總是不好的，我們的程式就是要對某些東西做出改變，完全無法改變任何東西的程式是沒有用的。

RxJava試著用接下來的2個概念來解決剛剛提到的3個概念上的問題
####  4. Declarative code
- 又叫Fucntional programming，Fucntional programming不產生任何side effect。
- Declarative code定義的是行為。
- RxJava試著在Declarative code和Imperative programming取一個平衡點，它定義行為，然後依順序執行。
    
####  5. Reactive systems
Reactive systems通常有幾個特性：
- Reponseive：保持UI在最新狀態
- Resilient：每個行為都是獨立定義的，而且有辦法靈活的處理錯誤。
- Elastic：程式的十座可以處理不同的工作量
- Message driven：每個元件使用Message driven（訊息驅動）的方式來互相溝通，並改進可用性與獨立性，解開（decouple）生命週期與實作的關聯。

### Rx的三大組成
####   1. Observables
`Observable<T>`是Rx的基礎之一，Observable允許觀察者觀察它，並接收它發出來的資料。
##### Observables 的基礎：event
Observable會以3種事件（event）來發出資料：
1. **next**：**next** event會伴隨著一筆資料，這也是觀察者用來接收資料的event。
2. **complete**：**complete** event表示Observable已經「成功的」結束了它的生命週期，在**complete** event之後，觀察者不會再收到任何**next** event。
3. **error**：**error** event表示Observable在發生錯誤的情況下結束它的生命週期。跟**complete** event依樣，後續不會再有任何**next** event。

一個Observable用next所發出來的一連串資料我們稱為"sequence"。sequence可以分為兩種：
1. Finite sequnece：
    想像你要下載一個檔案，我們的code大概是這個樣子：
    ```kotlin
    API.download(file = "http://www...")
        .subscribeBy(
            onNext = { 
                // Handle downloading here
            },
            onComplete = {
                // Handle download finish here
            },
            onError = {
                // Handle error here
            },
        )
    ```
    `API.download()`會產生一個Obervable，然後我們藉由`subscribeBy`來訂閱他，並加入我們的處理程序，我們在`onNext`裡面處理接收到的檔案buffer，在`onComplete`裡面了解到檔案已經完成下載，可以做一些後續的處理，`onError`則是發生了某些錯誤，需要重來或是通知使用者之類。
    
2. Infinite sequence：
    Switch button就是一個例子，我們要處理switch button的code會是這樣：
    ```kotlin
    switch.checkdChanges()
        .subscribeBy(
            onNext = { isOn ->
                if (isOn) {
                    // Handle on here
                } else {
                    // Handle off here
                }
            }
        )
    ```
    可以看到這一段`subscribeBy()`裡面並沒有`onComplete`跟`onError`，因為switch button根本就不會產生這兩種event。
    
####   2. Operators
Operators用來處理Observable所發出來的資料，可能是過濾或者做一些轉換，或其他操作。再以switch button做例子，下面的code可以把switch button的狀態做幾個改變：
1. 我們只想收到on的狀態。
2. 把on的狀態轉為一個字串"We've been toggled on!"。
```kotlin
switch.checkdChanges()
    .filter { it == true }
    .map { "We've been toggled on!" }
    .subscribeBy(
        onNext = { message ->
            updateTextView(message)
        }
    )
```

####   3. Schedulers
Scheduler可以想像成thread，RxJava已經內建了好幾種scheduler，而且應該可以適用於大部分的情形。
例如IO scheduler可以讓你的檔案下載在背景執行，`TeampolineScheduler`可以讓你的程式同時執行， `ComputationScheduler`可以讓你將程式分配給不同的thread來處理需要大量運算的資料。
    
RxJava是一個很獨立的library，所以有2個library可以跟RxJava一起合作：
1. RxAndroid：提供Android Looper class跟RxJava的scheduler之間的橋接管道。
2. RxBinding：用來把UI的click listen之類的callback轉變為Observable的`subscribeBy`。

# 安裝
在`build.gradle`裡的`depedencies`區域加入：
```
implementation "io.reactivex.rxjava3:rxjava:3.0.2"
implementation "io.reactivex.rxjava3:rxkotlin:3.0.0"
implementation "io.reactivex.rxjava3:rxandroid:3.0.0"
```

# Observable
Standard Observable has three types of event:
1. next
2. complete
3. error

Obervable很適合用marble diagram來表示：
![[Pasted image 20210120150947.png]]

3個event解釋如下：
1. `onNext()`：`onNext()` event會伴隨著一筆資料，這也是觀察者用來接收資料的event。
2. `onComplete()`：`onComplete()` event表示Observable已經「成功的」結束了它的生命週期，在`onComplete()` event之後，觀察者不會再收到任何`onNext()` event。
3. `onError()`：`onError()` event表示Observable在發生錯誤的情況下結束它的生命週期。跟`onComplete()` event依樣，後續不會再有任何`onNext()` event。

另外，要注意：一個Observable在沒有被訂閱的情況下，「**是不會發送任何event的**」。

A example of usage of standard Observable:
```kotlin
API.download("http://...")
    .subscribeBy(
        onNext = { /* do something */ }, 
        onComplete = { /* do something */ }, 
        onError = { /* do something */ }, 
    )
```


## 建立`Observable`的方法
### 1. `just`
```kotlin
val observable = Observable.just(1, 2, 3)
```
變數observable的內容設為1個"1、2、3"三個數，型別會是`Observable<Int!>!`。
如果使用了onNext來發送event的話，將會依序發送1、2、3。
但如果是：
```kotlin
val observable = Observable.just(listOf(1, 2, 3))
```
變數observable的內容會是一個list，這個list的內容是"1、2、3"。型別是`Observable<List<Int>!>!`。
如果使用了onNext來發送event的話，將發送一個包含1、2、3的list。
    
### 2. `fromIterable`
用來將list的內容轉變為一個一個單獨的element給Observable。
```kotlin
val observable = Observable.fromIterable(listOf(2, 3, 1))
```
變數observable的型別會是`Observable<Int!>!`，而不是`Observable<List<Int>!>!`。

### 3. `empty`
建立一個「空的」Observable，可以用來表示一個馬上就會結束的事情，或是不包含任何東西的情況。
```kotlin
val observable = Observable.empty<Unit>()  
observable.subscribeBy(  
    onNext = { println(it) },  
    onComplete = { println("Completed") }  
)
```
用`empty()`所建立的observable只會發出`onComplete()` event，所以上面的`onNext()` event永遠不會發生。
還有，Observable所包含的element一定要有一個型別，而且不可以是null，所以上面的`empty()`必須明白的寫出`Unit`型別：`empty<Unit>() `。

### 4. `never`
建立一個不會發出任何event的observable。

### 5. `range`
產生一個範圍的數列，參數型別必須是整數（`Int`）。
```kotlin
val observable = Observable.range(1, 10)
```
上例的`onNext()`會依序發送1~10的數字出來。

### 6. `create`
用來定義自己的event發送方法。
範例：
```kotlin
val observable = Observable.create<String> { emitter ->  
    emitter.onNext("A")  
    emitter.onNext("C")  
    emitter.onNext("B")  
    emitter.onComplete()  
}  
val subscription = observable.subscribeBy(  
    onNext     = { println("Received: $it") },  
    onComplete = { println("Completed") },  
    onError    = { println("Completed") }  
)
```
`create`必須帶入要發送的型別，例如`Int`、`String`或是任何class，此例中是`create<String>`，表示會送出的element是`String`型別。
然後`create`則是發送的實作，範例是會發送"A" -> "C" -> "B"，然後用`onComplete`來結束。
注意：要是observable沒有`onComplete`或是`onError`，然後`Disposable`（也就是訂閱者）也沒有呼叫`dispose()`，則會造成memory leak。

### 7. `defer`
`defer`會建立一個Observable factory，每一次呼叫這個factory都會產生一個新的Observable。`defer`只有一個參數，就是我們要「製造」Observable的方法：
```kotlin
var flip = false  

val factory: Observable<Int> = Observable.defer {  
 flip = !flip  
 if (flip) {  
        Observable.just(1, 2, 3)  
    } else {  
        Observable.just(4, 5, 6)  
    }  
}
```
`defer`後面的lambda就是我們要「製造」Observable的方法。當`flip`是`true`的時候，我們產生`Observable.just(1, 2, 3)`，反之則產生`Observable.just(4, 5, 6)`。Observable裡面所帶的element都是整數，這也是為什麼factory的型別是`Observable<Int>`。
接下來訂閱這個factory：
```kotlin
for (i in 0..3) {  
    val subscription = factory.subscribe {  
        println("Factory out: $it")  
    }  
    disposables.add(subscription)  
}
disposables.dispose()
```
上面的例子產生了4個Observable。依照flip的值來產生不一樣內容的Observable。

How to subscrible a Observable
## 訂閱`Observable`的方法
1. `observable.subscrible()`
2. `observable.subscribleBy()`

Remember to release the resource. Call `disposable()` if you don't need Observable anymore. Or use `CompositeDisposable()` to collect all Disposable and release them.

## 特殊的Observable
### 1. `Single`
`Single`只有`onSuccess`跟`onError`兩種event。在發出`onSuccess`或是`onError`之後，`Single`就結束了。
譬如說讀取檔案，只會有讀取成功跟讀取失敗兩種情況，下面的範例讀取一個檔案，要是檔案不存在就發送`onError()`，反之就發送`onSuccess()`。
```kotlin
val subscriptions = CompositeDisposable()  

fun loadText(filename: String): Single<String> {  
    return Single.create create@{ emitter ->  
        val file = File(filename)  

        if (!file.exists()) {  
            emitter.onError(FileNotFoundException("Can't find $filename"))  
            return@create  
        }  

        val contents = file.readText(Charsets.UTF_8)  
        emitter.onSuccess(contents)  
    }  
}  

// Use the single observable  
val subscription = loadText("Copyright.txt")  
    .subscribeBy(  
        onSuccess = { println("Success read: $it") },  
        onError = { println("Error: $it") }  
 )  
subscriptions.add(subscription)
```
`loadText()`這個function會返回`Single<String>`物件，要是讀取檔案成功，就把檔案內容用`onSuccess()`發送出來：
```kotlin
val contents = file.readText(Charsets.UTF_8)  
emitter.onSuccess(contents)  
```
要是檔案不存在，就發出`onError()`：
```kotlin
emitter.onError(FileNotFoundException("Can't find $filename")) 
```
    
### 2. `Completable`
`Completable`只有`onCompleted`跟`onError`兩種event。跟`Single`一樣，在發出`onCompleted`或是`onError`之後，`Completable`就結束了。
    
### 3. `Maybe`
`Maybe`是`Single`跟`Completable`的混合，他有`onSuccess(value)`、`onCompleted`跟`onError`三種event。`Maybe`只會發出這三種的其中一種event，然後就結束了。

## 停止訂閱或是結束一個`Observable`
### 使用`Disposable.dispose()`
每一次呼叫`observable.subscrible()`或是`observable.subscribleBy()`都會回傳一個`Disposable`物件，當我們不再需要訂閱一個Observable的時候，我們必須呼叫`dispose()`停止訂閱：
```kotlin
val alphaSequnce = Observable.just("A", "B", "C")
val subscription = alphaSequece.subscribe {
    println(it)
}

subscription.dispose()
```

### 使用`CompositeDisposable.dispose()`
對每一個`Disposable`物件在停止訂閱之後都要呼叫一次`dispose()`是很煩人的，RxJava提供了一個`CompositeDisposable` class。它可以收納所有的`Disposable`物件，然後一次停止：
```kotlin
val subscriptions = CompositeDisposable()  
val subscriptionNumbers = Observable.just(1, 2, 3).subscribe {  
    println("Numbers: $it")  
}  
val subscriptionAlphabets = Observable.just("A", "B", "C").subscribe {  
    println("Alphabets: $it")  
}  
  
subscriptions.add(subscriptionNumbers)  
subscriptions.add(subscriptionAlphabets)  
subscriptions.dispose() <-- subscriptionNumbers 與 subscriptionAlphabets 都會一起呼叫dispose()
```
忘記呼叫`dispose()`可能會造成memory leak。

# Subjects
Observable必須在建立的時候就指定好資料，之後沒辦法再新增資料。而Subject可以在建立資料之後，再新增資料，Subject也會將新增的資料再馬上轉發給它的訂閱者。
## 1. `PublishSubject`
`PublishSubject`剛開始是沒有任何資料的，它也只會將最新的資料發送給它的訂閱者。另外，要是`PublishSubject`本身結束了（已經送出了`onComplete` event），那麼新的訂閱者將不會收到任何資料，但是會收到`onComplete` event。
```kotlin
val publishSubject = PublishSubject.create<Int>()  

publishSubject.onNext(0)  

val subscriptionOne = publishSubject.subscribe {  
    println(it)  
}  

publishSubject.onNext(1)  
publishSubject.onNext(2)  

val subscriptionTwo = publishSubject.subscribe {  
    println("2: $it")  
}  

publishSubject.onNext(3)  
subscriptionOne.dispose()  
publishSubject.onNext(4)  
publishSubject.onComplete()  
publishSubject.onNext(5)  
subscriptionTwo.dispose()  

val subscriptionThree = publishSubject.subscribeBy(  
    onNext = { println("3: $it") },  
    onComplete = { println("3: Completed") }  
)
```
上例中的`subscriptionThree`只會收到`onComplete` event，也就是只會印出`"3: Completed"`。
    
## 2. `BehaviorSubject`
行為跟`PublishSubject`類似，但是`BehaviorSubject`會發送「最後一筆資料」給新的訂閱者。如果`BehaviorSubject`最後的event是`onError`，那麼新的訂閱者也會收到`onError` event。例：
```kotlin
val subscriptions = CompositeDisposable()  
val behaviorSubject = BehaviorSubject.createDefault("Initial value")  

behaviorSubject.onNext("X")  

val subscriptionOne = behaviorSubject.subscribeBy(  
    onNext = { println("1: $it") },  
    onError = { println("1: ERROR, $it") }  
)  

behaviorSubject.onError(RuntimeException("Error!"))  
behaviorSubject.onNext("Y") // <-- 不會有人收到，因為已經被onError給terminate了  

subscriptions.add(behaviorSubject.subscribeBy(  
    onNext = { println("2: $it") },  
    onError = { println("2: $it") }  
))
```
另外，可以直接取得`BehaviorSubject`目前的值，以上例來說，只要用`behaviorSubject.value`就可以，這方法可以很方便的在Rx與非Rx的程式中交換資料。
例子中是用static method `BehaviorSubject.createDefault()`來建立一個有初始值的`BehaviorSubject`，當然也可以跟`PublishSubject`一樣，用`BehaviorSubject.create()`來建立。

## 3. `ReplaySubject`:
`BehaviorSubject`會發送會後一筆資料，`ReplaySubject`就是發送最後n筆資料。我們可以用`ReplaySubject.createWithSize()`這個static method來建立一個`ReplaySubject`。例：
```kotlin
val replaySubject = ReplaySubject.createWithSize<String>(2)
```
變數`replaySubject`的buffer容量是2，型別是`String`。

## 4. `AsyncSubject`
`AsyncSubject`的行為比較特別，`AsyncSubject`只會結束的時候，同時發出最後一筆資料。也就是說，即便一直提供資料給`AsyncSubject`，它也不會發出任何`onNext` event給它的訂閱者，直到它收到`onComplete`的時候，它才會同時發出最後一筆`onNext`與`onComplete`給它的訂閱者。

## 5. `RxRelay`
`RxRelay`永遠不會發出`onComplete`或是`onError`。下面例子建立了一個`PublishRelay`：
```
val publishRelay = PublishRelay.create<Int>()
```

要使用`RxRelay` library，必須在build.gradle裡面加入：
```
implementation "com.jakewharton.rxrelay3:rxrelay:3.0.0"
```
    
# Operators
## 1. Filtering Operators
### `ignoreElement`
`ignoreElement()`會忽略掉由[[20200207 - Study RxKotlin#Subjects]]丟出來的**next** event，訂閱者只會收到`onCompleted`跟`onError`這兩種event，也就是讓Subject退化成[[20200207 - Study RxKotlin#2 Completable]]。
例：
```kotlin
val subscriptions = CompositeDisposable()  
val strikes = PublishSubject.create<String>()  
  
subscriptions.add(  
    strikes.ignoreElements()
        .subscribeBy {
            println("Done")  <-- 只會收到onComplete跟onError
        }  
)
```

### `elementAt`
`elementAt()`只會處理「第n個」**next** event，n之前跟n之後的都會被忽略。例如：
```kotlin
val subscriptions = CompositeDisposable()  
val strikes = PublishSubject.create<String>()  
  
subscriptions.add(  
    strikes.elementAt(2)  <-- 只要收第2個
        .subscribeBy {  
            println("Get $it")
        }  
)  
  
strikes.onNext("A")  <-- 第0個
strikes.onNext("B")  <-- 第1個
strikes.onNext("C")  <-- 第2個
strikes.onNext("D")  <-- 第3個
```
上面例子只要收「第2個next event」，所以只會收到"**C**"。這也是一個`onSuccess` event。
`elementAt()`也等於是把Subject退化成[[20200207 - Study RxKotlin#3 Maybe]]。
要是Subject的「第n個」還沒收到就結束了，那就是收到`onComplete` event。

### `filter`
`filter()`接收一個lambda函數，每一個next event所帶的element都必須經過這個函數的「驗證」，只有驗證結果為`true`的時候，才會pass給訂閱者。例：
```kotlin
val subscriptions = CompositeDisposable()  
  
subscriptions.add(  
    Observable.fromIterable(listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))  
        .filter {  
            it > 5                        <-- 數值必須大於5才可以pass給訂閱者
        }  
        .subscribe {  
            print("Get number: $it\n")  
        }  
)
```
所以上面的結果會收到6、7、8、9、10。

### `skip`
忽略「前n個」next event。
```kotlin
Observer.just(1, 2, 3, 4 ,5)
    .skip(3)          <-- 忽略前3個
    .subscribe {
        println("it")
    }
// Output 4, 5
```

### `skipWhile`
`skipWhile`也是用一個lambda來當作通過條件，跟[[20200207 - Study RxKotlin#filter]]類似。但不像`filter`會去檢查「每一個」進來的element，`skipWhile`是「當lambda檢查到第一個`false`的時候，後面全部通過」。
另一個跟`filter`不同的是，`skipWhile`是在檢查為`true`把next event忽略掉，檢查到`false`的時候開始放行。
例如，我們要收集字串，但我們要當字串是"start"的時候才開始收集字串，例：
```kotlin
val subscriptions \= CompositeDisposable()  
  
subscriptions.add(  
    Observable.just("1", "2", "3", "Start", "1", "3", "2")  
        .skipWhile { it != "Start" }  <-- 比對為false開始放行
        .subscribe {  
            println("Get $it")  
        }  
)

// Output
Get Start
Get 1
Get 3
Get 2
```

### `skipUntil`
與前面的skip operator不同，`skipUntil`不是用lambda來決定skip的條件，而是依賴於「另一個subject」，`skipUntil`會一直忽略，直到「另一個subject」發出`onNext` event。例：
```kotlin
val subject = PublishSubject.create<String>()
val trigger = PublishSubject.create<String>()

subject
    .skipUntil(trigger)
    .subscribe {
        println("it")
    }
```
In this code you'll get nothing, until `trigger` sent an `onNext()` event.
Example:
```kotlin
subject.onNext("A") // Ignored
subject.onNext("B") // Ignored
trigger.onNext("1") // TRIGGER!
subject.onNext("C") // send out
```
![[Pasted image 20210202155003.png]]

### `take`
`take`跟[[20200207 - Study RxKotlin#skip]]相反，`take`是接收「前n個」訊息，之後全部忽略。

### `takeWhile`
`takeWhile`是用lambda當判斷條件，當判斷為`true`的時候放行，一旦判斷為`false`，之後的所有訊息都會被忽略。
```kotlin
exampleOf("takeWhile") {  
 val subscriptions = CompositeDisposable()  
  
    subscriptions.add(  
        Observable.fromIterable(listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 1))  
            .takeWhile {  
                it < 5  
            }
            .subscribe {  
                println("Get $it")  
            }  
    )  
}
```
上例中，判斷到5的時候，`it < 5`為`false`，所以之後的都會忽略，即使最後那個1也是一樣被忽略。
`takeWhile`跟[[20200207 - Study RxKotlin#skipWhile]]一樣，但是行為相反。

### `takeUntil`
跟[[20200207 - Study RxKotlin#skipUntil]]相反的行為，會一直接收訊息，直到他依賴的subject發出訊息後停止。

### `distinctUntilChanged`
`distinctUntilChanged`用來過濾「連續且相同」的訊息，例如連續的"Hi"，那麼就只會收到第1個"Hi"，之後的都不會收到。但是一旦收到的訊息改變了，再次收到以前發過的訊息，只要它沒有跟前一筆相同，那麼就還是會接收。例：
```kotlin
val subscriptions = CompositeDisposable()  
  
subscriptions.add(  
    Observable.just(5, 5, 3, 3, 1)  
        .distinctUntilChanged()  
        .subscribe {
            println(it)
        }  
)

// Output
5
3
1
```

`distinctUntilChanged`預設用class的`equal()` method來判斷，當然我們也可以給它一個lambda來當客製我們的條件，被lambda判斷為`true`的話，該訊息就會被忽略：
```kotlin
val subscriptions = CompositeDisposable()  
  
subscriptions.add(  
    Observable.just("ABC", "CCD", "FAG")  
        .distinctUntilChanged { first, second ->  
            first[second.length - 1] == second[0]  
        }
        .subscribe {  
            println("Get $it")  
        }  
)
```
上例中，我們希望「第2個字串的開頭字母要是跟第1個字串的結尾字母一樣的話，那麼就不要顯示」。第1筆"ABC"一定會接收，第2筆"CCD"則會被忽略，第3筆"FAG"會被接收。

## 2. Transforming Operators
### 1. toList
`toList`可以把每一個單獨從Observalbe發出來的元素變成一個list，如：
```kotlin
val subscriptions = CompositeDisposable()  
val items = Observable.just("A", "B", "C")  
  
subscriptions.add(  
    items.toList()  
        .subscribeBy {  
            println(it)  
        }  
)
```
原本會單獨發出的"A", "B", "C"，現在變成只翠發出一個List，其內容是`["A", "B", "C"]`。

### 2. map
`map`跟Kotlin的`map`行為上差不多，只是Kotlin的`map`是作用在List上，而RxJava的`map`是作用在Observable上。`map`根據你提供的lambda函式來對每一個element做轉換，如下例，將每一個羅馬數字轉換為阿拉伯數字：
```kotlin
val subscriptions = CompositeDisposable()  
  
subscriptions.add(  
    Observable.just("M", "C", "V", "I")  
        .map {  
            it.romanNumeralIntValue()  
        }  
        .subscribeBy {  
            println(it)  
        }  
)
```
注意到了嗎？`map`用來轉換Observable所包含的item型別，上例中，Observable的item本來是一個字串（`String`），被`map`轉換為數字（`Int`）。

### 3. flatMap
`flatMap`用來處理「Observable發出來的Observable」，並且「記錄每一個變化」。用例子比較好說明：
```kotlin
class Student(val score: BehaviorSubject<Int>)

val subscriptions = CompositeDisposable()  
val ryan = Student(BehaviorSubject.createDefault(80))  
val charlotte = Student(BehaviorSubject.createDefault(90))  
val student = PublishSubject.create<Student>()  
  
student  
    .flatMap {  
        it.score
    }  
    .subscribeBy {  
        println(it)  
    }  
    .addTo(subscriptions)  
  
student.onNext(ryan)        <-- 1
ryan.score.onNext(85)       <-- 2
  
student.onNext(charlotte)   <-- 3
ryan.score.onNext(95)       <-- 4
charlotte.score.onNext(100) <-- 5
```

我們有一個叫做student的`PublishSubject`，另外有兩個Student class（分別是ryan與charlotte），而這個Student class有一個member叫做score，score的類別是`BehaviorSubject<Int>`。我們的`flatMap` lambda不做任何轉換，直接bypass分數。
1. student先選擇ryan來發出第一個onNext event，ryan原本的分數是80，所以我們會收到80。
2. ryan變更為85，所以我們會收到85。
3. student選擇了charlotte並發出一個onNext event，charlotte原本的分數是90，所以我們會收到90。
4. ryan變更為95，所以我們會收到95。
5. charlotte變更為100，所以我們會收到100。
結果會收到80、85、90、95、100。

ryan跟charlotte都是獨立的Observable，但透過`flatMap`我們可以把它們的值（以及後續的變化）變成一連串的數值，這就是`flat`的意思。
![[rxJava_flatMap.png]]

### 4. switchMap
`switchMap`跟`flatMap`類似，也是處理「Observable發出來的Observable」，但是差別在於`switchMap`一但切換到新的Observable之後，上一個Observale的訊息就部會收到了，以`flatMap`的例子來說，在`student.onNext(charlotte)`這一行之後，ryan的改變就不會收到了。例：
```kotlin
val ryan = Student(BehaviorSubject.createDefault(80))  
val charlotte = Student(BehaviorSubject.createDefault(90))  
val student = PublishSubject.create<Student>()  
  
student  
    .switchMap {  
        it.score  
    }  
    .subscribe {  
        println(it)  
    }  
  
student.onNext(ryan)  
ryan.score.onNext(85)  
  
student.onNext(charlotte)  
ryan.score.onNext(95)  
charlotte.score.onNext(100)
```
結果會收到80、85、95、100。`ryan.score.onNext(95)`這一行的95不會收到。

`switchMap`適合用在會「改變興趣」的場景，例如說原本是要持續收到台北氣溫的改變，接著使用者把地點改到高雄，那我們就會變成持續收到高雄的溫度變化而不是台北的，又或者說，你會隨著使用的的輸入持續的搜尋結果，例如使用者依序輸入k、o、t、l、i、n，每輸入一個字母我們就搜尋一次，但我們只關注最後一個字串搜尋，不在意之前的搜尋結果。

### 5. materialize
`materialize`能將Observable的值包裝成一個`Notification`，回到[[20200207 - Study RxKotlin#4 switchMap]]的例子，如果任何一個學生發出了`onError`的訊息，那麼連`student`本身都會因為這個Exception而中斷，所以即使切到了charlotte，我們也收不到charlotte的訊息了。`materialize`可以將`onError`包裝成一個`Notification`，讓exception留在ryan本身而不會影響到上面的student。
```kotlin
val subscriptions = CompositeDisposable()  
val ryan = Student(BehaviorSubject.createDefault(80))  
val charlotte = Student(BehaviorSubject.createDefault(90))  
val student = BehaviorSubject.createDefault<Student>(ryan)  
  
// 1  
val studentScore = student.switchMap { it.score.materialize() }  <-- HERE!
// 2  
subscriptions.add(  
    studentScore  
        .subscribe {  
            println(it)  
        }  
)  
// 3  
ryan.score.onNext(85)  
ryan.score.onError(RuntimeException("Error!"))  
ryan.score.onNext(90)  
// 4  
student.onNext(charlotte)
```
![[rxkotlin_materialize.png]]

### 6. dematerialize
`dematerialize`用來反解`materialize`所包裝的東西，例如上例中，會將`materialize`所包裝出來的`Observable<Notification<Int!>!>!`反解為`Observable<Int>!`，例：
```kotlin
subscriptions.add(  
    studentScore  
    .filter {  
        if (it.error != null) {  
            println("Got error: ${it.error}")  
            false  
        } else {  
            true  
        }  
    }  
    .dematerialize { it }  
    .subscribe {  
        println(it)  
    }  
)
```
要注意的是如果發生Exception的話，直接`println`還是會產生Exception，所以需要用`filter`來把error給濾掉。
![[rxkotlin_dematerialize.png]]

在Subject所發出的element仍然是Subject的時候，如果element發生error（Exception），會導致上層的Subject也跟著停止，`materialize`/`dematerialize`可以用來包裝element，讓element所發出的東西都變成`Notification`，這樣就部會影響上層的Subject了。

## 3. Combining Operators
### startWith
用來在Observable本身所帶的item前面再加上其他item。實際的有`startWithIterable()`與`startWithItem()`。
例：
```kotlin
val subscriptions = CompositeDisposable()  
val numbers = Observable.just(3, 4, 5)  
val startWith = numbers.startWithIterable(listOf(1, 2))  
  
startWith.subscribe {  
    println(it)  
}.addTo(subscriptions)
```

### concat
`concat`是一個static method，用來合併2個Observable。`concat`會先等第一個Observable結束，然後再等待第二個，之後把它們合併起來。
```kotlin
val subscriptions = CompositeDisposable()  
val first = Observable.just(1, 2,  3)  
val second = Observable.just(3, 4, 5)  
  
Observable.concat(first, second)  
    .subscribe {  
        println(it)  
    }  
    .addTo(subscriptions)
```

### concatWith
跟[[20200207 - Study RxKotlin#concat]]一樣，但是`concatWith`是一個member function，而不是一個static method。`concatWith`一樣會先等自己結束，然後再等第二個Obervable（當參數的那一個）結束，之後再合併。
```kotlin
val subscriptions = CompositeDisposable()  
val first = Observable.just(1, 2,  3)  
val second = Observable.just(3, 4, 5)  
  
first.concatWith(second)  
    .subscribe {
        println(it)
    }
    .addTo(subscriptions)
```

> ## 注意
> 要被合併的兩個Observable類型必須要一樣，不可以一個是`Obsrvable<String>`而另一個是`Observable<Int>`，compiler會報錯喔。

### concatMap
`concatMap`接受一個lambda函示，並回傳另一個Observable序列，`concatMap`會保證Observable的順序。
Given multiple Observable, and map each Observable to a lambda function. And make sure the sequence of given Observable list.

### merge
`merge`是一個static function。
`merge`會按照接收的順序把element合併起來，例：
```kotlin
val subscriptions = CompositeDisposable()  
val odd = PublishSubject.create<Int\>()  
val even = PublishSubject.create<Int\>()  
  
Observable.merge(odd, even)  
    .subscribe {  
        println(it)  
    }  
    .addTo(subscriptions)  
  
odd.onNext(1)  
even.onNext(2)  
odd.onNext(3)  
even.onNext(4)  
odd.onNext(5)  
even.onNext(6)
```
odd與even兩個交互發出elemet，merge依順序接收，而不是像[[20200207 - Study RxKotlin#concat]]是依照Observable的順序。
`merge` complete的時機點定義如下：
- 當來源的Observable與內部的Observable都complete的時候，merge本身也會發出complete。
- 內部Observable結束的順序跟接收的順序沒有關係。（一律按照接收點）
- 如果有任何Observable發生error，`merge`會轉發這個error，然後自己發生terminate。

Q:
- What's the different between `flatMap()`? -> 很大的差別，`flatMap()` 有map的功能。

### mergeWith
就像`concatWith()`與`concat()`的關係，`mergeWith()`跟`merge`也是一樣的關係。
`mergeWith()`是一個member function，必須由某個Observable instance來呼叫。

### combineLatest
combineLatest是`Observables`的static funtion（注意不是`Observable`）。
combineLatest只會接收2個Observalbe的「最後一個」elements，然後交由你所提供的lambda來處置，例：
```kotlin
val subscriptions = CompositeDisposable()
val left = PublishSubject.create<String>()
val right = PublishSubject.create<String>()

Observables.combineLatest(left, right) { leftString, rightString ->
    "$leftString, $rightString"
}.subscribe {
    println(it)
}.addTo(subscriptions)

left.onNext("Hello")
right.onNext("World")
left.onNext("It's nice to")
right.onNext("be here")
left.onNext("Actually, it's super great to")
```

重點：
1. 在上例中是直接結合2個字串，但是其實可以是任何用途。
2. 在實務中，`combineLatest`可以用來結合2個不同型別的Observable，然後再回傳另一個不同型別的Observable。`combineLatest`回傳的Observable型別由lambda決定。
3. `combineLatest`必須在「每一個」Observable都發出element之後才會動作。如果不確定Observable是否會發出element，可以使用[[20200207 - Study RxKotlin#startWith]]來讓Observable有一個初始值，這樣可以避免`combineLatest`永遠不會發生的情況。
4. 如果有某個Observable已經complete，`combineLatest`會保留它的最後一個element，然後繼續結合更新的element。
5. 直到最後一個Observable complete，`combineLatest`才會complete。
![[combineLatest.png]]

### zip
- `zip` wait until each if the inner Ovservables emits a new value.

### Triggers
#### withLastestFrom
- `withLatestFrom` is useful in all situations where you want the current(latest) value emittted frim an Observable, but only when a particular trigger occurs.

#### sample
- Just like `withLastFrom`. But each time the trygger Ivsercable emits a value, `sample` emits the latest value from the "other" Obervable, but only if it arrived since the last "tick". You can combine `withLastFrom` and `distinctUntilChanged` to do the same behavior of `sample`.
```
exampleOf("sample") {
    val subscriptions = CompositeDisposable()
    val button = PublishSubject.create<Unit>()
    val editText = PublishSubject.create<String>()

    editText.sample(button)
        .subscribe {
            println(it)
        }.addTo(subscriptions)

    editText.onNext("Par")
    editText.onNext("Pari")
    editText.onNext("Paris")
    button.onNext(Unit) 
    button.onNext(Unit) <- button emits twice, but editText only emit last value
}

// Output
--- Example of: sample ---
Paris
```

### Switchs
- ambWith
    - Think of `amb` as in ambiguous.
    - `ambWith` connect to two Observables. And wait any of them who emit element first. If any Observable emit element, another one will be unsubscribed.
### reduce
- `reduce` accumulates a summary value.

### scan
- Like `reduce`, but emit per input value.

## 4. Time-Based Operators
### Buffering
#### replay
- This operator creates a new sequence that records the last N elements emitted by the source Observable.

#### replayAll

#### window
- Difference is that it emits an Observable of the buffered items, instead of emitting an array.

### Time-Shifting
#### delaySubscription
- Delay the time a subscriber starts receiving elements from its subscription.

#### delay
- This operator subscribes immediateley to the source observable, but delays every emitted element by the specified amount of time.

### Timer
#### Observable.interval
- Produce an infinite Observable sequence of Int values.

#### Observable.timer
- Specify a "due time" as the time that elapsed between the point of subscription and the first emitted value.
- If the "repeat period" is not assigned, the timer Observable will emit once, the complete.

#### timeout
- Emit an TimeoutException error event. If not caught, it terminates the sequence.

## 5. Explore Operators


# 參考資料：
- [RxMarbles: Interactive diagrams of Rx Observables](https://rxmarbles.com/#delayWhen)