第一步

首先回顧下前面的知識點：

flow提供的只是一個 「擴展函數」 返回的是一個保存了這個方法的類實例，并且該類提供emit方法以供flow中調用

構建Flow

「flow方法」

object Flow {
    fun  flow(collect: Collector<T>.() -> Unit): SafeFlowCollector {
        return SafeFlowCollector(collect)
    }
}

定義一個Flow類，內部提供flow方法。

「SafeCollector」 類：

class SafeFlowCollector<T>(val collect: Collector.() -> Unit) {
    //將該Function保存在調用flow后創建的實例中獲取實例創建FlowCollector
    fun collectFunction(a: (T) -> Unit) {
        val co = Collector(a)
        co.collect()
    }

「Collector類」

class Collector<T>(val action: (a: T) -> Unit) {
    fun emit(value: T) {
        action(value)
    }
}

這是flow方法需要創建的三個類，雖然功能不多，但是對于簡單的構建流還是綽綽有余的。

分析

可以看到flow方法傳入的方法參數collect被定義為了Collector的擴展函數，并且保存在了剛創建的SafeCollector的類中用collect函數表示。

「第一個功能」 ：flow參數提交的類型和collect中收到的類型一致，我采用了更加直接的形式定義flow時需要設置傳輸的類型，在emit和collect中都是對應的類型。

「實現」

flow定義類型和emit類型保持一致：通過Collector< T >實現

flow定義類型和收集到的類型一致：通過SafeFlowCollector< T >實現

第二步

構建collect收集器

第一步發射器設置好后，我們限制了發送的類型和接受的類型，并且將發送邏輯保存在了實例中。接下來我們需要調用該實例對象以觸發發送邏輯，在發送邏輯中還需要調用到我們收集的邏輯。

因此收集邏輯需要單獨存放，因此需要單獨構建一個類，這個類還必須可以調用到發送邏輯。

?注：Flow中采用的是collect收集觸發flow流發送邏輯，本人使用時是按照collectFunction定義的。兩者邏輯一樣只是名稱不同

?

flow要構造成哪個類的構造函數，該類就需要持有collect傳入的方法。這也是Collector的功能

SafeFlowCollector和Collector

Collector保存collect傳入的方法，flow擴展Collector以使他可以觸發發射邏輯也就是flow傳入的參數。

class SafeFlowCollector<T>(val collect: Collector.() -> Unit)  //擴展函數

//發送邏輯emit，這個action是從哪里來的呢？
class Collector<T>(val action: (a: T) -> Unit) {
    fun emit(value: T) {
        action(value)
    }
}

再解collectFunction方法：

上面說到action方法，接著看：

fun collectFunction(a: (T) -> Unit) {
    //可以看到調用collectFunction方法將傳入的方法參數保存到了Collector中也就是action方法
    val co = Collector(a)
    //觸發collect，collect也就是flow中傳入的方法。
    co.collect()
}

原生flow，collect的demo：

?注：將上述三個類拷貝到您的項目中即可調用

?

Flow.flow {
    Log.i(TAG, "emit before")
    emit("1")
    Log.i(TAG, "emit after")
}.collectFunction {
    Log.i(TAG, "collectionFunction is : $it")
}

總結

因此可以看到調用collectFunction方法會調用到flow傳入的方法中，在flow傳入的方法中調用emit又會執行collectFunction傳入的方法。

?ps:collectFunction類比于原生Flow中的collect方法即可

?

擴展中間轉換符

flow和collect我們支持了，現在我們來擴展轉換操作符。這里還是簡單實現，原生Flow的實現看的很繞。

map操作符

map方法接受的是一個方法。并且該方法的參數是原數據，經過轉換后返回的值是collect接受的值。

首先我們需要確定幾個點：

1、map的參數如何確定？

?map的參數即上一個Flow emit的值，而在我們這個里面emit的值是通過flow< T >指定的，所以參數直接寫T就可以。2.map的返回值如何確定？

?

?map的返回值要經過兩個階段，收到上一個flow發送的值調用轉換函數把值傳入得到結果，因此map中最后一行即為返回值。

?

3.支持鏈式調用map

?map之后還需要再次經歷map或者collectFunction方法，因此他返回的也必須是一個flowSafeCollector。參數是map傳入方法的返回值。

?

確定好這些來實現，由于需要調用到上一個flow的collect方法， 「原生中是擴展函數this即代表上一個Flow。本demo異曲同工，直接定義為該類的函數」 。于是實現如下， 「修改SafeFlowCollector」 類即可：

class SafeFlowCollector<T>(val collect: Collector.() -> Unit) {
    fun collectFunction(a: (T) -> Unit) {
        val co = Collector(a)
        co.collect()
    }

    //對比發現只是擴展了這個map方法
    fun  map(mapFunction: (T) -> R): SafeFlowCollector {
                //外層調用經map轉換后的collectionFunction會走到flow里面
        return Flow.flow {
                //this@SafeFlowCollector是為了保證調用到調用map函數的flow觸發這個flow的收集
                this@SafeFlowCollector.collectFunction {
                    //當最內層flow函數調用emit，此collectFunction會收到回調，進行轉換后調用flow返回給最外層的collectFunction方法。
                    emit(mapFunction(it))
                }
        }
    }

}

讀者可以直接將上面的SafeFlowCollector換成這個即可。

可支持map轉換Flow的demo

Flow.flow {
    emit("2")
}.map {
    it.toInt()
}.map {
    it.toString()
}.collectFunction {
    Log.i(TAG, "onCreate: it   $it")
}

「最后還是放出這張圖片：」

1649984327(1).png

擴展中間操作符

還是像map一樣，直接把該方法添加到 「SafeFlowCollector」 類中即可

//新增擴展zip操作流，用于多個流發射，就近原則誰先返回用誰
//tips1:收集器類型及·返回流類型，由于返回值相同。因此復用調用方流的泛型即可
//2:開啟收集后觸發多個流的收集，利用標志位進行判斷是否發射。目前采用這種方式。缺點：流發射后應該關閉但是此處只是限制了流的發射邏輯
fun zip(twoFlow: SafeFlowCollector<T>): SafeFlowCollector {
    return Flow.flow {
        var a = false //通過定義condition判斷，不支持協程需要手動切換線程測試效果
        //開啟兩個劉的收集，誰先收集到誰先返回
        this@SafeFlowCollector.collectFunction {
            if (!a) {
                a = true
                this.emit(it)
            }
        }
        twoFlow.collectFunction {
            if (!a) {
                a = true
                this.emit(it)
            }
        }
    }
}

?該種方案可支持多個Flow發射，zip中將參數變為可變參數即可。SafeCollector中擴展

?

zip操作符Demo

val flow1 = Flow.flow<Int> {
        emit(2)
}
Flow.flow<String> {
    Thread{
        emit("111")
    }.interrupt()
}.map {
    it.toInt()
}.zip(flow1).collectFunction {
    Log.i(TAG, "onCreate: ZIP操作符下的Flow收集器收集到的數據位 $it")
}

?上述方案采用標志位實現不太優雅，有更好的方式可以提出~~

?

「不支持協程~~~，需要調度手動切換線程」

總結

「轉換操作符只是中間操作符的一種，其他中間操作符原理大致都一樣。都是經過在封裝一次flow然后觸發上級flow的收集，最后調用到最里層的flow，調用emit在一層層經過中間操作符處理給到最外層」