У меня была такая же ошибка. Я решил ошибку, запустив Visual Studio 2013 в режиме «работать как администратор», как при использовании проектов веб-приложений, в которых задействован IIS Express.
Потребность в тайм-ауте заставит вас переписать основную часть логики lineStream
. С другой стороны, с такой переписью вы можете избежать промежуточного Iterator
и напрямую вставить строки в Subject
. Для логики тайм-аута вы можете использовать метод Monix timeoutOnSlowUpstream
, но вам все равно придется обработать ошибку тайм-аута и закрыть запущенный процесс.
Также есть выбор, что делать с длинным выходом и несколькими подписчиками. В этом коде я решил использовать ограниченный буфер replayLimited
. В зависимости от ваших потребностей вы можете выбрать другую стратегию. Вот эскиз решения:
object ProcessHelper {
import scala.sys.process.{Process, BasicIO}
import scala.concurrent.duration.FiniteDuration
import monix.eval.Task
import monix.execution.Scheduler
import monix.reactive.subjects.ConcurrentSubject
import monix.reactive.Observable
private class FinishedFlagWrapper(var finished: Boolean = false)
def buildProcessLinesObservable(cmd: String, timeout: FiniteDuration, bufferLines: Int = 100)(implicit scheduler: Scheduler): Observable[String] = {
// works both as a holder for a mutable boolean var and as a synchronization lock
// that is required to preserve semantics of a Subject, particularly
// that onNext is never called after onError or onComplete
val finished = new FinishedFlagWrapper()
// whether you want here replayLimited or some other logic depends on your needs
val subj = ConcurrentSubject.replayLimited[String](bufferLines)
val proc = Process(cmd).run(BasicIO(withIn = false,
line => finished.synchronized {
if (!finished.finished)
subj.onNext(line)
}, None))
// unfortunately we have to block a whole thread just to wait for the exit code
val exitThread = new Thread(() => {
try {
val exitCode = proc.exitValue()
finished.synchronized {
if (!finished.finished) {
finished.finished = true
if (exitCode != 0) {
subj.onError(new RuntimeException(s"Process '$cmd' has exited with $exitCode."))
}
else {
subj.onComplete()
}
}
}
}
catch {
// ignore when this is a result of our timeout
case e: InterruptedException => if(!finished.finished) e.printStackTrace()
}
}, "Process-exit-wait")
exitThread.start()
subj.timeoutOnSlowUpstream(timeout)
.guarantee(Task(finished.synchronized {
if (!finished.finished) {
finished.finished = true
proc.destroy()
exitThread.interrupt()
}
}))
}
}
Пример использования будет выглядеть примерно так:
def test(): Unit = {
import monix.execution.Ack._
import monix.reactive._
import scala.concurrent._
import scala.concurrent.duration._
import monix.execution.Scheduler.Implicits.global
val linesO = ProcessHelper.buildProcessLinesObservable("python3 test.py", 5 seconds, 2) // buffer is reduced to just 2 lines just for this example
linesO.subscribe(new Observer[String] {
override def onNext(s: String): Future[Ack] = {
println(s"Received '$s'")
Future.successful(Continue)
}
override def onError(ex: Throwable): Unit = println(s"Error '$ex'")
override def onComplete(): Unit = println("Complete")
})
try {
println(linesO.toListL.runSyncUnsafe())
println(linesO.toListL.runSyncUnsafe()) // second run will show only last 2 values because of the reduced buffer size
println("Finish success")
}
catch {
case e: Throwable => println("Failed with " + e)
}
}