Как остановить процесс ScalaZ, созданный time.awakeEvery?

Question

Я узнал, что по scalaz.stream.time.awakeEvery(1.second) Я могу создать процесс, который создает событие каждую секунду. Совершенно очевидно. Затем я смогу сопоставить этот процесс для выполнения каждой задачи каждую секунду. Все идет нормально.

Что делать, если я хочу прекратить эту деятельность? Я пробовал p.kill.run.run, где p - это процесс, созданный time.awakeEvery(1.second), но он не работает.

Так что код выглядит следующим образом:

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") 

Тем не менее, он печатает время после выполнения печати «Убитые» и продолжается и навсегда.

Как остановить процесс и освободить используемые им ресурсы (потоки)? Я узнал, что могу остановить его, отключив ScheduledThreadPoolExecutor, но разве нет более элегантного способа?

Answer 1

Вы можете взглянуть на комбинатор runAsyncInterruptibly на scalaz.concurrent.Task. Вы можете передать ему AtomicBoolean, который затем отменит выполнение, если установлено значение true, или вернуть ему функцию, которая при вызове отменяет выполнение.

Использование 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")