Spring Boot with Kotlin Coroutines and RSocket

한줄 요약

Coroutine을 봐서 반가움과 동시에, RSocket이라는 새로운 개념을 가볍게 만나보기.

Part 1: Initial setup and introduction to the project

git clone을 통해서 코드를 단순히 받아주면 된다.
다만 하나 수정해줘야 할 게 있다면 build.gradle 파일에 들어있는 javafaker가 더이상 지원을 하지 않는 것 같아서, datafaker를 대신 사용해줘야 한다.
https://stackoverflow.com/questions/65775918/snakeyaml-1-27-android-jar-not-found-error-while-running-tests

기본 제공된 프로젝트는 위와 같은 구조를 하고 있다는 것을 한번 생각하고 넘어가면 될 듯 하고, 직접 수정해나가기 전에 코드를 한번씩은 들여다 보는거 추천한다.

Part 2: Adding persistence and integration tests

이 파트에선 JDBC와 H2로 real DB를 만든다.

Create database schema and configuration

sql 코드의 경우 그대로 따라하면 되는데, application.properties에서 DB 셋팅을 해주는데 가이드에 있는 코드가 잘 동작하지 않았다. 만약 잘 되면 상관이 없다. 그러나 잘 되지 않는다면 나는 아래와 같이 문제를 해결하였다.
1. schema.sql을 resource 폴더에 단독으로 둔다.(sql 폴더를 만들지 않고)
2. application.properties를 다음과 같이 작성한다.

spring.datasource.url=jdbc:h2:file:./build/data/testdb
spring.datasource.driverClassName=org.h2.Driver
spring.datasource.username=sa
spring.datasource.password=password
spring.sql.init.mode=always

working with data

그 후에는 MessageRepository를 만들고 이를 이용해 PersistentMessageService를 만들어 주면 된다. MessageService의 실제 구현채라고 생각하면 된다.

Adding integration tests

테스트의 경우 코드를 그대로 따라가주면 된다.
알아두거나 궁금증이 생긴 부분들만 체크해둔다.

@Autowired 사용

테스트 코드에서 lateinit를 통해 TestRestTemplate, MessageRepository를 받아온다. 여기서 파라미터가 아닌 lateinit을 해서 @Autowired를 붙히는 이유를 확실히 모르겠다. 정확히는 뭐가 맞는지를 모르겠는 상태다. 추가적으로 찾아봐도 Dependency injection에서 lateinit을 사용했때와 constructor로 했을 때의 큰 차이를 확실히 느끼진 못하겠다. 추후 공부 과제로 남겨둔다.

@BeforeEach, @AfterEach

각 테스트의 직전과 직후에 실행시키는 코드에 붙히는 annotation으로 여기서는 미리 data를 지정해두기 위해서 사용한다.

이후 테스트 코드

이후 테스트 코드는 직접 읽고 적어보면 충분히 이해가 되는 정도이다.

이제 part 3으로 넘어간다.

Part 3: Implementing extensions

Extension을 이용해서 코드를 간결하게 깔끔하게 만드는 방식이다. Kotlin의 Extension을 적극 활용하는 방법을 알려준다. asDomainObject, asViewModel 처럼 이른바 mapper extension을 만들어주는 것으로 중복되는 mapping 코드를 간결화 시킬 수 있다.

또 이 Part 3에서는 Markdown을 적용시키는 예시가 나오는데, 추후에 마크다운을 이용할 필요가 있을 때 한번 정도 체크하기에 좋을 것 같다.

특별한 내용은 없어 간단히 넘어간다.

Part 4: Refactoring to Spring WebFlux with Kotlin Coroutines

coroutine 적용을 위해 기존 코드를 리팩토링 한다. 여러 내용들이 나오는 것을 기대했으나 말 그대로 Coroutine을 이용하여 reactive하게 만드는 기초중에 기초라고 보인다.

우선 coroutine을 추가하여 모두 suspend 함수로 바꾸고, standard blocking Web MVC를 reactive and non-blocking한 WebFlux로 바꾸고, JDBC 또한 R2DBC로 바꿔준다.

그리고 R2DBC initialize를 위한 코드를 추가해준다.

## application.kt
~... rlwhszhem dlgn
@Configuration
class Config {

    @Bean
    fun initializer(connectionFactory: ConnectionFactory): ConnectionFactoryInitializer {
        val initializer = ConnectionFactoryInitializer()
        initializer.setConnectionFactory(connectionFactory)
        val populator = CompositeDatabasePopulator()
        populator.addPopulators(ResourceDatabasePopulator(ClassPathResource("./sql/schema.sql")))
        initializer.setDatabasePopulator(populator)
        return initializer
    }
}

## application.properties (기존 코드 제거)
spring.r2dbc.url=r2dbc:h2:file:///./build/data/testdb;USER=sa;PASSWORD=password

이 파트에서도 특별한 내용까지는 없다. 기억해갈것은 CoroutineCrudRepository가 따로 만들어져 있으며, 이를 이용하면 된다는 것!! 정도가 될 것 같다.

Part 5: Streaming with RSocket

RSocket을 이용해 메시지를 streaming-like 하게 바꿔본다.

RSocket이란

Netflix에서 만든 프로토콜의 한 종류라고 하며, 비교대상을 찾자면 gRPC가 있다. 간단히 알아보니 gRPC는 구글에서 만든 HTTP/2 기반의 RPC(지난 튜토리얼에서 접했었다.) 프레임워크이며, 일반적으로 우리가 생각하는 HTTP 통신, REST api 등과 모두 관련되어 있다고 생각하면 된다. RSocket은 아주 다른 개념이며 stream 형태라는 여러 말들이 있다. 우선은 튜토리얼의 경우 얕게 진행해봤다~ 이런 느낌이구나~ 를 알아가는게 주 목적이기에 이러한 내용은 추후에 추가적으로 알아보도록 하자. 다만, 지금 짚고 넘어가고 싶다면 이 글을 추천한다.

Code

Rsocket을 적용하며 많은 부분들이 바뀌지만 핵심이 되는 파트를 꼽아보았다. 바로 MessageResource.kt 이다.

## 기존 코드
@RestController
@RequestMapping("/api/v1/messages")
class MessageResource(val messageService: MessageService) {

    @GetMapping
    suspend fun latest(@RequestParam(value = "lastMessageId", defaultValue = "") lastMessageId: String): ResponseEntity<List<MessageVM>> {
        val messages = if (lastMessageId.isNotEmpty()) {
            messageService.after(lastMessageId)
        } else {
            messageService.latest()
        }

        return if (messages.isEmpty()) {
            with(ResponseEntity.noContent()) {
                header("lastMessageId", lastMessageId)
                build<List<MessageVM>>()
            }
        } else {
            with(ResponseEntity.ok()) {
                header("lastMessageId", messages.last().id)
                body(messages)
            }
        }
    }

    @PostMapping
    suspend fun post(@RequestBody message: MessageVM) {
        messageService.post(message)
    }
}

## 수정 후 코드
@Controller
@MessageMapping("api.v1.messages")
class MessageResource(val messageService: MessageService) {

    @MessageMapping("stream")
    suspend fun receive(@Payload inboundMessages: Flow<MessageVM>) =
            messageService.post(inboundMessages)

    @MessageMapping("stream")
    fun send(): Flow<MessageVM> = messageService
            .stream()
            .onStart {
                emitAll(messageService.latest())
            }
}

얼핏 봤을 땐 코드가 매우 간결해졌는데, 새로운 annotation 들이 많이 보인다. 여기서 중요한 사실은 MessageMapping annotation에 두 함수 "모두" "stream"을 쓰고 있다는 것이다. 앞서서 살펴본 RSocket이 4가지 비동기 메시징 모델이 있는데 그 중에서 "Request-Stream" or "Channel" 형식을 보인다. 물론 이 두개에 대해서도 아직 완벽한 이해는 없는 상태이다.

나머지 코드들은 따라 써가면서 확인해보는 걸 추천한다.

마치며

오랜만에 Coroutine을 보게되서 기쁨과 동시에 Coroutine, Flow에 대한 deep한 공부도 한번 필요할 것 같다는 생각이 들었다. 또, RSocket이라는 새로운 개념을 처음 접하게 되었는데, 이 또한 지금 감으로만 알 것 같은 상태라 추가적인 공부가 필요하다.
이렇게 새롭고 복잡해보이는걸 마주할 때 마다 하는 생각이 있다. 머리를 비우고, 우선 가볍게 따라만 가 보는거다. 이런게 있다 정도를 알아두는 것 만으로도 실제 필요한 지점이 왔을 때 큰 도움이 된다는 것을 잊지말자.