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

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，指示何时完成添加流。应该只有一个流同时addStreams 到StreamSink。

根据您想要/需要做什么，您现在有 2 个选择：

• 将进程的输出多路复用到标准输出中。
• 等待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;
});

不过，它不会交错 stdout 和 stderr。

Answer 2

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

要将两个（或更多）流交叉添加到同一个接收器，您必须手动执行此操作。一种方法是使用一种通用的方法来交错流，如以下代码。另一种选择是自己监听两个流并将事件添加到接收器：

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 = ...
}