액터에 커스텀 영속성 부여하기 by kotlin

class HelloStateStoreActor private constructor(
    private val context: ActorContext<HelloStateStoreActorCommand>,
    private val persistenceId : String,
    private val durableRepository: DurableRepository,
) : AbstractBehavior<HelloStateStoreActorCommand>(context) {

    companion object {
        fun create(persistenceId: String, durableRepository: DurableRepository): Behavior<HelloStateStoreActorCommand> {
            return Behaviors.setup { context -> HelloStateStoreActor(context, persistenceId, durableRepository) }
        }
    }

    private lateinit var helloStoreState: HelloStoreState

    init {
        context.log.info("Create HelloStateStoreActor - $persistenceId")
        initStoreState()
    }

    override fun createReceive(): Receive<HelloStateStoreActorCommand> {
        return newReceiveBuilder()
            .onMessage(HelloStore::class.java, this::onHello)
            .onMessage(HelloLimitStore::class.java, this::onHelloLimit)
            .onMessage(GetHelloCountStore::class.java, this::onGetHelloCount)
            .onMessage(GetHelloTotalCountStore::class.java, this::onGetHelloTotalCount)
            .onMessage(ChangeStateStore::class.java, this::onChangeState)
            .onMessage(ResetHelloCountStore::class.java, this::onResetHelloCount)
            .onSignal(PreRestart::class.java, this::onPreRestart)
            .build()
    }

    private val materializer = Materializer.createMaterializer(context.system)

    private val helloLimitSource = Source.queue<HelloLimitStore>(100, OverflowStrategy.backpressure())
        .throttle(3, Duration.ofSeconds(1))
        .to(Sink.foreach { cmd ->
            context.self.tell(HelloStore(cmd.message, cmd.replyTo))
        })
        .run(materializer)

    private fun onPreRestart(signal: PreRestart): Behavior<HelloStateStoreActorCommand> {
        context.log.info("HelloStateStoreActor actor restart: ")
        val parentPath = context.self.path().parent()
        val parentRef: ActorRef<SupervisorCommand> = ActorRefResolver.get(context.system).resolveActorRef(parentPath.toString())
        parentRef.tell(RestartChild(context.self.path().name(), context.self))

        return this
    }

    private fun onHello(command: HelloStore): Behavior<HelloStateStoreActorCommand> {

        if(command.message == "Crash") {
            throw RuntimeException("Crash")
        }

        when (helloStoreState.happyState) {
            HappyState.HAPPY -> {
                if (command.message == "Hello") {
                    helloStoreState.helloCount++
                    helloStoreState.helloTotalCount++

                    persist(helloStoreState, true)

                    command.replyTo.tell(HelloResponseStore("Kotlin"))
                    context.log.info("onHello-Kotlin - helloCount: ${helloStoreState.helloCount}")
                }
            }
            HappyState.ANGRY -> {
                command.replyTo.tell(HelloResponseStore("Don't talk to me!"))
            }
        }
        return this
    }

    private fun onHelloLimit(command: HelloLimitStore): Behavior<HelloStateStoreActorCommand> {
        helloLimitSource.offer(command).thenAccept { result ->
            when (result) {
                is QueueOfferResult.`Enqueued$` -> context.log.info("Command enqueued successfully")
                is QueueOfferResult.`Dropped$` -> context.log.error("Command dropped")
                is QueueOfferResult.Failure -> context.log.error("Failed to enqueue command", result.cause())
                is QueueOfferResult.`QueueClosed$` -> context.log.error("Queue was closed")
            }
        }
        return this
    }

    private fun onGetHelloCount(command: GetHelloCountStore): Behavior<HelloStateStoreActorCommand> {
        command.replyTo.tell(HelloCountResponseStore(helloStoreState.helloCount))
        context.log.info("onGetHelloCount-helloCount: ${helloStoreState.helloCount}")
        return this
    }

    private fun onGetHelloTotalCount(command: GetHelloTotalCountStore): Behavior<HelloStateStoreActorCommand> {
        command.replyTo.tell(HelloCountResponseStore(helloStoreState.helloTotalCount))
        return this
    }

    private fun onChangeState(command: ChangeStateStore): Behavior<HelloStateStoreActorCommand> {
        helloStoreState.happyState = command.newHappy
        persist(helloStoreState, true)
        return this
    }

    private fun onResetHelloCount(command: ResetHelloCountStore): Behavior<HelloStateStoreActorCommand> {
        context.log.info("Reset hello count")
        helloStoreState.helloCount = 0
        persist(helloStoreState, true)
        return this
    }

    private fun persist(newState: HelloStoreState, isAsync: Boolean = false) {
        if(isAsync) {
            persistAsync(newState)
        }
        else {
            runBlocking {
                durableRepository.createOrUpdateDurableStateEx(persistenceId, 1L, newState).awaitSingle()
            }
        }
    }

    private fun persistAsync(newState: HelloStoreState) {
        Source.single(newState)
            .mapAsync(1) { state ->
                durableRepository.createOrUpdateDurableStateEx(persistenceId, 1L, state).toFuture()
            }
            .runWith(Sink.ignore(), materializer)
    }

    private fun initStoreState() {
        runBlocking {
            var isNew: Boolean = false
            var result = durableRepository.findByIdEx<HelloStoreState>(persistenceId, 1L).awaitFirstOrNull()
            if(result != null) {
                helloStoreState = result
            }
            else{
                isNew = true
                helloStoreState = HelloStoreState(HappyState.HAPPY, 0, 0)
            }

            context.log.info("Init HelloStoreState - isNew: $isNew, $helloStoreState")

        }
    }
}

sealed class HelloStashCommand

//Stash
data class InitialState(val happyState: HappyStashState, val helloCount: Int, val helloTotalCount: Int) : HelloStashCommand()
data class DbError(val error:RuntimeException) : HelloStashCommand()
data class SavaState(val state: HelloStashState, val replyTo:ActorRef<Done>) : HelloStashCommand()
data class GetState(val replyTo:ActorRef<HelloStashState>) : HelloStashCommand()
data object SaveSuccess : HelloStashCommand()



class HelloStashActor private constructor(
    private val context: ActorContext<HelloStashCommand>,
    private val persistenceId : String,
    private val durableRepository: DurableRepository,
    private val buffer: StashBuffer<HelloStashCommand>
) {

    companion object {
        fun create(persistenceId: String, durableRepository: DurableRepository,
                   buffer: StashBuffer<HelloStashCommand>): Behavior<HelloStashCommand> {

            return Behaviors.withStash(100, {
                    Behaviors.setup {
                        context ->
                            context.pipeToSelf(
                                durableRepository.findByIdEx<HelloStashState>(persistenceId, 1L).toFuture(),
                                { value, cause ->
                                    if (cause == null) {
                                        if(value == null) {
                                            // First State
                                            InitialState(HappyStashState.HAPPY, 0, 0)
                                        } else {
                                            InitialState(value.happyState, value.helloCount, value.helloTotalCount)
                                        }
                                    } else {
                                        DbError(RuntimeException(cause))
                                    }
                                })

                            HelloStashActor(context, persistenceId, durableRepository, buffer).start()
                    }
                })
        }
    }

    //private lateinit var buffer: StashBuffer<HelloStashCommand>

    init {
        context.log.info("Create HelloStashStoreActor")

    }

    private fun start() : Behavior<HelloStashCommand> {
        return Behaviors.receive(HelloStashCommand::class.java)
            .onMessage(InitialState::class.java, this::onInitialState)
            .onMessage(DbError::class.java, this::onDBError)
            .onMessage(HelloStashCommand::class.java, this::stashOtherCommand)
            .build()
    }

    private fun onInitialState(command: InitialState): Behavior<HelloStashCommand> {
        var state:HelloStashState = HelloStashState(command.happyState, command.helloCount, command.helloTotalCount)
        return buffer.unstashAll(active(state));
    }

    private fun onDBError(command: DbError): Behavior<HelloStashCommand> {
        throw command.error;
    }

    private fun stashOtherCommand(command: HelloStashCommand): Behavior<HelloStashCommand> {
        buffer.stash(command)
        return Behaviors.same()
    }

    private fun active(state:HelloStashState) : Behavior<HelloStashCommand>{
        return Behaviors.receive(HelloStashCommand::class.java)
            .onMessage(GetState::class.java, {message -> onGetState(state, message)})
            .onMessage(SavaState::class.java, this::onSaveState)
            .build()
    }

    private fun onGetState(state:HelloStashState, message: GetState): Behavior<HelloStashCommand> {
        message.replyTo.tell(state)
        return Behaviors.same()
    }

    private fun onSaveState( message: SavaState): Behavior<HelloStashCommand> {
        context.pipeToSelf(
            durableRepository.createOrUpdateDurableStateEx<HelloStashState>(persistenceId, 1L, message.state).toFuture(),
            { value, cause ->
                if (cause == null) {
                    message.replyTo.tell(Done.getInstance())
                    SaveSuccess
                } else {
                    DbError(RuntimeException(cause))
                }
            })
        return saving(message.state, message.replyTo)
    }


    private fun saving(state:HelloStashState, replyTo:ActorRef<Done>) : Behavior<HelloStashCommand> {
        return Behaviors.receive(HelloStashCommand::class.java)
            .onMessage(SaveSuccess::class.java, {message -> onSaveSucess(state, replyTo)})
            .onMessage(DbError::class.java, this::onDBError)
            .onMessage(HelloStashCommand::class.java, this::stashOtherCommand)
            .build()
    }

    private fun onSaveSucess(state:HelloStashState, replyTo: ActorRef<Done>): Behavior<HelloStashCommand> {
        replyTo.tell(Done.getInstance())
        return buffer.unstashAll(active(state));
    }
}