Aufhalten eines Prozesses [Aufgabe, O] auf Benutzereingaben

Question

Ich versuche, die folgende Funktion

def haltOnUserInput[O](process: Process[Task, O]): Process[Task, O] 

die process stoppt zu schreiben, wenn der Benutzer eine Zeile auf stdin sendet. In diesem Szenario ist es in Ordnung, zu warten, bis die aktuelle Berechnung im Prozess endet, bevor der Prozess selbst beendet wird.

Ich habe versucht, die folgenden:

scala> :paste 
// Entering paste mode (ctrl-D to finish) 

import scalaz.{ -\/, \/-, \/ } 
import scalaz.stream._ 
import scalaz.concurrent.Task 

def haltOnUserInput[O](process: Process[Task, O]): Process[Task, O] = { 
    process.either(io.stdInLines).flatMap { 
    case -\/(o) => Process.emit(o) 
    case \/-(_) => println("here"); Process.halt 
    } 
} 

Und teste ich wie folgt aus:

scala> val oneSec = scala.concurrent.duration.Duration("1s") 
oneSec: scala.concurrent.duration.Duration = 1 second 

scala> val test = haltOnUserInput(Process.awakeEvery(oneSec)).take(10).map(_.toString).to(io.stdOutLines).run 
test: scalaz.concurrent.Task[Unit] = scalaz.concurrent.Task@7a5e41bb 

scala> test.run 
1000144294 nanoseconds 
2000148316 nanoseconds 
here 
3000130736 nanoseconds 
here 
4000124898 nanoseconds 
5000189134 nanoseconds 
6000201269 nanoseconds 
here 
7000127797 nanoseconds 
8000132194 nanoseconds 
9000191001 nanoseconds 
10000126974 nanoseconds 

Wie Sie die Benutzereingabe sehen können, bestätigt wird („hier“ gedruckt wird, mehrmals) aber der Prozess wird nicht unterbrochen. Ich bin mir nicht sicher, ob sich flatMap wie erwartet verhält in Bezug auf Process.halt.

Irgendwelche Maßnahmen, um haltOnUserInput korrekt zu schreiben?

Answer 1

Eine andere Lösung ist wye.interrupt zu verwenden:

val input = io.stdInLines.take(1).map(_ => true) 
val dory = Process.awakeEvery(1.second).map(_ => println("Hi!")) 
val process = input.wye(dory)(wye.interrupt) 
process.run.run 

Answer 2

Hier ist meine Implementierung von haltOnUserInput:

def haltOnUserInput[O](process: Process[Task, O]): Process[Task, O] = { 
    val stop = Process.constant(()) either io.stdInLines map (_.isRight) 
    process.until(stop) 
    }