How to stop ScalaZ Process created by time.awakeEvery?

Question

I learned that by scalaz.stream.time.awakeEvery(1.second) I can create a process that creates an event every one second. Quite obvious. I can then map that process to accomplish some task every second. So far so good.

What if I want to stop this activity? I tried p.kill.run.run, where p is the process created by time.awakeEvery(1.second), but it doesn't work.

So the code looks like this:

implicit val sc = new java.util.concurrent.ScheduledThreadPoolExecutor(1)
val p = time.awakeEvery(1.second)
p.map(println(_)).run.runAsync {
  // just some printlines
}

Thread.sleep(4000)
println("Killing")
p.kill.run.run
println("Killed")

Still, it prints time after printing "Killed" and goes on and on forever.

How can I stop the process, and free the resources (threads) used by it? I found out I can stop it by shutting down the ScheduledThreadPoolExecutor, but isn't there some more elegant way?

Answer 1

You might take a look at the runAsyncInterruptibly combinator on scalaz.concurrent.Task. You can pass it an AtomicBoolean, which will then cancel execution when set to true, or have it return a function that when called will cancel execution.

Using an AtomicBoolean:

implicit val sc = new java.util.concurrent.ScheduledThreadPoolExecutor(1)

val cancel = new java.util.concurrent.atomic.AtomicBoolean(false)

val p = time.awakeEvery(1.second)
p.map(println).run.runAsyncInterruptibly(_ => (), cancel)

Thread.sleep(4000)
println("Killing")
cancel.set(true)
println("Killed")