scala - 合并 scalaz-stream 输入进程在 stdin 上似乎为 "wait"

Question

我有一个简单的程序:

import scalaz._
import stream._

object Play extends App {
  val in1 = io.linesR("C:/tmp/as.txt")
  val in2 = io.linesR("C:/tmp/bs.txt")

  val p = (in1 merge in2) to io.stdOutLines
  p.run.run
}

文件 as.txt包含五个 a s 和文件 bs.txt包含 3 b s。我看到这种输出:

a
b
b
a
a
b
a
a
a

但是，当我更改 in2 的声明时如下:

val in2 = io.stdInLines

然后我得到了我认为意外的行为。根据文档 1 ，程序应该根据哪个流更快地提供数据从每个流中非确定性地提取数据。这应该意味着我看到一堆 a s 立即打印到控制台，但这根本不是发生的事情。

确实，直到我按下 ENTER ， 没发生什么事。很明显，如果我随机选择一个流来从中获取下一个元素，那么行为看起来很像我期望的那样，如果该流被阻塞，合并的进程也会阻塞(即使另一个流包含数据)。

到底是怎么回事？

1 - 好吧，好吧，文档很少，但Dan Spiewak 在 his talk 中说的很清楚它会捕获第一个提供数据的人

Answer 1

问题出在stdInLines的实现上.它是阻塞的，它从不 Task.fork s 一个线程。

尝试更改 stdInLines 的实现对这个:

def stdInLines: Process[Task,String] =
    Process.repeatEval(Task.apply { 
    Option(scala.Console.readLine())
    .getOrElse(throw Cause.Terminated(Cause.End))
})

原版 io.stdInLines正在运行 readLine()在同一个线程中，所以它总是在那里等待，直到你输入一些东西。