Kotlin + Spring WebFlux Functional Endpoint에서 ServerSentEvent 사용법

1. 서론

Spring WebFlux를 사용하더라도 Annotation 기반의 Request Mapping을 사용하는 사례가 많다.
기능이 문제되는 점은 없고 사실 IDE의 지원도 Annotation 기반 형식이 더 좋다.
아직도 Intellij에서 Functional Endpoint를 인식해주지 못한다!

하지만 Functional Endpoint를 사용하고 싶을 때는 구글링해도 예제가 많이 없다.
실제 코드는 간단하나 관련 문서를 찾기 힘든 관계로 메모를 겸해서 구성을 포스팅한다.

2. 구성

주요 기술 스택은 다음과 같다

• Spring WebFlux
• Kotlin
• Kotlin coroutine

Spring Initializer에서 Reactive Web으로 구성한다.
※ Spring boot 3.x 버전은 JDK 17 이상을 요구하기에 주의한다

build.gradle.kts는 다음과 같다.

import org.jetbrains.kotlin.gradle.tasks.KotlinCompile

plugins {
    id("org.springframework.boot") version "3.0.4"
    id("io.spring.dependency-management") version "1.1.0"
    kotlin("jvm") version "1.7.22"
    kotlin("plugin.spring") version "1.7.22"
}

group = "com.example"
version = "0.0.1-SNAPSHOT"
java.sourceCompatibility = JavaVersion.VERSION_17

configurations {
    compileOnly {
        extendsFrom(configurations.annotationProcessor.get())
    }
}

repositories {
    mavenCentral()
}

dependencies {
    implementation("org.springframework.boot:spring-boot-starter-webflux")
    implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
    implementation("io.projectreactor.kotlin:reactor-kotlin-extensions")
    implementation("org.jetbrains.kotlin:kotlin-reflect")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
    implementation("org.springframework.boot:spring-boot-starter-actuator")
    developmentOnly("org.springframework.boot:spring-boot-devtools")
    testImplementation("org.springframework.boot:spring-boot-starter-test")
    testImplementation("io.projectreactor:reactor-test")
}

tasks.withType<KotlinCompile> {
    kotlinOptions {
        freeCompilerArgs = listOf("-Xjsr305=strict")
        jvmTarget = "17"
    }
}

tasks.withType<Test> {
    useJUnitPlatform()
}

Random Int를 돌려주는 Service, Handler, Router를 작성해본다.

1. Service
   코루틴을 사용하기 위해 suspend 키워드가 필요하다.
   Flow는 내부적으로 suspend 키워드를 호출하기 때문에 Flow를 리턴하는 함수는 suspend 키워드를 사용하지 않는다

package com.example.webfluxsse

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.runBlocking
import org.springframework.stereotype.Service
import kotlin.random.Random

@Service
class RandomService {

    suspend fun generateRandomList(flowSize: Int): List<Int> {
        return (0 until flowSize).map { generateRandom(0, 100) }
    }

    fun generateRandomFlow(flowSize: Int): Flow<Int> = runBlocking {
        generateRandomList(flowSize).asFlow()
    }

    private fun generateRandom(from: Int, to: Int): Int {
        return Random.nextInt(from, to)
    }
}

2. Handler
   Service를 주입받아 ServerSentEvent를 발생시키는 Handler이다.
   Flow를 리턴 시에는 bodyAndAwait, suspend된 일반 리턴값을 사용하기 위해서는 bodyValueAndAwait을 사용한다.

package com.example.webfluxsse

import org.springframework.http.codec.ServerSentEvent
import org.springframework.stereotype.Controller
import org.springframework.web.reactive.function.server.*

@Controller
class RandomHandler(
    private val randomService: RandomService
) {

    suspend fun getRandomInt(serverRequest: ServerRequest) : ServerResponse {
        val data = randomService.generateRandomList(10)
        //List<Int>의 ServerSentEvent
        return ServerResponse.ok().sse().bodyValueAndAwait(ServerSentEvent.builder(data).build())
    }

    suspend fun getRandomIntFlow(serverRequest: ServerRequest) : ServerResponse {
    	//Flow<Int>가 순차적으로 발생된다
        return ServerResponse.ok().sse().bodyAndAwait(randomService.generateRandomFlow(10))
    }
}

3. Router
   coRouter 는 coroutine을 사용 가능하게 해주는 router DSL이다.
   일반적인 roter DSL은 Java Reactor에서 사용하는 Mono 및 Flux를 입출력으로 사용한다.

package com.example.webfluxsse

import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.web.reactive.function.server.coRouter


@Configuration
class RandomRouter {
    @Bean
    fun randomEndpoint(randomHandler: RandomHandler) = 
    // suspend fun을 Handler로 등록하기 위해서는 coRouter가 필요
    coRouter {
        GET("/randomInt", randomHandler::getRandomInt)
        GET("/randomIntFlow", randomHandler::getRandomIntFlow)
    }

}

3. 결론

WebSocket이 대부분의 상황에서 유효하지만 클라이언트의 상호작용이 적고 서버에서 읽기만 하는 경우 굳이 어렵게 WebSocket으로 구현할 필요가 없다고 본다.

채팅처럼 실시간으로 단건 push가 필요하다면 Flow, 전체 작업 목록을 실시간으로 조회 시 List를 리턴하면 된다.

코드 링크: https://github.com/Clickin/webflux-sse-example