spring r2dbc + coroutines 질문

Question

안녕하세요coroutines + spring webflux + r2dbc를 사용하고 있는데 asFlow 관련에서 질문드리고 싶은 것이 있습니다.@Repository class MemberRepositoryCustomImpl( private val client: DatabaseClient ) : MemberRepositoryCustom { override suspend fun findAllWithTeam(): Flux { val query = """ SELECT m.*, t.name AS team_name, t.leader_id AS team_leader_id, t.created_at AS team_created_at, t.modified_at AS team_modified_at FROM member m LEFT JOIN Team t ON m.team_id = t.team_id """.trimIndent() return client.sql(query) .fetch() .all() .map { memberWithTeamMapper(it) } } } @Service @Transactional class MemberService( private val memberRepository: MemberRepository ) { suspend fun findMembersWithTeam(): Flow { return memberRepository.findAllWithTeam().asFlow() } }위와 같은 경우에는 문제 없이 동작하는데요.아래와 같이 repository 단에 asFlow를 붙이고 service단에서 asFlow를 제거하면 문제가 발생합니다.@Repository class MemberRepositoryCustomImpl( private val client: DatabaseClient ) : MemberRepositoryCustom { override suspend fun findAllWithTeam(): Flow { val query = """ SELECT m.*, t.name AS team_name, t.leader_id AS team_leader_id, t.created_at AS team_created_at, t.modified_at AS team_modified_at FROM member m LEFT JOIN Team t ON m.team_id = t.team_id """.trimIndent() return client.sql(query) .fetch() .all() .map { memberWithTeamMapper(it) } .asFlow() } } @Service @Transactional class MemberService( private val memberRepository: MemberRepository ) { suspend fun findMembersWithTeam(): Flow { return memberRepository.findAllWithTeam() } }아래와 같은 문제가 발생하는데요.에러 상으로는 classCastException인데 단지 위치만 바꿨을 뿐인데 이런 문제가 발생하는 이유가 궁금합니다.구글링해도 이와 같은 이슈는 찾지 못했네요.. 혹시 비슷한 경험이 있으신 분 계신가요?java.lang.ClassCastException: class reactor.core.publisher.FluxOnAssembly cannot be cast to class kotlinx.coroutines.flow.Flow (reactor.core.publisher.FluxOnAssembly and kotlinx.coroutines.flow.Flow are in unnamed module of loader 'app') at org.springframework.core.CoroutinesUtils.asFlux(CoroutinesUtils.java:137) ~[spring-core-6.0.9.jar:6.0.9] Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: Assembly trace from producer [reactor.core.publisher.MonoFlatMapMany] : reactor.core.publisher.Mono.flatMapMany(Mono.java:3107) org.springframework.core.CoroutinesUtils.invokeSuspendingFunction(CoroutinesUtils.java:115) Error has been observed at the following site(s): *_______Mono.flatMapMany ⇢ at org.springframework.core.CoroutinesUtils.invokeSuspendingFunction(CoroutinesUtils.java:115) *_________Flux.usingWhen ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.lambda$invokeWithinTransaction$3(TransactionAspectSupport.java:943) |_ Flux.onErrorMap ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.lambda$invokeWithinTransaction$3(TransactionAspectSupport.java:956) *_______Mono.flatMapMany ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.invokeWithinTransaction(TransactionAspectSupport.java:942) |_ Flux.contextWrite ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.invokeWithinTransaction(TransactionAspectSupport.java:957) |_ Flux.contextWrite ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.invokeWithinTransaction(TransactionAspectSupport.java:958) *_______Mono.flatMapMany ⇢ at org.springframework.core.CoroutinesUtils.invokeSuspendingFunction(CoroutinesUtils.java:115) |_ Flux.map ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.lambda$encode$7(AbstractJackson2Encoder.java:188) |_ Flux.switchIfEmpty ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.lambda$encode$7(AbstractJackson2Encoder.java:200) *________Flux.concatWith ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.lambda$encode$7(AbstractJackson2Encoder.java:201) |_ Flux.doOnNext ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.lambda$encode$7(AbstractJackson2Encoder.java:205) |_ Flux.doAfterTerminate ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.lambda$encode$7(AbstractJackson2Encoder.java:206) *___Flux.deferContextual ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.encode(AbstractJackson2Encoder.java:151) |_ Mono.doOnError ⇢ at org.springframework.http.server.reactive.AbstractServerHttpResponse.writeWith(AbstractServerHttpResponse.java:210) |_ checkpoint ⇢ Handler com.example.server.controller.MemberController#findMembersWithTeam(Continuation) [DispatcherHandler] |_ Mono.onErrorResume ⇢ at org.springframework.web.reactive.DispatcherHandler.handleResult(DispatcherHandler.java:194) *___________Mono.flatMap ⇢ at org.springframework.web.reactive.DispatcherHandler.lambda$handleResult$8(DispatcherHandler.java:195) *___________Mono.flatMap ⇢ at org.springframework.web.reactive.DispatcherHandler.handleRequestWith(DispatcherHandler.java:184) *___________Mono.flatMap ⇢ at org.springframework.web.reactive.DispatcherHandler.handle(DispatcherHandler.java:154) *_____________Mono.defer ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.filter(DefaultWebFilterChain.java:106) |_ Mono.onErrorResume ⇢ at org.springframework.web.server.handler.ExceptionHandlingWebHandler.handle(ExceptionHandlingWebHandler.java:77) *_____________Mono.error ⇢ at org.springframework.web.server.handler.ExceptionHandlingWebHandler$CheckpointInsertingHandler.handle(ExceptionHandlingWebHandler.java:98) |_ checkpoint ⇢ HTTP GET "/member/with-team" [ExceptionHandlingWebHandler]

Clickin · Answer

리포지토리의 반환 타입이 여전히 Flux니까 리포지토리단에서 Flow 타입으로 변환된걸 반환시에 Flux 타입으로 캐스팅하려다 문제가 생기겠죠

KS J · Answer

Flow는 suspend 개념에 포함되지 않기 때문에 suspend 키워드를 제거하시면 정상적으로 실행됩니다

spring r2dbc + coroutines 질문

2개의 답변

#r2dbc 목록