Scala - получение обратного вызова при выходе внешнего процесса

Я хочу заменить много своего perl scala. Одна из вещей, которые я обычно делаю, - это двоичные файлы вызовов (обычно скомпилированные С++, но могут быть java, другие скрипты perl, q скрипты и т.д.), Предоставленные мне другими группами в моей компании.

Например, чтобы выполнить некоторую сложную математику, я бы начал использовать один из чужих двоичных файлов, а затем подключил к нему свои входы. Затем я слушал поток stdout для результатов и поток stderr для диагностических сообщений. В perl я сделал бы это с помощью виджета POE::Wheel::Run. Я придумал что-то аналогичное (и намного приятнее) в scala, но я хотел бы сделать его более надежным. Это небольшая оболочка вокруг объекта ProcessIO. Это выглядит так:

class Exe(command: String, out: String => Unit, err: String => Unit) {

    import scala.sys.process._
    import scala.io._
    import java.io._
    import scala.concurrent._

    val inputStream = new SyncVar[OutputStream];

    val process = Process(command).run(
        new ProcessIO(
            stdin => inputStream.put(stdin),
            stdout => Source.fromInputStream(stdout).getLines.foreach(out),
            stderr => Source.fromInputStream(stderr).getLines.foreach(err)));

    def write(s: String): Unit = synchronized {
        inputStream.get.write((s + "\n").getBytes)
    }

    def close(): Unit = {
        inputStream.get.close
    }
}

Я бы использовал его следующим образом:

val exe = new Exe("tr [a-z] [A-Z]", 
                  out => println("o: " + out), 
                  err => println("e: " + err))
exe.write("lower")
exe.close()

Что печатает:

o: LOWER

Это дает мне 90%, но было бы неплохо получить обратный вызов, когда процесс завершится. Он может выйти из-за того, что я закрыл входной поток и его внутренний цикл остановился, он может выйти сам по себе, или он может выйти, потому что я его убил. В обратном вызове было бы полезно узнать, почему он остановился, и код выхода.

Я немного похудел на то, как это сделать, любая помощь будет оценена (и любые изменения в этом коде, конечно, приветствуются - я немного нуб).

Я использую 2.9.0.1

Ответы

import scala.concurrent.ops.spawn
implicit def ProcessWithCallback(p: Process) {
  def whenTerminatedDo(callback: Int => Unit) = spawn{
    val exitValue = p.exitValue; callback(p)
  }
}

class Exe(command:String, out:String=>Unit, err:String=>Unit, onExit:Int=>Unit) {

    import scala.sys.process._
    import scala.io._
    import java.io._
    import scala.concurrent._
    import scala.concurrent.ops.spawn

    val inputStream = new SyncVar[OutputStream];

    val process = Process(command).run(
        new ProcessIO(
            stdin => inputStream.put(stdin),
            stdout => Source.fromInputStream(stdout).getLines.foreach(out),
            stderr => Source.fromInputStream(stderr).getLines.foreach(err)));

    spawn { onExit(process.exitValue()) }

    def write(s:String):Unit = synchronized {
        inputStream.get.write((s + "\n").getBytes)
    }

    def close():Unit = {
        inputStream.get.close
    }
}

import java.util.concurrent.CountDownLatch

val latch = new CountDownLatch(1)

val exe = new Exe("tr [a-z] [A-Z]",
        out => println("o: " + out), 
        err => println("e: " + err), 
        code=> {println(code) ; latch.countDown() })
exe.write("lower") 
exe.close()

latch.await

o: LOWER
0