揭秘 WebFlux 中如何修改 request/response body

#揭秘 WebFlux 中如何修改 request/response body| 来源: 网络整理| 查看: 265

完整的代码请参考 https://github.com/javahongxi/webflux-sample

我们的一些企业对于HTTP服务有一些非正常的做法，它们客户端的请求body是加密的，即在服务端需要对请求body进行解密，而服务端响应的body也要求加密。本文就来揭秘这一需求在 WebFlux 中如何实现，我们给 request/response body 均增加一个表示时间戳的字段 start/end 来模拟请求数据解密和响应数据加密，思路如下。

首先我们需要知道，WebFlux 的过滤器/拦截器是统一用 WebFilter 来表示的，与 Spring MVC 类似，对于 application/json 请求，WebFlux 读取 body inputstream 也只能读取一次，对于query params / form-data / form-x-www 请求，可以反复获取。所以，怎么修改 application/json 请求的 request/response body 是一个难点。网上找遍了例子，大都是错误的例子或是hello demo，完全没法解决本文的需求。经过苦心探索，本人终于找到了解决方案，代码如下。

import lombok.extern.slf4j.Slf4j; import org.hongxi.sample.webflux.support.*; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.annotation.Order; import org.springframework.http.MediaType; import org.springframework.http.codec.ServerCodecConfigurer; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.stereotype.Component; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebFilter; import org.springframework.web.server.WebFilterChain; import reactor.core.publisher.Mono; import java.util.Map; /** * Created by shenhongxi on 2021/4/29. */ @Slf4j @Order(-1) @Component public class ModifyBodyFilter implements WebFilter { @Autowired private ServerCodecConfigurer codecConfigurer; @Override public Mono filter(ServerWebExchange exchange, WebFilterChain chain) { if (WebUtils.shouldNotFilter(exchange)) { return chain.filter(exchange); } return ParamUtils.from(exchange) .map(params -> decorate(exchange, params)) .flatMap(chain::filter); } private ServerWebExchange decorate(ServerWebExchange exchange, Map params) { if (params.isEmpty()) { return exchange; } ServerHttpResponse serverHttpResponse = new ModifiedServerHttpResponse(exchange, codecConfigurer.getReaders()); MediaType contentType = exchange.getRequest().getHeaders().getContentType(); if (contentType != null && contentType.isCompatibleWith(MediaType.APPLICATION_JSON)) { Map decrypted = Crypto.decrypt(params); exchange.getAttributes().put(WebUtils.REQUEST_PARAMS_ATTR, decrypted); byte[] rawBody = JacksonUtils.serialize(decrypted); ServerHttpRequest serverHttpRequest = new ModifiedServerHttpRequest(exchange.getRequest(), rawBody); return exchange.mutate().request(serverHttpRequest).response(serverHttpResponse).build(); } ServerWebExchange serverWebExchange = new ModifiedServerWebExchange(exchange); return serverWebExchange.mutate().response(serverHttpResponse).build(); } }

import org.springframework.core.ResolvableType; import org.springframework.http.MediaType; import org.springframework.http.codec.DecoderHttpMessageReader; import org.springframework.http.codec.HttpMessageReader; import org.springframework.http.codec.json.Jackson2JsonDecoder; import org.springframework.web.bind.support.WebExchangeDataBinder; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; import java.util.Collections; import java.util.Map; /** * Created by shenhongxi on 2021/4/29. */ public abstract class ParamUtils { @SuppressWarnings("rawtypes") private static final HttpMessageReader messageReader = new DecoderHttpMessageReader(new Jackson2JsonDecoder()); @SuppressWarnings("unchecked") public static Mono from(ServerWebExchange exchange) { Map data = exchange.getAttribute(WebUtils.REQUEST_PARAMS_ATTR); if (data != null) { return Mono.just(data); } Mono params; MediaType contentType = exchange.getRequest().getHeaders().getContentType(); if (contentType != null && contentType.isCompatibleWith(MediaType.APPLICATION_JSON)) { params = (Mono) messageReader.readMono( ResolvableType.forType(Map.class), exchange.getRequest(), Collections.emptyMap()); } else { params = WebExchangeDataBinder.extractValuesToBind(exchange); } return params.doOnNext(e -> exchange.getAttributes().put(WebUtils.REQUEST_PARAMS_ATTR, e)); } }

import io.netty.buffer.ByteBufAllocator; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.http.HttpHeaders; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpRequestDecorator; import reactor.core.publisher.Flux; import java.nio.charset.StandardCharsets; /** * Created by shenhongxi on 2021/4/29. */ public class ModifiedServerHttpRequest extends ServerHttpRequestDecorator { private final byte[] rawBody; public ModifiedServerHttpRequest(ServerHttpRequest delegate, byte[] rawBody) { super(delegate); this.rawBody = rawBody; } @Override public Flux getBody() { NettyDataBufferFactory nettyDataBufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT); DataBuffer buffer = nettyDataBufferFactory.allocateBuffer(this.rawBody.length); buffer.write(this.rawBody); return Flux.just(buffer); } @Override public HttpHeaders getHeaders() { // 必须 new，不能直接操作 super.getHeaders()（readonly） HttpHeaders headers = new HttpHeaders(); headers.addAll(super.getHeaders()); headers.setContentLength(this.rawBody.length); return headers; } /** * @return body json string */ public String bodyString() { return new String(rawBody, StandardCharsets.UTF_8); } }

import org.reactivestreams.Publisher; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferUtils; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.ReactiveHttpOutputMessage; import org.springframework.http.codec.HttpMessageReader; import org.springframework.http.server.reactive.ServerHttpResponseDecorator; import org.springframework.web.reactive.function.BodyInserter; import org.springframework.web.reactive.function.BodyInserters; import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.util.List; /** * Created by shenhongxi on 2021/4/29. */ public class ModifiedServerHttpResponse extends ServerHttpResponseDecorator { private final ServerWebExchange exchange; private final List> messageWriters() { return exchangeStrategies.messageWriters(); } @Override public Optional serverRequest() { return Optional.empty(); } @Override public Map hints() { return Collections.emptyMap(); } }

import java.util.function.Supplier; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferFactory; import org.springframework.http.HttpHeaders; import org.springframework.http.ReactiveHttpOutputMessage; import org.springframework.web.server.ServerWebExchange; /** * Created by shenhongxi on 2021/5/3. */ public class CachedBodyOutputMessage implements ReactiveHttpOutputMessage { private final DataBufferFactory bufferFactory; private final HttpHeaders httpHeaders; private boolean cached = false; private Flux body = Flux .error(new IllegalStateException("The body is not set. " + "Did handling complete with success?")); public CachedBodyOutputMessage(ServerWebExchange exchange, HttpHeaders httpHeaders) { this.bufferFactory = exchange.getResponse().bufferFactory(); this.httpHeaders = httpHeaders; } @Override public void beforeCommit(Supplier

【本文地址】

公司简介

联系我们