如何创建扩展函数，使rx订阅到flowable cleaner？

要创建扩展函数，使rx订阅到flowable cleaner，可以按照以下步骤进行：

首先，确保你已经了解并熟悉了RxJava和Flowable的基本概念和用法。
创建一个扩展函数，可以是一个静态函数或者一个在类中的函数。该函数的输入参数应该是一个RxJava的Observable对象，输出参数为一个Flowable对象。
在函数内部，使用Flowable的create方法来创建一个新的Flowable对象。在create方法的参数中，传入一个FlowableOnSubscribe对象，用于定义Flowable的订阅行为。
在FlowableOnSubscribe的subscribe方法中，将RxJava的Observable对象转换为Flowable对象。可以使用Observable的toFlowable方法来实现转换。
最后，返回创建的Flowable对象。

下面是一个示例代码：

import io.reactivex.Flowable
import io.reactivex.Observable
import io.reactivex.FlowableOnSubscribe

fun <T> Observable<T>.toFlowableCleaner(): Flowable<T> {
    return Flowable.create(FlowableOnSubscribe<T> { emitter ->
        this.subscribe(
            { data -> emitter.onNext(data) },
            { error -> emitter.onError(error) },
            { emitter.onComplete() }
        )
    }, BackpressureStrategy.BUFFER)
}

在这个示例中，我们创建了一个名为toFlowableCleaner的扩展函数，它将Observable对象转换为Flowable对象。通过调用Flowable的create方法，我们定义了Flowable的订阅行为，并在其中将Observable的订阅行为进行了转换。最后，我们返回创建的Flowable对象。

这样，你就可以在使用RxJava的过程中，通过调用toFlowableCleaner函数，将Observable对象转换为Flowable对象，实现订阅到Flowable cleaner的效果。

注意：在实际使用中，你可能需要根据具体的业务需求和场景，对扩展函数进行进一步的优化和定制。