Interceptor에서 reactor Context 유지하는 방... - 인프런

Kevin의 알기 쉬운 Spring Reactive Web Applications: Reactor 1부

Context의 특징

Interceptor에서 reactor Context 유지하는 방법

24.07.15 09:47 작성

수정됨

외부에서 들어온 요청 헤더 정보를 Reactor Context에 저장하고, 내부적으로 grpc호출 할때 clientInterceptor에서 저장된 Context를 읽어오고 싶은데, Context0과 함께 읽지 못하는 문제가 있습니다.

Interceptor에서도 읽어오려면 어떻게 해야할까요 ㅜㅜ

외부 요청 헤더를 Context에 저장하는 코드

@Component
class RequestHeaderFilter() : WebFilter {
    companion object {
        const val X_REQUEST_ID_KEY = "X-Request-Id"
    }
    override fun filter(exchange: ServerWebExchange, chain: WebFilterChain): Mono<Void> {
        val requestId = exchange.request.headers.getFirst(HeaderNames.REQUEST_ID)
        return chain.filter(exchange)
            .contextWrite { context ->
                context.put(X_REQUEST_ID_KEY, requestId ?: "test!!")
            }
            .doFinally {
            }
            .subscribeOn(Schedulers.boundedElastic())
    }
}

grpc 요청 interceptor에서 reator Context 조회하여 grpc Metadata에 추가하는 코드

class CustomHeaderInterceptor() : ClientInterceptor {
    companion object  {
        const val X_REQUEST_ID_KEY = "X-Request-Id"
        val X_REQUEST_ID_HEADER_KEY: Metadata.Key<String> = Metadata.Key.of("x-request-id", Metadata.ASCII_STRING_MARSHALLER)
    }

    override fun <ReqT : Any, RespT : Any> interceptCall(
        method: MethodDescriptor<ReqT, RespT>,
        callOptions: CallOptions,
        next: Channel
    ): ClientCall<ReqT, RespT> {
        return object : ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
            override fun start(responseListener: Listener<RespT>, headers: Metadata) {
                Mono.deferContextual { context ->
                    val requestId: String = context.getOrDefault(X_REQUEST_ID_KEY, fallbackRequestId()) ?: fallbackRequestId()
                    headers.put(X_REQUEST_ID_HEADER_KEY, requestId)
                    delegate().start(responseListener, headers)
                    Mono.just(requestId)
                }.subscribe()
            }
        }
    }

    private fun fallbackRequestId() = "${UidUtils.generateUid()}"

java spring project-reactor webflux

답변 2

Kevin

지식공유자

2024. 07. 15. 12:46

안녕하세요. 제가 해외에 있어서 답변이 조금 늦어져서 죄송합니다.

gRPC는 개념은 알고 있지만 사용한 적은 없다는 점을 먼저 말씀 드리고 답변을 드려보면,

class CustomHeaderInterceptor() : ClientInterceptor {
    companion object  {
        const val X_REQUEST_ID_KEY = "X-Request-Id"
        val X_REQUEST_ID_HEADER_KEY: Metadata.Key<String> = Metadata.Key.of("x-request-id", Metadata.ASCII_STRING_MARSHALLER)
    }

    override fun <ReqT : Any, RespT : Any> interceptCall(
        method: MethodDescriptor<ReqT, RespT>,
        callOptions: CallOptions,
        next: Channel
    ): ClientCall<ReqT, RespT> {
        return object : ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
            override fun start(responseListener: Listener<RespT>, headers: Metadata) {
                Mono.deferContextual { context ->
                    val requestId: String = context.getOrDefault(X_REQUEST_ID_KEY, fallbackRequestId()) ?: fallbackRequestId()
                    headers.put(X_REQUEST_ID_HEADER_KEY, requestId)
                    delegate().start(responseListener, headers) // (1) gRPC 호출 시작 부분?
                    Mono.just(requestId)
                }.subscribe()
            }
        }
    }

(1)의 gRPC 호출을 시작하는 코드가 Mono.deferContextual(...) 내부가 아니라 subscribe(..) 내부에서 동작해야 되지 않을까 하는 생각이 드네요.

Mono.just(requestId) 대신에 Mono.just(headers)로 리턴하고, subscribe() 내부에서 gRPC로 headers를 전달하는게 자연스러워 보여서요.

그렇게 생각한 이유는 Reactor의 Sequence가 비동기로 동작하기 때문에 Mono 내부에서 gRPC가 호출되지 않을 거라고 생각했기 때문입니다.

gRPC 호출을 Reactor Sequece 동작이 끝난 후에 일어나도록 하면 정상적으로 동작할 것 같다라는 생각이 듭니다.

오류 해결 되시길 바래볼게요.

감사합니다.

인프런 AI 인턴

2024. 07. 15. 09:51

안녕하세요, 인프런 AI 인턴입니다.

yj na님, Spring WebFlux 환경에서 Interceptor를 통해 Reactor Context 정보를 유지하는 방법에 관해 문의하셨군요. 이와 관련하여 Kevin 강사님의 강의 내에서 비슷한 주제로 질의응답이 이루어진 사례를 찾았습니다. 강의는 Reactive Programming과 Reactor Context를 다루고 있으며, 특히 Reactor Context의 저장 및 조회 방법에 초점을 맞춘 답변이 유용할 수 있을 것 같습니다.

`.contextWrite의 실행시점`에서 Reactor Context에 값을 저장하고, 이후에 해당 Context에서 값을 읽어오는 과정에 대한 깊은 이해를 돕는 답변을 확인하실 수 있습니다. 이 답변에서는 .contextWrite() 메서드가 코드의 순서와 관계없이 먼저 실행되어 Context에 값을 저장하는 방식과 이후에 해당 값을 Mono.deferContextual을 통해 조회하는 예제가 제공됩니다. gRPC 호출 시 Context를 관리하는 방법에 대한 유사한 과제를 해결하는 데 도움이 될 것입니다.

Reactor Context를 사용하여 특정 데이터(예: 요청 헤더 정보)를 저장하고, 다른 부분에서 이를 읽어오는 작업은 리액티브 프로그래밍에서 중요한 패턴 중 하나입니다. 이 과정에서 Context의 데이터가 전파되는 방식을 이해하는 것이 중요합니다.

직접적인 코드 구현에 대한 구체적인 해결책을 제시하는 답변은 아니지만, Context의 작동 원리와 관련하여 유사한 질문에 대한 설명은 문제 해결 과정에서 도움이 될 수 있습니다.

저는 질문자님의 이해를 돕고자 지식공유자님의 기존 답변을 참고하여 유사한 답변 링크를 추천해드리고 있습니다.
현재 베타 기능으로 답변이 만족스럽지 않을 수 있는 점 양해 부탁드립니다. 🙏
추가적으로 궁금한 점이 있으시면, 이어서 질문해 주세요. 곧 지식공유자께서 답변해 주실 것입니다.

인프런 커뮤니티 질문&답변

Interceptor에서 reactor Context 유지하는 방법