如何终止SBT Shell启动的FS2流?



如果我在SBT shell中运行这个程序,然后取消它,它将继续打印&;hello&;。我必须退出SBT才能让它停止。为什么呢?

import cats.effect.{ExitCode, IO, IOApp}
import fs2.Stream
import scala.concurrent.duration._
object FS2 extends IOApp {
override def run(args: List[String]) = 
Stream.awakeEvery[IO](5.seconds).map { _ =>
println("hello")
}.compile.drain.as(ExitCode.Error)
}

正如在评论中已经提到的,你的应用程序运行在另一个线程中,它永远不会终止,因为流是无限的,所以你必须在应用程序接收到SIGTERM或SIGINT等信号时手动终止它(每当你点击ctrl+c终止应用程序时,它就会发出)。

你可以这样做:

  1. 创建一个Deferred
  2. 实例
  3. 在接收到TERM或INT信号后触发interruptWhen。例如:

import sun.misc.Signal
object FS2 extends IOApp {
override def run(args: List[String]): IO[ExitCode] = for {
cancel <- Deferred[IO, Either[Throwable, Unit]] //deferred used as flat telling if terminations signal
//was received
_ <- (IO.async_[Unit]{ cb =>
Signal.handle(
new Signal("INT"), //INT and TERM signals are nearly identical, we have to handle both
(sig: Signal) => cb(Right(()))
)
Signal.handle(
new Signal("TERM"),
(sig: Signal) => cb(Right(()))
)
} *> cancel.complete(Right(()))).start //after INT or TERM signal is intercepted it will complete
//deferred and terminate fiber
//we have to run method start to run waiting for signal in another fiber
//in other case program will block here
app <- Stream.awakeEvery[IO](1.seconds).map { _ => //your stream
println("hello")
}.interruptWhen(cancel).compile.drain.as(ExitCode.Error)  //interruptWhen ends stream when deferred completes
} yield app
}

这个版本的应用程序将终止当你在sbt shell中点击ctrl + c

相关内容

  • 没有找到相关文章

最新更新