DartのIsolateを使って相互に通信をする例。

Isolate→spawn元(main)の方向の通信は、ReceivePort.sendPortをspawn時に渡してやれば良いが、逆方向の通信(main→Isolate)は、spawnしたIsolate先でReceivePortを作成し、そのsendPortをmainに戻してもらい、そのsendport経由で送信することになる。

sendPortをどうやって戻すかだが、以下ではIsolateに渡すReceivePort経由で受けとっている。(これ以外の良い方法は、グローバル変数にするとか以外にあるかな。どなたかご存知の方教えて!)

そのとき、sendPortをStream.firstで受けとろうとすると、Stream.firstはSteramをlistenし受信後にsubscriptionをcancelしてしまうので、listenerを一回限りしか設定できないsingle-subscription streamでは継続受信できなくなってしまう。なのでStream. asBroadcastStreamでbroadcast streamにしておく必要がある。

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

main() async {
  try {
    var singleChannel = new ReceivePort();
    await Isolate.spawn(echo, singleChannel.sendPort);

    var multiChannel = singleChannel.asBroadcastStream();
    var callbackSendPort = await multiChannel.first;
    multiChannel.listen((msg) {
      print('main: received ${msg}');
      callbackSendPort.send('hoy!');
    });
  } catch (e) {
    print('error ${e}');
  }
}

// the entry point for the isolate
void echo(sendPort) async {

  var callBackReceivePort = new ReceivePort();
  sendPort.send(callBackReceivePort.sendPort);
  callBackReceivePort.listen((msg){
    print('echo: received ${msg}');
  });
  while (true) {
    await new Future.delayed(new Duration(seconds: 1));
    sendPort.send("hello");
  }
}

結果

% dart web/index2.dart
main: received hello
echo: received hoy!
main: received hello
echo: received hoy!
main: received hello
echo: received hoy!
:

sendPortを送り戻してもらう他の方法としては、firstを使わずにストリームで最初に受けとる要素として取り出すという方法もあるだろう。
こんな感じ。

    var callbackSendPort = null;
    multiChannel.listen((msg) {
      if (callbackSendPort == null) {
        callbackSendPort = msg;
      }
      else {
        print('main: received ${msg}');
        callbackSendPort.send('hoy!');
      }
    });

これが嫌なのはまちがいない¹が、asBroadcastStreamも相当である。良い方法があればおしえてください。

追記

https://www.cresc.co.jp/tech/java/Google_Dart/DartLanguageGuide.pdf
18.5節アイソレート間の通信リンクの確立

を見ると、後者の方法に近く、ただ

    var callbackSendPort = null;
    multiChannel.listen((msg) {
      if (msg is SendPort) {
        callbackSendPort = msg;
      }
      else {
        print('main: received ${msg}');
        callbackSendPort.send('hoy!');
      }
    });

に相当するような、実行時の型で切りわけをしていますね。また接続状態を状態遷移でしっかりと管理している。

追記2

StreamIteratorを使うのが良いみたい。asBroadcastStream()が不要であり、型も気持ち守れて、状態変数を導入しなくてよい。

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

main() async {
  try {
    var channel = new ReceivePort();
    await Isolate.spawn(echo, channel.sendPort);

    StreamIterator itr = new StreamIterator(channel);
    if (await itr.moveNext()) {
       SendPort callbackSendPort = itr.current;
       while (await itr.moveNext()) {
         print('main: received ${itr.current}');
         callbackSendPort.send('hoy!');
       }
    }
  } catch (e) {
    print('error ${e}');
  }
}

// the entry point for the isolate
Future echo(sendPort) async {

  var callBackReceivePort = new ReceivePort();
  sendPort.send(callBackReceivePort.sendPort);
  callBackReceivePort.listen((msg){
    print('echo: received ${msg}');
  });
  while (true) {
    await new Future.delayed(new Duration(seconds: 1));
    sendPort.send("hello");
  }
}

こちらも参考に
https://speakerdeck.com/uehaj/dart-isolate-port-and-capabilities

型制約できなくなる気がする。Dartにユニオン型ってあったっけ。 ↩

Dartのアイソレート間で相互通信をする例

追記

追記2