JNI - 在本机 cpp 回调函数中使用 RxJava 的奇怪行为

Posted 2023-02-17

【中文标题】JNI - 在本机 cpp 回调函数中使用 RxJava 的奇怪行为

【英文标题】：JNI - Weird behavior using RxJava in native cpp callback function

【发布时间】：2021-01-18 16:08:10

【问题描述】：

我有一个 c++ 函数，我想从我的 Kotlin 代码中调用它。 那个 c++ 函数获取一个回调函数作为参数，做一些工作并在完成时调用回调。

我之前已经做过几次了，一切都很好。但是，我想以某种方式包装它，而不是传递回调，而是返回一个 Observable，该 Observable 将在调用回调时发出一个值。

我用更简单的代码创建了一个示例。 到目前为止我做了什么：

Kotlin 代码：

fun someFunc(str: String): Observable<String> 
    val subject = PublishSubject.create<String>()
    nativeFunc(object: TestCallback 
        override fun invoke(event: String) 
            println("Callback invoked. subject = $subject")
            subject.onNext("$event - $str")
        
    )
    return subject


private external fun nativeFunc(callback: TestCallback)

Kotlin 中回调函数的接口：

interface TestCallback 
    fun invoke(event: String)

本机 JNI 代码：

extern "C"
JNIEXPORT void JNICALL
Java_com_myProject_TestClass_nativeFunc(JNIEnv *env, jobject thiz, jobject callback) 
    env->GetJavaVM(&g_vm);
    auto g_callback = env->NewGlobalRef(callback);

    std::function<void()> * pCompletion = new std::function<void()>([g_callback]() 
        JNIEnv *newEnv = GetJniEnv();
        jclass callbackClazz = newEnv->FindClass("com/myproject/TestCallback");
        jmethodID invokeMethod = newEnv->GetMethodID(callbackClazz, "invoke", "(Ljava/lang/String;)V");
        string callbackStr = "Callback called";
        newEnv->CallVoidMethod(g_callback, invokeMethod, newEnv->NewStringUTF(callbackStr.c_str()));
        newEnv->DeleteGlobalRef(g_callback);
    );
    pCompletion->operator()(); // <--Similar function is passed to the c++ function. Lets skip that

一起运行的测试函数

@Test
fun testSubject() 
    val testClass = TestClass()
    val someList = listOf("a", "b", "c")
    var done = false
    Observable.concat(someList.map  testClass.someFunc(it) )
        .take(3)
        .doOnNext  println("got next: $it") 
        .doOnComplete  done = true 
        .subscribe()
    while (!done);

测试函数运行 3 次 someFunc 函数（返回一个 Observable 实例，完成时发出一个字符串）并将所有 Observable 连接在一起。

我希望打印的内容：

Callback invoked. subject = io.reactivex.subjects.PublishSubject@1f7acc8
got next: Callback called - a
Callback invoked. subject = io.reactivex.subjects.PublishSubject@7c9b161
got next: Callback called - b
Callback invoked. subject = io.reactivex.subjects.PublishSubject@6f24486
got next: Callback called - c

然而实际结果是：

Callback invoked. subject = io.reactivex.subjects.PublishSubject@1f7acc8
Callback invoked. subject = io.reactivex.subjects.PublishSubject@7c9b161
Callback invoked. subject = io.reactivex.subjects.PublishSubject@6f24486

似乎一切都按预期工作，但是，尽管这条线 println("Callback invoked. subject = $subject") 打印（带有正确的主题地址）， onNext 不工作并且由于某种原因没有发出任何东西。 我检查了没有本机回调的相同功能，一切正常。

有什么建议吗？？？

【问题讨论】：

建议：摆脱 Kotlin 并使用 Java。

这对解决问题有什么帮助？

您是否有可能阻止执行该打印？如果删除 while 循环会发生什么？

@Michael 测试将结束。没有本机代码的类似功能按预期工作。此外，正在调用回调并打印日志，但我没有看到 onNext

的发射

尝试打印$event检查是否有任何错误发生并且未被捕获。顺便说一句，我强烈建议使用 (flows from) kotlinx.coroutines 而不是 (observables from) rx，它更轻量级并且具有更好的 API。

【参考方案1】：

所以经过一番研究，我发现：

当我从 Java 调用 C/C++ 函数时，JNI 不会在后台创建任何新线程。 [see here]。因此， 代码同步运行，这意味着 - 主题发出一个项目，然后函数返回主题并被订阅。所以订阅后，它错过了发射的项目并丢失了它。 我说错了“我检查了没有本机回调的相同功能并且一切正常。”。我可能在那里犯了一个错误，使非本机代码异步，这给了我“准时”返回的主题并按预期打印了日志。

解决方案是将 PublishSubject 更改为 BehaviorSubject 或 ReplaySubject 以缓存发出的项目并在订阅后获取它。 另一种解决方案可能是将调用切换到本机函数以在另一个线程中运行，因此在函数运行时，主题已经返回并被订阅。