Kotlin流Java互操作回调

Question

当我想将Kotlin Flows与普通回调一起使用时，我一直在寻找合适的解决方案或最佳实践。我的用例是，我编写了一个内部使用kotlin Flow的kotlin库，并且我必须假设用户将使用Java。因此，我认为最好的解决方案是将一个基本的回调接口重载到我的流方法中，并在collect中调用它，如下所示：

class KotlinClass {
interface Callback {
fun onResult(result: Int)
}
private fun foo() = flow {
for (i in 1..3) {
emit(i)
}
}
fun bar(callback: Callback) {
runBlocking {
foo().collect { callback.onResult(it) }
}
}
private fun main() {
bar(object : Callback {
override fun onResult(result: Int) {
TODO("Not yet implemented")
}
})
}

在我的Java应用程序中，我可以简单地使用它：

public class JavaClass {
public void main() {
KotlinClass libraryClass = new KotlinClass();
libraryClass.bar(new KotlinClass.Callback() {
@Override
public void onResult(int result) {
// TODO("Not yet implemented")
}
});
} 
}

我不确定该怎么办，因为我想让我的Kotlin库以良好的方式使用Java和Kotlin的Flows。

我遇到了callbackFlow，但这似乎只是在我想让我们称之为流的情况下——实现一个基于回调的API？因为我是Kotlin和Flows的新手，如果我的问题有缺陷，导致我错过了Kotlin的一些基本概念，请道歉。

Answer 1

我会让Java客户端对流有更多的控制权。我会在回调接口中添加一个onStart和onCompletion方法。除此之外，我还将使用自己的CoroutineScope——也许可以从Java客户端进行自定义。而且我不会阻止Kotlin函数中的调用线程——没有runBlocking。

@InternalCoroutinesApi
class KotlinClass {
val coroutineScope = CoroutineScope(Dispatchers.Default)
interface FlowCallback {
@JvmDefault
fun onStart() = Unit
@JvmDefault
fun onCompletion(thr: Throwable?) = Unit
fun onResult(result: Int)
}
private fun foo() = flow {
for (i in 1..3) {
emit(i)
}
}
fun bar(flowCallback: FlowCallback) {
coroutineScope.launch {
foo().onStart { flowCallback.onStart() }
.onCompletion { flowCallback.onCompletion(it) }
.collect { flowCallback.onResult(it) }
}
}
fun close() {
coroutineScope.cancel()
}    
}

现在，Java客户端完全可以控制如何启动、收集和取消流。例如，您可以使用闩锁来等待完成，设置超时并取消couroutine范围。首先，这看起来像是很多代码，但通常您需要这种灵活性。

public class JavaClass {
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
KotlinClass libraryClass = new KotlinClass();
libraryClass.bar(new KotlinClass.FlowCallback() {
@Override
public void onCompletion(@Nullable Throwable thr) {
latch.countDown();
}

@Override
public void onResult(int result) {
System.out.println(result);
}
});
try {
latch.await(5, TimeUnit.SECONDS);
} finally {
libraryClass.close();
}
}
}

Answer 2

您不需要在Kotlin代码中创建接口。你可以这样定义条：

fun bar(callback: (Int) -> Unit) {
runBlocking {
foo().collect { callback(it) }
}
}

从Java代码中，您可以调用这样的函数：

public class JavaClass {
public static void main(String[] args) {
KotlinClass libraryClass = new KotlinClass();
libraryClass.bar(v -> { System.out.println(v); return Unit.INSTANCE; });
} 
}

Answer 3

如果有人想知道一个通用的解决方案。以下是我们对@rene增强的回答。

1. 接受泛型类型
2. 可配置的coroutineScope

// JavaFlow.kt
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.InternalCoroutinesApi
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.launch
@InternalCoroutinesApi
class JavaFlow<T>(
private val coroutineScope: CoroutineScope = CoroutineScope(Dispatchers.Default)
) {
interface OperatorCallback <T> {
@JvmDefault
fun onStart() = Unit
@JvmDefault
fun onCompletion(thr: Throwable?) = Unit
fun onResult(result: T)
}
fun collect(
flow: Flow<T>,
operatorCallback: OperatorCallback<T>,
) {
coroutineScope.launch {
flow
.onStart { operatorCallback.onStart() }
.onCompletion { operatorCallback.onCompletion(it) }
.collect { operatorCallback.onResult(it) }
}
}
fun close() {
coroutineScope.cancel()
}
}

Java调用方：

// code omitted...
new JavaFlow<File>().collect(
// compressImageAsFlow is our actual kotlin flow extension
FileUtils.compressImageAsFlow(file, activity),
new JavaFlow.OperatorCallback<File>() {
@Override
public void onResult(File result) {
// do something with the result here
SafeSingleton.setFile(result);
}
}
);

// or using lambda with method references
// new JavaFlow<File>().collect(
//        FileUtils.compressImageAsFlow(file, activity),
//        SafeSingleton::setFile
// );
// Change coroutineScope to Main
// new JavaFlow<File>(CoroutineScopeKt.MainScope()).collect(
//        FileUtils.compressImageAsFlow(file, activity),
//        SafeSingleton::setFile
// );

OperatorCallback.onStart和OperatorCallback.onCompletion是可选的，请根据需要覆盖它。