第一步

首先回顾下前面的知识点：

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")
}