Flutter Rust Bridge中的Stream内存泄漏问题分析

问题背景

在Flutter Rust Bridge项目中，当使用create_stream创建流时，存在一个潜在的内存泄漏问题。这个问题源于ReceivePort在特定情况下未能正确关闭，导致资源无法释放。

问题本质

当前实现中，ReceivePort的关闭逻辑被放置在接收到"done"事件的处理分支中。然而，如果Dart端在接收到"done"事件前取消了订阅，ReceivePort将永远不会被关闭，从而造成内存泄漏。

技术细节分析

在当前的实现中，流执行的核心代码如下：

Stream<S> _executeStreamInner<S, E extends Object>(StreamTask<S, E>? task) async* {
  final portName = ExecuteStreamPortGenerator.create(task!.constMeta.debugName);
  final receivePort = broadcastPort(portName);
  
  task.callFfi(receivePort.sendPort.nativePort);
  
  final codec = task.codec;
  task = null;
  
  await for (final raw in receivePort) {
    try {
      yield codec.decodeObject(raw);
    } on CloseStreamException {
      receivePort.close();  // 问题点：仅在此处关闭
      break;
    }
  }
}

这段代码存在两个主要问题：

1. 资源泄漏风险：当订阅被取消时，ReceivePort不会被关闭
2. 竞态条件：如果尝试在finally块中关闭端口，可能导致Rust端仍在发送数据时端口已关闭

更深层次的问题

除了内存泄漏外，当前的流实现还存在另一个潜在问题：由于使用了async*语法，实际的Rust create_stream调用是异步执行的。这可能导致在流尚未准备好时，Rust端就尝试发送事件，造成数据丢失或错误。

例如，以下代码可能会失败：

final subscription = s.listen((event) {
  print("event: ${event.msg}");
});

print("sending 1");
sendEvent(msg: "msg 1");  // 可能在流准备好前就执行

解决方案思路

要彻底解决这些问题，需要考虑以下几个方面：

1. 资源管理：确保在任何情况下都能正确关闭ReceivePort
2. 同步初始化：确保流在监听前已经完全初始化
3. 两端同步：保持Dart和Rust端的生命周期同步

一个可能的改进方向是重构流的创建逻辑，将关键部分移出async*函数，确保同步执行必要的初始化操作。同时，需要设计更健壮的资源释放机制，确保在各种情况下都能正确清理资源。

总结

Flutter Rust Bridge中的流实现虽然功能强大，但在资源管理和初始化顺序上存在一些需要改进的地方。通过深入分析这些问题，我们可以更好地理解跨语言通信中的资源管理挑战，并为类似场景下的开发提供有价值的参考经验。