如果 Flow 的介紹只停留在一方丟資料一方接資料，然後接資料的一方還把 callback 傳給丟資料的人，那相信大家讀完可能會越想越不對勁，今天我們就把單純的傳接球變成多人接力吧。

Operator

不管是 Rx、Flow 或是 Kotlin 的 collection，其實都有一系列的 operator 可以做一層一層的資料轉換，而透過這些 operator 不斷地串接，我們就可以達成各式各樣的需求。

Map

讓我們從最簡單的 map 開始，顧名思義 map 就是把一個資料從原本的格式/值透過一個 function 轉成另外一個格式/值。

簡單範例如下：

flowOf(1, 2, 3)
    .map{ it * 2 }
    .collect{
        println(it)
    }

我們透過 flowOf 把 1,2,3 轉成 Flow ，然後透過 map 把原始資料都乘以 2 最後輸出，所以 console 上應該會看到 2, 4, 6 這樣的結果，值得一提的是 map 不只可以改變值，也可以把型別做轉換，如以下範例：

flowOf(1, 2, 3)
    .map{ "value: $it" }
    .collect{
        println(it)
    }

這樣最後的輸出就會變成

value: 1
value: 2
value: 3

而 map 是怎麼做的呢？讓我們稍微來看一下他的程式碼：

public inline fun <T, R> Flow<T>.map(crossinline transform: suspend (value: T) -> R): Flow<R> =transform{value->
return@transform emit(transform(value))
}

首先，我們看到有個 transform 的 function，在 emit 前會呼叫我們傳入的 transform function，那 transform 又做了什麼呢？原始碼如下：

public inline fun <T, R> Flow<T>.transform(
    @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = flow { // Note: safe flow is used here, because collector is exposed to transform on each operation
    collect { value ->
        // kludge, without it Unit will be returned and TCE won't kick in, KT-28938
        return@collect transform(value)
    }
}

transform 建立了一個 Flow，在下游開始 collect 它的時候，它也會開始 collect 上游，而收到的值再呼叫一次 transform 然後 emit 給下游，是不是寫得非常漂亮呢。

OnCompletion

Operator 不一定都要做資料的轉換，也可以是通知類別的比如說: onStart、onComplete，顧名思義就是告訴我們一個 Flow 什麼時候開始以及結束的通知，以下我們以 onComplete 為範例做說明：

 
flowOf(1, 2, 3)
    .map{ it * 2 }
    .onCompletion {
        println("finish")
    }
    .collect{
        println(it)
    }

這邊就會印出：

2
4
6
finish

我們也來稍微看一下原始碼：

public fun <T> Flow<T>.onCompletion(
    action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit
): Flow<T> = unsafeFlow { // Note: unsafe flow is used here, but safe collector is used to invoke completion action
    try {
        collect(this)
    } catch (e: Throwable) {
        /*
         * Use throwing collector to prevent any emissions from the
         * completion sequence when downstream has failed, otherwise it may
         * lead to a non-sequential behaviour impossible with `finally`
         */
        ThrowingCollector(e).invokeSafely(action, e)
        throw e
    }
    // Normal completion
    val sc = SafeCollector(this, currentCoroutineContext())
    try {
        sc.action(null)
    } finally {
        sc.releaseIntercepted()
    }
}

unsafeFlow 其實跟一般的 Flow 差不多，而這個 function block 裡基本上就做二件事情，collect 然後呼叫 action，這樣就完成了在結束的時後呼叫原本的 onCompletion 目標。

是不是很有趣呢？Flow 提供的 operator 遠遠不止 map 跟 onCompletion，甚至你也可以依據自己的需求寫自己的 operator。希望大家都能多多使用 operator 來簡化自己的程式邏輯。

Kotlin 的技術傳教士 - 范聖佑 近期也出了一本關於 Collection 的書 - Kotlin Collection 全方位解析攻略
裡面也有蠻多 operator 的介紹，歡迎大家有興趣的參考看看
https://www.tenlong.com.tw/products/9786263331136