共计 8133 个字符，预计需要花费 21 分钟才能阅读完成。
Spring Cloud Gateway 在有些场景中须要获取 request body 内容进行参数校验或参数批改，咱们通过在 GatewayFilter 中获取申请内容来获取和批改申请体，上面咱们就基于 ServerWebExchange 来实现：
ServerWebExchange 命名为服务网络交换器，寄存着重要的申请 - 响应属性、申请实例和响应实例等等，有点像 Context 的角色，其中有两个重要的接口办法：
// 获取 ServerHttpRequest 对象
ServerHttpRequest getRequest();
// 获取 ServerHttpResponse 对象
ServerHttpResponse getResponse();创立一个 GatewayFilter，必须实现 Ordered 接口，返回一个小于 - 1 的 order 值，这是因为 NettyWriteResponseFilter 的 order 值为 -1，咱们须要笼罩返回响应体的逻辑，自定义的 GlobalFilter 必须比 NettyWriteResponseFilter 优先执行。
public class RequestGatewayFilter implements GatewayFilter, Ordered {
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {ServerHttpRequest request = exchange.getRequest();
HttpHeaders headers = request.getHeaders();
// 解决参数
MediaType contentType = headers.getContentType();
if (exchange.getRequest().getMethod().equals(HttpMethod.POST)) {
//Content-type 为“application/json”if (MediaType.APPLICATION_JSON.isCompatibleWith(contentType)) {return readBody(exchange, chain);
}
//Content-type 为“application/x-www-form-urlencoded”else if(MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(contentType)){GatewayContext gatewayContext = new GatewayContext();
gatewayContext.setRequestHeaders(headers);
gatewayContext.getAllRequestData().addAll(request.getQueryParams());
exchange.getAttributes().put(GatewayContext.CACHE_GATEWAY_CONTEXT,gatewayContext);
return readFormData(exchange, chain, gatewayContext);
}
//Content-type 为“multipart/form-data”else if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {GatewayContext gatewayContext = new GatewayContext();
gatewayContext.setRequestHeaders(headers);
gatewayContext.getAllRequestData().addAll(request.getQueryParams());
exchange.getAttributes().put(GatewayContext.CACHE_GATEWAY_CONTEXT,gatewayContext);
return readMultipartData(exchange, chain, gatewayContext);
}
} else {return readGetData(exchange, chain);
}
return chain.filter(exchange);
}
@Override
public int getOrder() {return -2;}
}解决 content-type 为 application/json 的办法：
/**
* ReadJsonBody
*
* @param exchange
* @param chain
* @return
*/
private Mono<Void> readBody(ServerWebExchange exchange, GatewayFilterChain chain) {
/**
* join the body
*/
return DataBufferUtils.join(exchange.getRequest().getBody()).flatMap(dataBuffer -> {byte[] bytes = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(bytes);
DataBufferUtils.release(dataBuffer);
/**
* validate request params or form data
*/
Result checkResult = null;
try {String bodyString = new String(bytes, "utf-8");
Map bodyMap = JSONObject.parseObject(bodyString,Map.class);
// 校验参数
checkResult = validParam(exchange, bodyMap);
if(checkResult.getCode()!=0){return errorInfo(exchange, checkResult.getCode(), checkResult.getMessage());
}
} catch (UnsupportedEncodingException e) {e.printStackTrace();
}
Flux<DataBuffer> cachedFlux = Flux.defer(() -> {DataBuffer buffer = exchange.getResponse().bufferFactory().wrap(bytes);
DataBufferUtils.retain(buffer);
return Mono.just(buffer);
});
/**
* repackage ServerHttpRequest
*/
ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(exchange.getRequest()) {
@Override
public Flux<DataBuffer> getBody() {return cachedFlux;}
};
ServerHttpResponse originalResponse = exchange.getResponse();
originalResponse.getHeaders().setContentType(MediaType.APPLICATION_JSON);
DataBufferFactory bufferFactory = originalResponse.bufferFactory();
ServerHttpResponseDecorator response = buildResponse(originalResponse, bufferFactory, (Map)checkResult.getResult());
/**
* mutate exchage with new ServerHttpRequest
*/
ServerWebExchange mutatedExchange = exchange.mutate().request(mutatedRequest).response(response).build();
/**
* read body string with default messageReaders
*/
return ServerRequest.create(mutatedExchange, MESSAGE_READERS).bodyToMono(String.class)
.doOnNext(objectValue -> {log.debug("[GatewayContext]Read JsonBody:{}", objectValue);
}).then(chain.filter(mutatedExchange));
});
}
解决 content-type 为 application/x-www-form-urlencoded 的办法：
/**
* ReadFormData
* @param exchange
* @param chain
* @return
*/
private Mono<Void> readFormData(ServerWebExchange exchange, GatewayFilterChain chain, GatewayContext gatewayContext){HttpHeaders headers = exchange.getRequest().getHeaders();
return exchange.getFormData().switchIfEmpty(Mono.defer(() -> Mono.just(new LinkedMultiValueMap<>()))
).flatMap(formDataMap -> {Charset charset = headers.getContentType().getCharset();
charset = charset == null? StandardCharsets.UTF_8:charset;
String charsetName = charset.name();
MultiValueMap<String, String> paramMap = exchange.getRequest().getQueryParams();
Map map = convertMap(paramMap);
/*
* formData is empty just return
*/
if((null == formDataMap || formDataMap.isEmpty()) && (null == map || map.isEmpty())){return chain.filter(exchange);
}
StringBuilder formDataBodyBuilder = new StringBuilder();
String entryKey;
List<String> entryValue;
try {
/*
* repackage form data
*/
for (Map.Entry<String, List<String>> entry : formDataMap.entrySet()) {entryKey = entry.getKey();
entryValue = entry.getValue();
if (entryValue.size() > 1) {for(String value : entryValue){formDataBodyBuilder.append(entryKey).append("=").append(value).append("&");
}
} else {formDataBodyBuilder.append(entryKey).append("=").append(entryValue.get(0)).append("&");
}
}
}catch (Exception e){e.printStackTrace();
}
/*
* substring with the last char '&'
*/
String formDataBodyString = "";
if(formDataBodyBuilder.length()>0){formDataBodyString = formDataBodyBuilder.substring(0, formDataBodyBuilder.length() - 1);
}
/*
* get data bytes
*/
byte[] bodyBytes = formDataBodyString.getBytes(charset);
int contentLength = bodyBytes.length;
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.putAll(exchange.getRequest().getHeaders());
httpHeaders.remove(HttpHeaders.CONTENT_LENGTH);
/*
* in case of content-length not matched
*/
httpHeaders.setContentLength(contentLength);
/**
* validate request params or form data
*/
Map<String, Object> bodyMap = StringUtils.isEmpty(formDataBodyString)?new HashMap<String, Object>():decodeFormBody(formDataBodyString);
Result checkResult = validParam(exchange, bodyMap);
if(checkResult.getCode()!=0){return errorInfo(exchange, checkResult.getCode(), checkResult.getMessage());
}
/*
* use BodyInserter to InsertFormData Body
*/
BodyInserter<String, ReactiveHttpOutputMessage> bodyInserter = BodyInserters.fromObject(formDataBodyString);
CachedBodyOutputMessage cachedBodyOutputMessage = new CachedBodyOutputMessage(exchange, httpHeaders);
log.debug("[GatewayContext]Rewrite Form Data :{}",formDataBodyString);
return bodyInserter.insert(cachedBodyOutputMessage, new BodyInserterContext())
.then(Mono.defer(() -> {
ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(exchange.getRequest()) {
@Override
public HttpHeaders getHeaders() {return httpHeaders;}
@Override
public Flux<DataBuffer> getBody() {return cachedBodyOutputMessage.getBody();
}
};
ServerHttpResponse originalResponse = exchange.getResponse();
originalResponse.getHeaders().setContentType(MediaType.APPLICATION_JSON);
DataBufferFactory bufferFactory = originalResponse.bufferFactory();
ServerHttpResponseDecorator response = buildResponse(originalResponse, bufferFactory, (Map)checkResult.getResult());
return chain.filter(exchange.mutate().request(decorator).response(response).build());
}));
});
}有时须要对返回的数据对立解决，那么能够通过封装 ServerHttpResponseDecorator 进行解决，ServerHttpResponse 装璜器 ServerHttpResponseDecorator，次要笼罩写入响应体数据缓冲区的局部。
private ServerHttpResponseDecorator buildResponse(ServerHttpResponse originalResponse, DataBufferFactory bufferFactory, Map result) {return new ServerHttpResponseDecorator(originalResponse) {
@Override
public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {System.out.println("++++++++++++++++++++++++1");
if (getStatusCode().equals(HttpStatus.OK) && body instanceof Flux) {Flux<? extends DataBuffer> fluxBody = Flux.from(body);
return super.writeWith(fluxBody.buffer().map(dataBuffers -> {DataBufferFactory dataBufferFactory = new DefaultDataBufferFactory();
DataBuffer join = dataBufferFactory.join(dataBuffers);
byte[] content = new byte[join.readableByteCount()];
join.read(content);
DataBufferUtils.release(join);
// 流转为字符串
String responseData = new String(content, Charsets.UTF_8);
System.out.println(responseData);
Map map = JSON.parseObject(responseData);
// 解决返回的数据
//To do
byte[] uppedContent = responseData.getBytes(Charsets.UTF_8);
originalResponse.getHeaders().setContentLength(uppedContent.length);
return bufferFactory.wrap(uppedContent);
}));
} else {System.out.println("----------"+getStatusCode());
}
return super.writeWith(body);
}
@Override
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {return writeWith(Flux.from(body).flatMapSequential(p -> p));
}
};
}
正文完
