Kotlin中如何将CompletableFuture<Optional<T>>转换为Flow<T?>

问题现状 #

最初尝试基于Reactor的Mono做中转实现类型转换，代码如下：

fun <T> CompletableFuture<Optional<T>>.asFlowOfNullable(): Flow<T?> =
    this.toMono().map { if (it.isPresent) it.get() else null }.asFlow()

上述代码运行报错，根因是asFlow()方法不支持可空类型的参数。

方案迭代 #

第一版实现（存在阻塞缺陷） #

最初的手写调整方案如下，额外封装了Optional转可空的通用扩展：

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOf
import java.util.Optional
import java.util.concurrent.CompletableFuture

fun <T> Optional<T>.orNull(): T? = orElse(null)

fun <T> CompletableFuture<Optional<T>>.asFlowOfNullable(): Flow<T?> = flowOf(this.join().orNull())

基于该实现可配合Axon的queryOptionalKotlin扩展做业务封装：

inline fun <reified R, reified Q> findById(q: Q, qgw: QueryGateway): Flow<R?> {
    return qgw.queryOptional<R, Q>(q).asFlowOfNullable()
}

注意：该方案存在严重的性能缺陷，CompletableFuture.join()是同步阻塞方法，会直接阻塞当前协程运行的线程，不符合协程非阻塞的设计原则，不推荐生产环境使用。

最终非阻塞实现 #

修复阻塞问题后的方案由@Joffrey提供，核心是利用JDK CompletableFuture转协程Deferred的官方扩展，全程非阻塞：

fun <T> Optional<T>.orNull(): T? = orElse(null)

fun <T> CompletableFuture<Optional<T>>.asDeferredOfNullable(): Deferred<T?> = thenApply { it.orNull() }.asDeferred()

如果业务场景确实需要返回Flow<T?>类型，可基于上述Deferred做一层轻量转换：

import kotlinx.coroutines.flow.asFlow

fun <T> CompletableFuture<Optional<T>>.asFlowOfNullable(): Flow<T?> = asDeferredOfNullable().asFlow()

内容的提问来源于stack exchange，提问作者Matthew Adams