Ngăn chặn một Process [Task, O] trên người dùng nhập vào

Question

Tôi cố gắng để viết hàm sau

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

mà dừng process khi người dùng gửi một dòng trên stdin. Trong trường hợp này, bạn có thể đợi quá trình tính toán hiện tại trong quá trình kết thúc, trước khi kết thúc quá trình.

Tôi đã thử những điều sau đây:

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 
    } 
} 

Và tôi đang thử nghiệm như thế:

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 

Như bạn thấy, đầu vào người dùng được thừa nhận ("ở đây" được in, nhiều lần) nhưng Quy trình không bị gián đoạn. Tôi không chắc chắn nếu flatMap hoạt động như dự kiến ​​wrt Process.halt.

Bất kỳ cách nào về cách viết haltOnUserInput chính xác?

Answer 1

Một giải pháp khác là sử dụng wye.interrupt:

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

Đây là triển khai thực hiện của tôi về haltOnUserInput:

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