第一步

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

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

構(gòu)建Flow

「flow方法」

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

定義一個Flow類，內(nèi)部提供flow方法。

「SafeCollector」 類：

class SafeFlowCollector<T>(val collect: Collector.() -> Unit) {
    //將該Function保存在調(diào)用flow后創(chuàng)建的實例中獲取實例創(chuàng)建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方法需要創(chuàng)建的三個類，雖然功能不多，但是對于簡單的構(gòu)建流還是綽綽有余的。

分析

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

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

「實現(xiàn)」

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

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

第二步

構(gòu)建collect收集器

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

因此收集邏輯需要單獨存放，因此需要單獨構(gòu)建一個類，這個類還必須可以調(diào)用到發(fā)送邏輯。

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

?

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

SafeFlowCollector和Collector

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

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

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

再解collectFunction方法：

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

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

原生flow，collect的demo：

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

?

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

總結(jié)

因此可以看到調(diào)用collectFunction方法會調(diào)用到flow傳入的方法中，在flow傳入的方法中調(diào)用emit又會執(zhí)行collectFunction傳入的方法。

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

?

擴展中間轉(zhuǎn)換符

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

map操作符

map方法接受的是一個方法。并且該方法的參數(shù)是原數(shù)據(jù)，經(jīng)過轉(zhuǎn)換后返回的值是collect接受的值。

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

1、map的參數(shù)如何確定？

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

?

?map的返回值要經(jīng)過兩個階段，收到上一個flow發(fā)送的值調(diào)用轉(zhuǎn)換函數(shù)把值傳入得到結(jié)果，因此map中最后一行即為返回值。

?

3.支持鏈式調(diào)用map

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

?

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

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

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

}

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

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

?該種方案可支持多個Flow發(fā)射，zip中將參數(shù)變?yōu)榭勺儏?shù)即可。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收集器收集到的數(shù)據(jù)位 $it")
}

?上述方案采用標志位實現(xiàn)不太優(yōu)雅，有更好的方式可以提出~~

?

「不支持協(xié)程~~~，需要調(diào)度手動切換線程」

總結(jié)

「轉(zhuǎn)換操作符只是中間操作符的一種，其他中間操作符原理大致都一樣。都是經(jīng)過在封裝一次flow然后觸發(fā)上級flow的收集，最后調(diào)用到最里層的flow，調(diào)用emit在一層層經(jīng)過中間操作符處理給到最外層」