从Zenoh C线程调用Dart回调遇隔离区错误，求解决方案

问题背景与现象 #

开发基于Zenoh协议的Dart应用时，遇到以下问题：

• 基础逻辑正常：Dart通过dart:ffi调用原生C代码、C代码直接回调Dart函数均无问题
• 引入Zenoh订阅者后出错：订阅者收到消息时会在独立线程中触发回调，尝试将消息传递给Dart时出现两个连续错误：
  1. 直接调用Dart回调函数时触发：

     error: Cannot invoke native callback outside an isolate.
     

  2. 尝试用dart:isolate的Send/ReceivePort传递数据时触发：

     error: Dart_NewSendPort expects there to be a current isolate. Did you forget to call Dart_CreateIsolateGroup or Dart_EnterIsolate?
     

核心原因 #

Dart的大部分API要求当前线程必须关联一个Isolate才能调用，但Zenoh的回调线程是原生线程，没有绑定任何Dart Isolate：

• 直接调用Dart回调函数：违反了"Dart回调只能在Isolate线程执行"的规则
• 调用Dart_NewSendPort：该API需要当前线程关联Isolate，而原生线程不满足这个条件

解决思路与修正方案 #

关键原则：使用Dart_PostCObject在原生线程中安全发送消息到Dart Isolate——这是少数不需要当前线程关联Isolate的Dart API之一。

修正步骤 #

1. 在Dart侧保持ReceivePort和SendPort的逻辑不变，确保nativePort传递给C代码
2. 在C侧：
   • 移除错误的Dart_NewSendPort调用，直接使用保存的dart_port
   • 直接调用Dart_PostCObject发送消息，无需创建新的SendPort
   • 若需要传递字符串（如Zenoh的sample数据），需将其封装为Dart_CObject_kString类型

修正后的代码 #

Dart代码（核心逻辑不变）

import 'dart:ffi';
import 'package:ffi/ffi.dart';
import 'dart:isolate';

// typedef Request Callback 
typedef RequestCallbackC = Void Function(Int32 port);
typedef RequestCallbackDart = void Function(int port);

class CWrapperZenoh{
  late final DynamicLibrary _zenohLib;
  late final RequestCallbackDart _requestCallbackDart;
  // for native c callback
  late final ReceivePort _receivePort;
  late final SendPort _sendPort;

  CWrapperZenoh(){
    _zenohLib = DynamicLibrary.open("bin/include/lib_c_lib.so");
    _requestCallbackDart = _zenohLib.lookup<NativeFunction<RequestCallbackC>> 
      ('request_callback').asFunction();

    _receivePort = ReceivePort();
    _sendPort = _receivePort.sendPort;

    // listen
    _receivePort.listen((message) {
      print("listen message\n");
      if (message is String) {
        dartCallbackString(message);
      } else if (message is int){
        print(message);
      }
    });
  }

// application calls this to initiate
callFunctions(){
  print("sendport: ${_sendPort.nativePort}");
  _requestCallbackDart(_sendPort.nativePort);
}

static void dartCallbackString(String s) {
  print("Received string from C: $s\n");
}
}

C代码（关键部分修正）

#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include "zenoh.h"
#include "include/dart_api.h"
#include "include/dart_api_dl.h"
#include <pthread.h>

z_owned_config_t* config_ptr;
z_owned_session_t* sessionPtr;
z_owned_subscriber_t* sub;
Dart_Port dart_port;

void callbackFuncToDart(const z_sample_t *sample, void *arg);

void request_callback(Dart_Port port){
    dart_port = port;

    // config
    config_ptr = malloc(sizeof(z_owned_config_t));
    if (config_ptr != NULL) {
        *config_ptr = zc_config_from_file("bin/include/DEFAULT_CONFIG_CONNECT.json5");
    }
    printf("config created\n");
    fflush(stdout);

    // open session
    sessionPtr = (z_owned_session_t*)malloc(sizeof(z_owned_session_t));
    *sessionPtr = z_open(config_ptr);
    
    // subscribing
    z_owned_closure_sample_t callback = z_closure(callbackFuncToDart);
    sub = malloc(sizeof(z_owned_subscriber_t));
    *sub = z_declare_subscriber(z_loan(*sessionPtr), z_keyexpr("geometry_msgs/msg/actualSpeed"), z_move(callback), NULL);
}

// callback to Dart
void callbackFuncToDart(const z_sample_t *sample, void *arg) {
    // 直接使用保存的dart_port发送消息，无需创建新SendPort
    Dart_CObject message;
    
    // 示例：发送Zenoh的sample字符串数据
    if (sample->payload.len > 0) {
        message.type = Dart_CObject_kString;
        // 将Zenoh的payload转为C字符串（注意：若payload不是UTF-8需额外处理）
        char* str = malloc(sample->payload.len + 1);
        memcpy(str, sample->payload.start, sample->payload.len);
        str[sample->payload.len] = '\0';
        message.value.as_string = str;
    } else {
        message.type = Dart_CObject_kString;
        message.value.as_string = "Empty payload";
    }

    // 发送消息到Dart Isolate
    if (Dart_PostCObject(dart_port, &message)){
        printf("Message sent\n");
        fflush(stdout);
    } else {
        printf("Failed to send message\n");
        fflush(stdout);
    }

    // 释放动态分配的字符串（Dart_PostCObject已复制内容，可安全释放）
    if (message.type == Dart_CObject_kString && message.value.as_string != "Empty payload") {
        free(message.value.as_string);
    }
}

额外注意事项 #

• 确保Dart侧的ReceivePort始终处于活跃状态，避免被GC回收
• 若传递复杂数据结构，需在C侧将其序列化为可传递的Dart_CObject类型（如嵌套数组、结构体可转为JSON字符串）
• Zenoh的payload可能不是UTF-8字符串，需根据实际数据格式处理编码转换

内容的提问来源于stack exchange，提问作者Mads Gadeberg