我正在重构一个Flutter类，使其在一个与UI分离的 Isolate 中运行。该类在构造时已经接收两个用于与调用方通信的 StreamController，一个用于输入，一个用于输出。重构之前，与调用方的通信在这里提供的简单代码中就能正常工作，在我的应用中也一样（其中流由 StreamBuilders消费）。

我本以为可以重用在调用方构造的 StreamController，以在单独的Isolate中与对象通信，从而避免重复代码，但这并不奏效。没有异常也没有错误。消息根本没传过来。

文档（https://dart.dev/language/isolates#receiveport-and-sendport）指出：

端口的行为与Stream对象类似（事实上，接收端口实现了Stream！）。你可以把SendPort和 ReceivePort看作是Stream的 StreamController和监听器，分别对应。SendPort就像一个StreamController，因为你可以用SendPort.send() 向它们“添加”消息，而这些消息由监听器处理，在本例中即为ReceivePort。ReceivePort再通过把它收到的消息作为参数传递给你提供的回调来处理。

我的问题是，是否可以以某种方式把我现有的 StreamController 转换成 SendPort 和 ReceivePort，从而避免重复代码？

import 'dart:async';
import 'dart:isolate';

void main() {
  // set it to true to run in Isolate
  const bool RUN_ISOLATE = true;

  // inSC: sending into Heavy
  final StreamController<HeavyMsg> inSC = StreamController<HeavyMsg>();
  // outSC: receiving from Heavy
  final StreamController<HeavyMsg> outSC = StreamController<HeavyMsg>();

  if (RUN_ISOLATE) {
    // subscribe to Heavy's output Stream and print what it sends
    outSC.stream.listen((msg) {
      print("main_isolate() : got message from Heavy: ${msg}");
    });
    // send it a message
    inSC.sink.add(HeavyMsg("a message from main_isolate()"));
    Heavy.constructAndGoIsolate(inSC, outSC);
  } else {
    // subscribe to Heavy's output Stream and print what it sends
    outSC.stream.listen((msg) {
      print("main_plain() : got message from Heavy: ${msg}");
    });
    Heavy hobj = Heavy(inSC, outSC);
    inSC.sink.add(HeavyMsg("a message from main_plain()"));
    hobj.go();
  }
}

class Heavy {
  // Stream for receiving from caller
  final StreamController<HeavyMsg> inSC;

  // Stream for sending to caller
  final StreamController<HeavyMsg> outSC;

  /* Constructor, parameters are the 2 StreamController's
     for communicating with the caller.
     The StreamControllers are constructed in the caller */
  Heavy(this.inSC, this.outSC) {
    print("Heavy() : constructor called ...");
    inSC.stream.listen((msg) {
      print("Heavy: received this message: ${msg}");
    });
  }

  /* the heavy calculations are done here and before, during and after
     messages are exchanged with the caller */
  void go() {
    print("Heavy.go() : called your should receive some data ...");
    outSC.sink.add(HeavyMsg('starting the computation'));
    /* insert heavy computation */
    outSC.sink.add(HeavyMsg('finished the computation'));
  }

  /* above this point is the original functionality without Isolates */

  /* below this point is the refactoring code added
     to using Isolates uses these two methods */
  static Future<void> constructAndGoIsolate(
      StreamController<HeavyMsg> inSC, StreamController<HeavyMsg> outSC) async {
    await Isolate.spawn<IsolateParams>(
        _constructAndGoIsolateHelper, IsolateParams(inSC, outSC));
  }

  static void _constructAndGoIsolateHelper(IsolateParams params) {
    Heavy hobj = Heavy(params.inSC, params.outSC);
    params.outSC.sink
        .add(HeavyMsg("a message from _constructAndGoIsolateHelper()"));
    hobj.go();
  }
}

/* The Message class for communicating via the Stream with the class */
class HeavyMsg {
  final String msg;

  HeavyMsg(this.msg);

  @override
  toString() {
    return "HeavyMsg: ${msg}";
  }
}

/* Class to provide parameters to the spawned Isolate function */
class IsolateParams {
  // basically we are supplying two StreamController's for communicating
  // with the caller
  final StreamController<HeavyMsg> inSC;
  final StreamController<HeavyMsg> outSC;

  IsolateParams(this.inSC, this.outSC);
}

附言：上述代码可以通过命令行使用 dart x.dart 运行。

解决方案

你可以使用来自 stream_channel 包的 IsolateChannel 类。

示例：

import 'dart:async';
import 'dart:isolate';

import 'package:multitasking/multitasking.dart';
import 'package:multitasking/work/isolated_work.dart';
import 'package:stream_channel/isolate_channel.dart';

Future<void> main() async {
  var recv = ReceivePort();
  var recvChannel = IsolateChannel<String>.connectReceive(recv);
  final sink = recvChannel.sink;
  final stream = recvChannel.stream;
  final sendPort = recv.sendPort;
  final work = IsolatedWork.withArgument(sendPort, _computation);

  unawaited(() async {
    try {
      await work.run();
    } on CancellationException {
      print('IsolatedWork terminated');
    }
  }());

  sink.add('Hello, I am parent');
  stream.listen((event) {
    print(event);
    if (event.contains('Hello')) {
      sink.add('Goodbye child');
    }
  });

  Timer(Duration(seconds: 2), () {
    print('Terminating work');
    work.terminate(force: true);
    print('Closing everything');
    recv.close();
    sink.close();
  });
}

Future<void> _computation(SendPort sendPort) async {
  var sendChannel = IsolateChannel<String>.connectSend(sendPort);
  final sink = sendChannel.sink;
  final stream = sendChannel.stream;
  stream.listen((event) {
    print(event);
    if (event.contains('Hello')) {
      sink.add('Hello, I am child');
    } else if (event.contains('Goodbye')) {
      sink.add('Goodbye parent');
    }
  });

  await Completer<void>().future;
}

结果：

Hello, I am parent
Hello, I am child
Goodbye child
Goodbye parent
Terminating work
Closing everything
IsolatedWork terminated