Halting a Process[Task, O] on user input

Question

I am trying to write the following function

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

which halts process when the user sends a line on stdin. In this scenario, it is ok to wait for the current computation in the process to end, before ending the process itself.

I have tried the following:

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

And I am testing like that:

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

As you can see, the user input is acknowledged ("here" is printed, several times) but the Process is not interrupted. I am not sure if flatMap behaves as expected wrt Process.halt.

Any take on how to write haltOnUserInput correctly?

Answer 1

Another solution is to use 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

Here is my implementation of haltOnUserInput:

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