Page History

Versions Compared

Key

This line was added.
This line was removed.
Formatting was changed.

액터에게 이벤트를 보낸후 응답값을 수신 받으려면 REST에서는 그 결과를 대기해야 합니다.

공식적으로는 ASK패턴으로 액터에게 메시지를 보내는 패턴은 크게는 두가지가 있습니다.

tell : send이후 응답값이 없어 just tell하는 경우
ask : send이후 응답값을 대기하는경우

자바가 기본 제공하는 CompletionStage가 맘에 안들기도 해서, ASK 패턴을 4가지 방식을 이용하는 코드를 작성 코르틴을 위한 유틸 클래스도 생성

Akka HTTP를 사용하면 더 심플하게 이용할수 있지만 코프링기반에 취향에 맞는 방식으로 통일할수 있습니다.

CompletionStage
AkkaStream
Async ( 코르틴 )
Mono ( WebFlux )

Code Block

theme	Emacs

@RestController
@RequestMapping("/api/admin/channel")
class AdminChannelController(private val actorSystem: ActorSystem<MainStageActorCommand>,
                             private val supervisorChannelActor: ActorRef<SupervisorChannelCommand>) {

    private val timeout: Duration = Duration.ofSeconds(5)

    @GetMapping("/list-counselor-managers")
    fun listCounselorManagers(): CompletionStage<List<String>>? {
        return AskPattern.ask(
            supervisorChannelActor,
            { replyTo: ActorRef<SupervisorChannelResponse> -> GetAllCounselorManagers(replyTo) },
            timeout,
            actorSystem.scheduler()
        ).thenApply { response ->
            when (response) {
                is AllCounselorManagers -> response.channels
                else -> emptyList()
            }
        }
    }

    @GetMapping("/list-counselor-managers-stream")
    fun listCounselorManagersByStream(): CompletionStage<List<String>> {
        return Source.single(Unit)
            .mapAsync(1) {
                AskPattern.ask(
                    supervisorChannelActor,
                    { replyTo: ActorRef<SupervisorChannelResponse> -> GetAllCounselorManagers(replyTo) },
                    timeout,
                    actorSystem.scheduler()
                )
            }
            .runWith(Sink.head(), actorSystem)
            .thenApply { response ->
                when (response) {
                    is AllCounselorManagers -> response.channels
                    else -> emptyList()
                }
            }
    }

    @GetMapping("/list-counselor-managers-async")
    fun listCounselorManagersByCoroutines(): List<String> {
        val response = AkkaUtils.runBlockingAsk(
            supervisorChannelActor,
            { replyTo: ActorRef<SupervisorChannelResponse> -> GetAllCounselorManagers(replyTo) },
            timeout, actorSystem
        )

        return when (response) {
            is AllCounselorManagers -> response.channels
            else -> emptyList()
        }
    }

    @GetMapping("/list-counselor-managers-mono")
    fun listCounselorManagersByMono(): Mono<List<String>> {
        return AkkaUtils.askActorByMono(
            supervisorChannelActor,
            { replyTo: ActorRef<SupervisorChannelResponse> -> GetAllCounselorManagers(replyTo) },
            timeout,
            actorSystem
        ).map { response ->
            when (response) {
                is AllCounselorManagers -> response.channels
                else -> emptyList()
            }
        }
    }

더 줄이고 싶지만 여기까지

Code Block

theme	Emacs

package com.example.kotlinbootlabs.module

import akka.actor.typed.ActorRef
import akka.actor.typed.ActorSystem
import akka.actor.typed.javadsl.AskPattern
import java.time.Duration
import kotlinx.coroutines.future.await
import kotlinx.coroutines.runBlocking

object AkkaUtils {
    suspend fun <T, R> askActor(
        actor: ActorRef<T>,
        message: (ActorRef<R>) -> T,
        timeout: Duration,
        actorSystem: ActorSystem<*>
    ): R {
        return AskPattern.ask(
            actor,
            message,
            timeout,
            actorSystem.scheduler()
        ).await()
    }

    fun <T, R> runBlockingAsk(
        actor: ActorRef<T>,
        message: (ActorRef<R>) -> T,
        timeout: Duration,
        actorSystem: ActorSystem<*>
    ): R = runBlocking {
        askActor(actor, message, timeout, actorSystem)
    }


    fun <T, R> askActorByMono(
        actor: ActorRef<T>,
        message: (ActorRef<R>) -> T,
        timeout: Duration,
        actorSystem: ActorSystem<*>
    ): Mono<R> {
        return Mono.fromCompletionStage(
            AskPattern.ask(
                actor,
                message,
                timeout,
                actorSystem.scheduler()
            )
        )
    }


}

AKKA가 Scala/JAVA만 기본으로 지원하고 Kotlin은 알아서 호환해야하다보니, 자바코드에서 오는 괴리감을 이 유틸 클래스가 처리해주는것으로

액터는 기본으로 기본언어가 가진 동시성 처리 스펙을 활용함으로, 액터이용에 앞서 기본언어가 지원하는 비동기 프로그래밍 방식을 먼저 학습할 필요가 있습니다.

Future and Promise

Content

Space Tools

Versions Compared

Old Version 5

New Version 6

Key