gateway请求拦截_spring cloud gateway 拦截request Body
在接入Spring-Cloud-Gateway時，可能有需求進(jìn)行緩存Json-Body數(shù)據(jù)或者Form-Urlencoded數(shù)據(jù)的情況。
由于Spring-Cloud-Gateway是以WebFlux為基礎(chǔ)的響應(yīng)式架構(gòu)設(shè)計，所以在原有Zuul基礎(chǔ)上遷移過來的過程中，傳統(tǒng)的編程思路，并不適合于Reactor Stream的開發(fā)。
網(wǎng)絡(luò)上有許多緩存案例，但是在測試過程中出現(xiàn)各種Bug問題，在緩存Body時，需要考慮整體的響應(yīng)式操作，才能更合理的緩存數(shù)據(jù)
下面提供緩存Json-Body數(shù)據(jù)或者Form-Urlencoded數(shù)據(jù)的具體實現(xiàn)方案，該方案經(jīng)測試，滿足各方面需求，以及避免了網(wǎng)絡(luò)上其他緩存方案所出現(xiàn)的問題
定義一個GatewayContext類，用于存儲請求中的數(shù)據(jù)
import org.springframework.util.MultiValueMap;
public class GatewayContext {
public static final String CACHE_GATEWAY_CONTEXT = "cacheGatewayContext";
/**
* cache json body
*/
private String cacheBody;
/**
* cache formdata
*/
private MultiValueMap formData;
/**
* cache reqeust path
*/
private String path;
public String getCacheBody() {
return cacheBody;
}
public void setCacheBody(String cacheBody) {
this.cacheBody = cacheBody;
}
public MultiValueMap getFormData() {
return formData;
}
public void setFormData(MultiValueMap formData) {
this.formData = formData;
}
public String getPath() {
return path;
}
public void setPath(String path) {
this.path = path;
}
}
1 . 該示例只支持緩存下面3種MediaType
APPLICATION_JSON--Json數(shù)據(jù)
APPLICATION_JSON_UTF8--Json數(shù)據(jù)
APPLICATION_FORM_URLENCODED--FormData表單數(shù)據(jù)
2 . 經(jīng)驗總結(jié):
在緩存Body時,不能夠在Filter內(nèi)部直接進(jìn)行緩存，需要按照響應(yīng)式的處理方式，在異步操作路途上進(jìn)行緩存Body，由于Body只能讀取一次，所以要讀取完成后要重新封裝新的request和exchange才能保證請求正常傳遞到下游
在緩存FormData時，FormData也只能讀取一次，所以在讀取完畢后，需要重新封裝request和exchange,這里要注意，如果對FormData內(nèi)容進(jìn)行了修改，則必須重新定義Header中的content-length已保證傳輸數(shù)據(jù)的大小一致
package com.weiresearch.idss.weiark.gateway.LogReader;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.server.HandlerStrategies;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.server.ServerWebExchange;
import io.netty.buffer.ByteBufAllocator;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
// https://segmentfault.com/a/1190000017898354
@Component
public class GatewayContextFilter
implements GlobalFilter {
/**
* default HttpMessageReader
*/
private static final List> messageReaders =
HandlerStrategies.withDefaults().messageReaders();
private Logger log = LoggerFactory.getLogger(GatewayContextFilter.class);
@Override
public Mono filter(
ServerWebExchange exchange,
GatewayFilterChain chain) {
/**
* save request path and serviceId into gateway context
*/
ServerHttpRequest request = exchange.getRequest();
String path = request.getPath().pathWithinApplication().value();
GatewayContext gatewayContext = new GatewayContext();
gatewayContext.setPath(path);
/**
* save gateway context into exchange
*/
exchange.getAttributes().put(GatewayContext.CACHE_GATEWAY_CONTEXT,
gatewayContext);
HttpHeaders headers = request.getHeaders();
MediaType contentType = headers.getContentType();
log.info("start-------------------------------------------------");
log.info("HttpMethod:{},Url:{}", request.getMethod(),
request.getURI().getRawPath());
if (request.getMethod() == HttpMethod.GET) {
log.info("end-------------------------------------------------");
}
if (request.getMethod() == HttpMethod.POST) {
Mono voidMono = null;
if (MediaType.APPLICATION_JSON.equals(contentType)
|| MediaType.APPLICATION_JSON_UTF8.equals(contentType)) {
voidMono =
readBody(exchange, chain, gatewayContext);
}
if (MediaType.APPLICATION_FORM_URLENCODED.equals(contentType)) {
voidMono =
readFormData(exchange, chain, gatewayContext);
}
return voidMono;
}
/* log.debug(
"[GatewayContext]ContentType:{},Gateway context is set with {}",
contentType, gatewayContext);*/
return chain.filter(exchange);
}
/**
* ReadFormData
*
* @param exchange
* @param chain
* @return
*/
private Mono readFormData(
ServerWebExchange exchange,
GatewayFilterChain chain,
GatewayContext gatewayContext) {
final ServerHttpRequest request = exchange.getRequest();
HttpHeaders headers = request.getHeaders();
return exchange.getFormData()
.doOnNext(multiValueMap -> {
gatewayContext.setFormData(multiValueMap);
log.info("Post x-www-form-urlencoded:{}",
multiValueMap);
log.info(
"end-------------------------------------------------");
})
.then(Mono.defer(() -> {
Charset charset = headers.getContentType().getCharset();
charset = charset == null ? StandardCharsets.UTF_8 : charset;
String charsetName = charset.name();
MultiValueMap formData =
gatewayContext.getFormData();
/**
* formData is empty just return
*/
if (null == formData || formData.isEmpty()) {
return chain.filter(exchange);
}
StringBuilder formDataBodyBuilder = new StringBuilder();
String entryKey;
List entryValue;
try {
/**
* repackage form data
*/
for (Map.Entry> entry : formData
.entrySet()) {
entryKey = entry.getKey();
entryValue = entry.getValue();
if (entryValue.size() > 1) {
for (String value : entryValue) {
formDataBodyBuilder.append(entryKey).append("=")
.append(
URLEncoder.encode(value, charsetName))
.append("&");
}
} else {
formDataBodyBuilder
.append(entryKey).append("=").append(URLEncoder
.encode(entryValue.get(0), charsetName))
.append("&");
}
}
} catch (UnsupportedEncodingException e) {
// ignore URLEncode Exception
}
/**
* 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;
ServerHttpRequestDecorator decorator =
new ServerHttpRequestDecorator(
request) {
/**
* change content-length
*
* @return
*/
@Override
public HttpHeaders getHeaders() {
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.putAll(super.getHeaders());
if (contentLength > 0) {
httpHeaders.setContentLength(contentLength);
} else {
httpHeaders.set(HttpHeaders.TRANSFER_ENCODING,
"chunked");
}
return httpHeaders;
}
/**
* read bytes to Flux
*
* @return
*/
@Override
public Flux getBody() {
return DataBufferUtils
.read(new ByteArrayResource(bodyBytes),
new NettyDataBufferFactory(
ByteBufAllocator.DEFAULT),
contentLength);
}
};
ServerWebExchange mutateExchange =
exchange.mutate().request(decorator).build();
/* log.info("[GatewayContext]Rewrite Form Data :{}",
formDataBodyString);*/
return chain.filter(mutateExchange);
}));
}
/**
* ReadJsonBody
*
* @param exchange
* @param chain
* @return
*/
private Mono readBody(
ServerWebExchange exchange,
GatewayFilterChain chain,
GatewayContext gatewayContext) {
/**
* join the body
*/
return DataBufferUtils.join(exchange.getRequest().getBody())
.flatMap(dataBuffer -> {
/*
* read the body Flux, and release the buffer
* //TODO when SpringCloudGateway Version Release To G.SR2,this can be update with the new version's feature
* see PR https://github.com/spring-cloud/spring-cloud-gateway/pull/1095
*/
byte[] bytes = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(bytes);
DataBufferUtils.release(dataBuffer);
Flux 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 getBody() {
return cachedFlux;
}
};
/**
* mutate exchage with new ServerHttpRequest
*/
ServerWebExchange mutatedExchange =
exchange.mutate().request(mutatedRequest).build();
/**
* read body string with default messageReaders
*/
return ServerRequest.create(mutatedExchange, messageReaders)
.bodyToMono(String.class)
.doOnNext(objectValue -> {
log.info("PostBody:{}", objectValue);
log.info(
"end-------------------------------------------------");
gatewayContext.setCacheBody(objectValue);
/* log.debug("[GatewayContext]Read JsonBody:{}",
objectValue);*/
}).then(chain.filter(mutatedExchange));
});
}
}
在后續(xù)Filter中，可以直接從ServerExchange中獲取GatewayContext，就可以獲取到緩存的數(shù)據(jù)，如果需要緩存其他數(shù)據(jù)，則可以根據(jù)自己的需求，添加到GatewayContext中即可
GatewayContext gatewayContext = exchange.getAttribute(GatewayContext.CACHE_GATEWAY_CONTEXT);
總結(jié)
以上是生活随笔為你收集整理的gateway请求拦截_spring cloud gateway 拦截request Body的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： UEFI+GPT安装Windows8和C
- 下一篇： UEFI主板GPT方式安装CentOS