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

如何解决JNI - 在原生 cpp 回调函数中使用 RxJava 的奇怪行为

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

我之前已经做过几次了，一切都很好。但是，我想以某种方式包装它，而不是传递回调，而是返回一个可在调用回调时发出值的 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);
}

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

我希望打印的内容：

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 不起作用并且由于某种原因不发出任何内容。我在没有原生回调的情况下检查了相同的功能，一切正常。

有什么建议？？？

解决方法

所以经过一些研究我发现：

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

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