如何使用 scalaz 流写入和读取外部进程答案

【问题标题】：How do you write to and read from an external process using scalaz streams如何使用 scalaz 流写入和读取外部进程
【发布时间】：2015-05-31 06:54:17
【问题描述】：

我希望能够将数据从 scalaz 流发送到外部程序，然后在未来大约 100 毫秒内获得该项目的结果。虽然我可以使用下面的代码通过将输出流Sink 与输入流Process 压缩然后丢弃Sink 的副作用来做到这一点，但我觉得这个解决方案可能非常脆弱。

如果外部程序有一个输入项错误，所有内容都将不同步。我觉得最好的办法是将某种增量 ID 发送到外部程序中，它可以在将来回显，这样如果发生错误，我们可以重新同步。

我遇到的主要问题是将数据发送到外部程序Process[Task, Unit] 的结果与程序Process[Task, String] 的输出结合在一起。我觉得我应该使用来自 wyn 的东西，但不太确定。

import java.io.PrintStream
import scalaz._
import scalaz.concurrent.Task
import scalaz.stream.Process._
import scalaz.stream._

object Main extends App {
/*
  # echo.sh just prints to stdout what it gets on stdin
  while read line; do
    sleep 0.1
    echo $line
  done
*/
  val p: java.lang.Process = Runtime.getRuntime.exec("/path/to/echo.sh")

  val source: Process[Task, String] = Process.repeatEval(Task{
     Thread.sleep(1000)
     System.currentTimeMillis().toString
  })

  val linesR: stream.Process[Task, String] = stream.io.linesR(p.getInputStream)
  val printLines: Sink[Task, String] = stream.io.printLines(new PrintStream(p.getOutputStream))

  val in: Process[Task, Unit] = source to printLines

  val zip: Process[Task, (Unit, String)] = in.zip(linesR)
  val out: Process[Task, String] = zip.map(_._2) observe stream.io.stdOutLines
  out.run.run
}

【问题讨论】：

标签： scala scalaz scalaz-stream

【解决方案1】：

在深入研究更高级的类型之后。看起来Exchange 完全符合我的要求。

import java.io.PrintStream

import scalaz._
import scalaz.concurrent.Task
import scalaz.stream._
import scalaz.stream.io._

object Main extends App {
/*
  # echo.sh just prints to stdout what it gets on stdin
  while read line; do
    sleep 0.1
    echo $line
  done
*/
  val program: java.lang.Process = Runtime.getRuntime.exec("./echo.sh")

  val source: Process[Task, String] = Process.repeatEval(Task{
     Thread.sleep(100)
     System.currentTimeMillis().toString
  })

  val read: stream.Process[Task, String] = linesR(program.getInputStream)
  val write: Sink[Task, String] = printLines(new PrintStream(program.getOutputStream))
  val exchange: Exchange[String, String] = Exchange(read, write)
  println(exchange.run(source).take(10).runLog.run)
}

【讨论】：