Spring Cloud Gateway：后过滤Web客户端请求

Question

我们使用Spring Cloud Gateway将请求路由到多个底层服务。对这些底层服务的调用将是连续的，并可能相互传递（来自一个服务的响应在下一个服务的请求中使用）。当我们需要在主请求之前顺序地发出这些请求时，我们有一个可行的解决方案，但在主请求之后，我们在将一个代理请求的响应输入到下一个代理请求的请求中时遇到了问题。

我们计划将响应从一个请求发送到下一个请求的方式是使用GatewayFilter中的WebClient发出请求，并将响应字符串存储在Exchange的属性存储区中。然后，在下一个代理请求期间，我们提供一个属性名，可选地从中提取请求正文。这在使用“预”过滤器时很好，因为第一个代理请求在第二个请求构建和执行之前就已经构建、执行和缓存了响应，所以属性链按预期工作。当使用“POST”过滤器时，问题就来了。在post代理中，web客户端请求都是在后续请求完成之前构建的。因此，属性存储区从来没有来自上一个请求的响应，这意味着下一个请求不能按预期工作，因为它没有有效的请求体。

我的理解是调用chain.filter（exchange）.then（mono.fromrunnable{...}）将导致.then逻辑仅在前面的筛选器完全完成后才执行。事实似乎并非如此。在其他筛选器类型中，如日志记录、响应操作等，post筛选器以正确的顺序执行，但在创建WebClient时，它们似乎不是这样。

有人对如何实现这种期望的行为有什么想法吗？

class PreProxyGatewayFilterFactory: AbstractGatewayFilterFactory<PreProxyGatewayFilterFactory.Params>(Params::class.java) {
    override fun apply(params: Params): GatewayFilter {
        return OrderedGatewayFilter(
            { exchange, chain ->
                ServerWebExchangeUtils.cacheRequestBody(exchange){
                    val cachedExchange = exchange.mutate().request(it).build()
                    executeRequest(cachedExchange, params)
                        .map { response ->
                            val body = response.body.toString()
                            cacheResponse(
                                response.body.toString(),
                                params.cachedResponseBodyAttributeName,
                                cachedExchange
                            )
                        }
                        .flatMap(chain::filter)
                }
            }, params.order)
    }

    private fun cacheResponse(response: String, attributeName: String?, exchange: ServerWebExchange): ServerWebExchange{
        if(!attributeName.isNullOrBlank()){
            exchange.attributes[attributeName] = response
        }
        return exchange
    }

    private fun executeRequest(exchange: ServerWebExchange, params: Params): Mono<ResponseEntity<String>>{
        val request = when(exchange.request.method){
            HttpMethod.PUT -> WebClient.create().put().uri(params.proxyPath).body(createProxyRequestBody(exchange, params.cachedRequestBodyAttributeName))
            HttpMethod.POST -> WebClient.create().post().uri(params.proxyPath).body(createProxyRequestBody(exchange, params.cachedRequestBodyAttributeName))
            HttpMethod.GET -> WebClient.create().get().uri(params.proxyPath)
            HttpMethod.DELETE -> WebClient.create().delete().uri(params.proxyPath)
            else -> throw Exception("Invalid request method passed in to the proxy filter")
        }
        return request.headers { headers ->
            headers.addAll(exchange.request.headers)
            headers.remove(CONTENT_LENGTH)
        }
        .exchange()
        .flatMap{ response ->
            response.toEntity(String::class.java)
        }
    }

    private fun createProxyRequestBody(exchange: ServerWebExchange, attributeName: String?): BodyInserter<out Flux<out Any>, ReactiveHttpOutputMessage> {
        val cachedBody = attributeName?.let { attrName ->
            exchange.getAttributeOrDefault<String>(attrName, "null")
        } ?: "null"
        return if(cachedBody != "null"){
            BodyInserters.fromPublisher(Flux.just(cachedBody), String::class.java)
        } else {
            BodyInserters.fromDataBuffers(exchange.request.body)
        }
    }

    data class Params(
            val proxyPath: String = "",
            val cachedRequestBodyAttributeName: String? = null,
            val cachedResponseBodyAttributeName: String? = null,
            val order: Int = 0
    )
}

class PostProxyGatewayFilterFactory: AbstractGatewayFilterFactory<PostProxyGatewayFilterFactory.Params>(Params::class.java) {
    override fun apply(params: Params): GatewayFilter {
        return OrderedGatewayFilter(
            { exchange, chain ->
                ServerWebExchangeUtils.cacheRequestBody(exchange){
                    val cachedExchange = exchange.mutate().request(it).build()

                    //Currently using a cached body does not work in post proxy
                    chain.filter(cachedExchange).then( Mono.fromRunnable{
                        executeRequest(cachedExchange, params)
                            .map { response ->
                                cacheResponse(
                                    response.body.toString(),
                                    params.cachedResponseBodyAttributeName,
                                    cachedExchange
                                )
                            }
                            .flatMap {
                                Mono.empty<Void>()
                            }
                    })
                }
            }, params.order)
    }

    private fun cacheResponse(response: String, attributeName: String?, exchange: ServerWebExchange): ServerWebExchange{
        if(!attributeName.isNullOrBlank()){
            exchange.attributes[attributeName] = response
        }
        return exchange
    }

    private fun executeRequest(exchange: ServerWebExchange, params: Params): Mono<ResponseEntity<String>>{
        val request = when(exchange.request.method){
            HttpMethod.PUT -> WebClient.create().put().uri(params.proxyPath).body(createProxyRequestBody(exchange, params.cachedRequestBodyAttributeName))
            HttpMethod.POST -> WebClient.create().post().uri(params.proxyPath).body(createProxyRequestBody(exchange, params.cachedRequestBodyAttributeName))
            HttpMethod.GET -> WebClient.create().get().uri(params.proxyPath)
            HttpMethod.DELETE -> WebClient.create().delete().uri(params.proxyPath)
            else -> throw Exception("Invalid request method passed in to the proxy filter")
        }
        return request.headers { headers ->
            headers.addAll(exchange.request.headers)
            headers.remove(CONTENT_LENGTH)
        }
        .exchange()
        .flatMap{ response ->
            response.toEntity(String::class.java)
        }
    }

    private fun createProxyRequestBody(exchange: ServerWebExchange, attributeName: String?): BodyInserter<out Flux<out Any>, ReactiveHttpOutputMessage> {
        val cachedBody = attributeName?.let { attrName ->
            exchange.getAttributeOrDefault<String>(attrName, "null")
        } ?: "null"
        return if(cachedBody != "null"){
            BodyInserters.fromPublisher(Flux.just(cachedBody), String::class.java)
        } else {
            BodyInserters.fromDataBuffers(exchange.request.body)
        }
    }


    data class Params(
            val proxyPath: String = "",
            val cachedRequestBodyAttributeName: String? = null,
            val cachedResponseBodyAttributeName: String? = null,
            val order: Int = 0
    )
}

Answer 1

最终能够为post filter代理找到一个有效的解决方案，从属性中提取它的请求体。这是一个相对简单的修复，我只是找不到答案。我只需要使用chain.filter（exchange）.然后（mono.fromrunnable{...execute proxy request...}），而不是使用chain.filter（exchange）.然后（mono.defer{...execute proxy request...}）。