合并scalaz流输入进程似乎是对stdin的“等待”。

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包含五个as，文件bs.txt包含3个bs。

a
b
b
a
a
b
a
a
a

但是，当我将in2声明更改如下时：

val in2 = io.stdInLines

然后我得到了我认为是意想不到的行为。根据文档1，程序应该不确定地从每个流中根据哪个流更快地提供数据。这应该意味着我看到一堆a立即被打印到控制台上，但这完全不是这样的。

事实上，在我按ENTER之前，什么都不会发生。很明显，如果我随机选择一个流从其中获取下一个元素，那么如果该流阻塞了，那么合并的进程块(即使另一个流包含数据)看起来很像我所期望的那样。

怎么一回事？

1 -好吧，文件很少，但是data在his talk中说得很清楚，它会抓住第一个提供数据的人。

Answer 1

问题在于stdInLines的实现。它是阻塞的，它从来不是Task.fork的线程。

尝试将stdInLines的实现更改为以下内容：

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

原始io.stdInLines在同一个线程中运行readLine()，因此它总是在那里等待，直到您键入某些内容。