从标准输出中删除一个流以便添加另一个流的正确方法是什么?

Question

我正在Dart中启动一个进程，该进程将其标准输出流附加到标准输出，以便可以将结果打印到终端，如下所示:

Process.start(executable, ['list','of','args']).then((proc) {
  stdout.addStream(proc.stdout);
  stderr.addStream(proc.stderr);
  return proc.exitCode;
});

然而，一旦完成，我想开始一个新的进程，并再次开始(这个函数将被调用几次)。有时，我得到一个错误:

Uncaught Error: Bad State: StreamSink is already bound to a stream

看dart文档，看起来我可能需要做一些像stdout.close()或stdout.flush()的事情，但这些似乎不能解决问题。什么是正确的方式来处理有多个流序列绑定到一个流汇?

Answer 1

addStream返回一个Future，表示流的添加何时完成。只能有一个流，addStream在同一时间对StreamSink。

根据你想要/需要做的事情，你现在有两个选择:

• 将进程的输出多路复用到标准输出。
• 等待addStream完成。

后者比较容易:

Process.start(executable, ['list','of','args']).then((proc) async {
  await stdout.addStream(proc.stdout);  // Edit: don't do this.
  await stdout.addStream(proc.stderr);
  return proc.exitCode;
});

注意句体上的async修饰语，以及句体中的两个await修饰语。

编辑:不立即收听stderr是错误的。(你的程序可能会阻塞它)。

如果你的程序输出足够小，你可以直接切换到Process.run:

Process.run(executable, ['list','of','args']).then((procResult) {
  stdout.write(procResult.stdout);
  stdout.write(procResult.stderr);
  return procResult.exitCode;
});

它不会让标准输出和标准错误交错。

Answer 2

您不能在同一接收器上一次多次调用addStream。接收器要么处于"手动"模式，要么处于"自动"模式，后者通过添加流来触发。在完成添加流之前，暂停将被路由到正在添加的流而不是控制器，并且在完成之前不允许手动添加事件。这就好像添加的流接管了汇，直到它完成。

要将两个(或更多)流交叉添加到同一个sink，您必须手动执行。一种方法是使用一种通用的方法来交错流，如下面的代码所示。另一个选择是自己监听两个流，并将事件添加到接收器:

Stream interleave(Iterable<Stream> streams) {
  List subscriptions = [];
  StreamController controller;
  controller = new StreamController(
    onListen: () {
      int active = 0;
      void done() {
        active--;
        if (active <= 0) controller.close();
      }
      for (var stream in streams) {
        active++;
        var sub = stream.listen(controller.add, 
                                onError: controller.addError,
                                onDone: done);
        subscriptions.add(sub);             
      }
    },
    onPause: () {
      for (var sub in subscriptions) { sub.pause(); }
    }, 
    onResume: () {
      for (var sub in subscriptions) { sub.resume(); }
    },
    onCancel: () {
      for (var sub in subscriptions) { sub.cancel(); }  
    }
  );
  return controller.stream;
}

(未彻底测试!)

Answer 3

您也可以使用

StreamSubscription subscr = proc.stdout.listen(io.stdout.add);
...
subscr.cancel();

或配置处理程序

class StdOutHandler { 
  IOSink sink; 
  call(data) => sink.add(); 
}
void main() {
  var proc = await Process.start(...);
  var stdoutHandler = new StdOutHandler()..sink = stdout;
  var subscription = proc.stdout.listen(stdoutHandler);
  ....
  stdoutHandler.sink = ...
}