如何取消游戏框架内的Akka流

我用 Monix Observable . 通过操作,我可以运行、取消和连接到正在运行的流。总之,我对Akka流解决方案的教育目的感兴趣。这里是monix解决方案:

class StreamService(implicit ec: Scheduler) {

  private val runningStream: AtomicAny[Option[RunningStream]] = AtomicAny(None)

  def run(): Option[Source[ByteString, NotUsed]] =
    runningStream.get match {
      case None =>
        val observable = Observable
          .interval(1.seconds)
          .map(_.toString)
          .doOnTerminate(cb => runningStream.set(None))
          .doOnSubscriptionCancel(() => runningStream.set(None))
          .publish

        val cancelable_ = observable.connect()

        this.runningStream.set(Some(RunningStream(cancelable_, observable)))
        connect()
      case _ => None
    }

  def connect(): Option[Source[ByteString, NotUsed]] =
    runningStream.get
      .map(rs => rs.observable.toReactivePublisher)
      .map(publisher => Source.fromPublisher(publisher).map(ByteString(_)))

  def cancel(): Unit =
    runningStream.get.foreach(_.cancelable.cancel())

}

object StreamService {
  case class RunningStream(cancelable: Cancelable, observable: ConnectableObservable[String])
}


class SomeController @Inject()(streamService: StreamService, cc: ControllerComponents)
  extends AbstractController(cc) {

  def run() = Action {
    val source = streamService.run().getOrElse(throw new RuntimeException("Stream already running"))
    Ok.chunked(source)
  }

  def connect() = Action {
    val source = streamService.connect().getOrElse(throw new RuntimeException("Stream not running"))
    Ok.chunked(source)
  }

  def cancel() = Action {
    streamService.cancel()
    Ok("ok")
  }
}