dart - 异步 future 流订阅错误

Question

有人可以解释以下代码有什么问题。我对函数 fInputData 进行了两次调用。第一个工作正常，第二个导致错误:
“未处理的异常”
“坏状态:流已经有订阅者”

我需要编写一个输入多个参数的测试控制台程序。

import "dart:async" as async;
import "dart:io"; 

void main() {
  fInputData ("Enter Nr of Iterations : ")
  .then((String sResult){
    int iIters;
    try {
      iIters = int.parse(sResult);
      if (iIters < 0) throw new Exception("Invalid");
    } catch (oError) {
      print ("Invalid entry");
      exit(1);
    }

    print ("In Main : Iterations selected = ${iIters}");

    fInputData("Continue Processing? (Y/N) : ")   // this call bombs
    .then((String sInput){
      if (sInput != "y" && sInput != "Y")
        exit(1);

      fProcessData(iIters);

      print ("Main Completed");
    });
  });
}

async.Future<String> fInputData(String sPrompt) {
  async.Completer<String> oCompleter = new async.Completer(); 

  stdout.write(sPrompt);
  async.Stream<String> oStream = stdin.transform(new StringDecoder());
  async.StreamSubscription oSub;
  oSub = oStream.listen((String sInput) {
    oCompleter.complete(sInput);
    oSub.cancel();
  });
  return oCompleter.future;
}

void fProcessData(int iIters) {
  print ("In fProcessData");
  print ("iIters = ${iIters}");

  for (int iPos = 1; iPos <= iIters; iPos++ ) {
    if (iPos%100 == 0) print ("Processed = ${iPos}");
  }
  print ("In fProcessData - completed ${iIters}");
} 

Answer 1

一些 background reading :

Streams comes in two flavours: single or multiple (also known as broadcast) subscriber. By default, our stream is a single-subscriber stream. This means that if you try to listen to the stream more than once, you will get an exception, and using any of the callback functions or future properties counts as listening.

You can convert the single-subscriber stream into a broadcast stream by using the asBroadcastStream() method.

所以你有两个选择——要么重用一个订阅对象。即调用一次监听，并使订阅对象保持事件状态。

或者使用广播流 - 请注意广播流和单订阅者流之间存在许多差异，您需要阅读这些内容并确保它们适合您的用例。

这是一个重用订阅者提出多个问题的示例:

import 'dart:async';
import 'dart:io';

main() {
    var console = new Console();
    var loop;
    loop = () => ask(console).then((_) => loop());
    loop();
}

Future ask(Console console) {
    print('1 + 1 = ...');
    return console.readLine().then((line) {
        print(line.trim() == '2' ? 'Yup!' : 'Nope :(');
    });
}

class Console {
    StreamSubscription<String> _subs;

    Console() {
        var input = stdin
            .transform(new StringDecoder())
            .transform(new LineTransformer());

        _subs = input.listen(null);
    }

    Future<String> readLine() {
        var completer = new Completer<String>();
        _subs.onData(completer.complete);
        return completer.future;
    }
}