Как реализовать жизненный цикл PostStop? - PullRequest
Новая
       56

Как реализовать жизненный цикл PostStop?

0 голосов
/ 27 июня 2019

У меня есть актер, реализация которого выглядит следующим образом: 

  def create(fsm: ActorRef[ServerHealth], cancel: Option[Cancellable]): Behavior[ServerHealthStreamer] =
    Behaviors.setup { context =>
      implicit val system = context.system
      implicit val materializer = ActorMaterializer()
      implicit val dispatcher = materializer.executionContext

      val kafkaServer = system
        .settings
        .config
        .getConfig("kafka")
        .getString("servers")

      val sink: Sink[ServerHealth, NotUsed] = ActorSink.actorRefWithAck[ServerHealth, ServerHealthStreamer, Ack](
        ref = context.self,
        onCompleteMessage = Complete,
        onFailureMessage = Fail.apply,
        messageAdapter = Message.apply,
        onInitMessage = Init.apply,
        ackMessage = Ack)

      val cancel = Source.tick(1.seconds, 15.seconds, NotUsed)
        .flatMapConcat(_ => Source.fromFuture(health(kafkaServer)))
        .map {
          case true =>
            KafkaActive
          case false =>
            KafkaInactive
        }
        .to(sink)
        .run()

      Behaviors.receiveMessage {
        case Init(ackTo) =>
          ackTo ! Ack
          Behaviors.same
        case Message(ackTo, msg) =>
          fsm ! msg
          ackTo ! Ack
          create(fsm, Some(cancel))
        case Complete =>
          Behaviors.same
        case Fail(_) =>
          fsm ! KafkaInactive
          Behaviors.same
      }
    }

Как вы можете видеть, я использую akka typed и у меня есть вопрос о реализации жизненного цикла PostStop.

Как реализовать хуки жизненного цикла в акке набрано?

Я пытаюсь заархивировать, когда актер получает сообщение PostStop, тогда поток будет отменен.

1 Ответ

1 голос
/ 03 июля 2019

Актер в Akka Typed уведомляется о событиях жизненного цикла актера как «сигналы», в данном случае сигнал PostStop .

...