diff --git a/fizz-bootstrap/pom.xml b/fizz-bootstrap/pom.xml index 3c3de67..f119d23 100644 --- a/fizz-bootstrap/pom.xml +++ b/fizz-bootstrap/pom.xml @@ -12,13 +12,13 @@ com.fizzgate fizz-bootstrap - 2.2.3 + 2.2.4-SNAPSHOT 1.8 5.2.13.RELEASE Dragonfruit-SR3 - Dysprosium-SR21 + Dysprosium-SR22 5.3.7.RELEASE 4.1.66.Final 4.4.14 diff --git a/fizz-common/pom.xml b/fizz-common/pom.xml index 8c26568..2225981 100644 --- a/fizz-common/pom.xml +++ b/fizz-common/pom.xml @@ -5,7 +5,7 @@ fizz-gateway-community com.fizzgate - 2.2.3 + 2.2.4-SNAPSHOT ../pom.xml 4.0.0 diff --git a/fizz-core/src/main/java/we/constants/CommonConstants.java b/fizz-common/src/main/java/we/constants/CommonConstants.java similarity index 100% rename from fizz-core/src/main/java/we/constants/CommonConstants.java rename to fizz-common/src/main/java/we/constants/CommonConstants.java diff --git a/fizz-core/src/main/java/we/flume/clients/log4j2appender/LogService.java b/fizz-common/src/main/java/we/flume/clients/log4j2appender/LogService.java similarity index 100% rename from fizz-core/src/main/java/we/flume/clients/log4j2appender/LogService.java rename to fizz-common/src/main/java/we/flume/clients/log4j2appender/LogService.java diff --git a/fizz-core/src/main/java/we/flume/clients/log4j2appender/ThreadContext.java b/fizz-common/src/main/java/we/flume/clients/log4j2appender/ThreadContext.java similarity index 100% rename from fizz-core/src/main/java/we/flume/clients/log4j2appender/ThreadContext.java rename to fizz-common/src/main/java/we/flume/clients/log4j2appender/ThreadContext.java diff --git a/fizz-common/src/main/java/we/util/ConvertedRequestBodyDataBufferWrapper.java b/fizz-common/src/main/java/we/util/ConvertedRequestBodyDataBufferWrapper.java new file mode 100644 index 0000000..113bf3e --- /dev/null +++ b/fizz-common/src/main/java/we/util/ConvertedRequestBodyDataBufferWrapper.java @@ -0,0 +1,33 @@ +/* + * Copyright (C) 2020 the original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package we.util; + +import org.springframework.core.io.buffer.DataBuffer; + +/** + * @author hongqiaowei + */ + +public class ConvertedRequestBodyDataBufferWrapper { + + public DataBuffer body; + + public ConvertedRequestBodyDataBufferWrapper(DataBuffer body) { + this.body = body; + } +} diff --git a/fizz-common/src/main/java/we/util/NettyDataBufferUtils.java b/fizz-common/src/main/java/we/util/NettyDataBufferUtils.java new file mode 100644 index 0000000..8b36a3f --- /dev/null +++ b/fizz-common/src/main/java/we/util/NettyDataBufferUtils.java @@ -0,0 +1,80 @@ +/* + * Copyright (C) 2020 the original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package we.util; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.NettyDataBuffer; +import org.springframework.core.io.buffer.NettyDataBufferFactory; +import org.springframework.core.io.buffer.PooledDataBuffer; +import org.springframework.lang.Nullable; +import we.flume.clients.log4j2appender.LogService; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +/** + * @author hongqiaowei + */ + +public abstract class NettyDataBufferUtils extends org.springframework.core.io.buffer.DataBufferUtils { + + private static final Logger log = LoggerFactory.getLogger(NettyDataBufferUtils.class); + + private static NettyDataBufferFactory dataBufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT); + + public static NettyDataBuffer from(String s) { + return from(s.getBytes(StandardCharsets.UTF_8)); + } + + public static NettyDataBuffer from(byte[] bytes) { + return (NettyDataBuffer) dataBufferFactory.wrap(bytes); + } + + public static NettyDataBuffer from(ByteBuffer byteBuffer) { + return dataBufferFactory.wrap(byteBuffer); + } + + public static NettyDataBuffer from(ByteBuf byteBuf) { + return dataBufferFactory.wrap(byteBuf); + } + + public static boolean release(@Nullable String reqId, @Nullable DataBuffer dataBuffer) { + if (dataBuffer instanceof PooledDataBuffer) { + PooledDataBuffer pooledDataBuffer = (PooledDataBuffer) dataBuffer; + if (pooledDataBuffer.isAllocated()) { + if (pooledDataBuffer instanceof NettyDataBuffer) { + NettyDataBuffer ndb = (NettyDataBuffer) pooledDataBuffer; + ByteBuf nativeBuffer = ndb.getNativeBuffer(); + int refCnt = nativeBuffer.refCnt(); + if (refCnt < 1) { + if (log.isDebugEnabled()) { + log.debug(nativeBuffer + " ref cnt is " + refCnt, LogService.BIZ_ID, reqId); + } + return false; + } + } + return pooledDataBuffer.release(); + } + } + return false; + } +} diff --git a/fizz-core/pom.xml b/fizz-core/pom.xml index 60cab3e..298dd92 100644 --- a/fizz-core/pom.xml +++ b/fizz-core/pom.xml @@ -5,7 +5,7 @@ fizz-gateway-community com.fizzgate - 2.2.3 + 2.2.4-SNAPSHOT ../pom.xml 4.0.0 diff --git a/fizz-core/src/main/java/we/filter/AggregateFilter.java b/fizz-core/src/main/java/we/filter/AggregateFilter.java index c22926d..c072d70 100644 --- a/fizz-core/src/main/java/we/filter/AggregateFilter.java +++ b/fizz-core/src/main/java/we/filter/AggregateFilter.java @@ -61,6 +61,7 @@ import we.flume.clients.log4j2appender.LogService; import we.plugin.auth.ApiConfig; import we.util.Constants; import we.util.MapUtil; +import we.util.NettyDataBufferUtils; import we.util.WebUtils; /** @@ -72,8 +73,6 @@ public class AggregateFilter implements WebFilter { private static final Logger LOGGER = LoggerFactory.getLogger(AggregateFilter.class); - private static final DataBuffer emptyBody = new NettyDataBufferFactory(new UnpooledByteBufAllocator(false, true)).wrap(Constants.Symbol.EMPTY.getBytes()); - @Resource private ConfigLoader configLoader; @@ -171,19 +170,12 @@ public class AggregateFilter implements WebFilter { }); } else { if (HttpMethod.POST.name().equalsIgnoreCase(method)) { - result = DataBufferUtils.join(request.getBody()).defaultIfEmpty(emptyBody).flatMap(buf -> { - if (buf != null && buf != emptyBody) { - try { - clientInput.put("body", buf.toString(StandardCharsets.UTF_8)); - } finally { - DataBufferUtils.release(buf); - } - } - return pipeline.run(input, clientInput, traceId); - }); - } else { - result = pipeline.run(input, clientInput, traceId); + DataBuffer buf = WebUtils.getRequestBody(exchange); + if (buf != null) { + clientInput.put("body", buf.toString(StandardCharsets.UTF_8)); + } } + result = pipeline.run(input, clientInput, traceId); } return result.subscribeOn(Schedulers.elastic()).flatMap(aggResult -> { LogService.setBizId(traceId); diff --git a/fizz-core/src/main/java/we/filter/CallbackFilter.java b/fizz-core/src/main/java/we/filter/CallbackFilter.java index 4639689..7fd71c6 100644 --- a/fizz-core/src/main/java/we/filter/CallbackFilter.java +++ b/fizz-core/src/main/java/we/filter/CallbackFilter.java @@ -18,14 +18,11 @@ package we.filter; import com.alibaba.fastjson.JSON; -import io.netty.buffer.UnpooledByteBufAllocator; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.core.annotation.Order; import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferUtils; -import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.data.redis.core.ReactiveStringRedisTemplate; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatus; @@ -90,33 +87,15 @@ public class CallbackFilter extends FizzWebFilter { ApiConfig ac = WebUtils.getApiConfig(exchange); if (ac != null && ac.type == ApiConfig.Type.CALLBACK) { CallbackConfig cc = ac.callbackConfig; - ServerHttpRequest req = exchange.getRequest(); - DataBuffer[] body = {null}; - return - DataBufferUtils.join(req.getBody()).defaultIfEmpty(WebUtils.EMPTY_BODY) - .flatMap( - b -> { - if (b != WebUtils.EMPTY_BODY) { - body[0] = b; - } - HashMap service2instMap = getService2instMap(ac); - HttpHeaders headers = WebUtils.mergeAppendHeaders(exchange); - pushReq2manager(exchange, headers, body[0], service2instMap, cc.id, ac.gatewayGroups.iterator().next()); - if (cc.type == CallbackConfig.Type.ASYNC || StringUtils.isNotBlank(cc.respBody)) { - return directResponse(exchange, cc); - } else { - return callbackService.requestBackends(exchange, headers, body[0], cc, service2instMap); - } - } - ) - .doFinally( - s -> { - if (body[0] != null) { - DataBufferUtils.release(body[0]); - } - } - ) - ; + DataBuffer body = WebUtils.getRequestBody(exchange); + HashMap service2instMap = getService2instMap(ac); + HttpHeaders headers = WebUtils.mergeAppendHeaders(exchange); + pushReq2manager(exchange, headers, body, service2instMap, cc.id, ac.gatewayGroups.iterator().next()); + if (cc.type == CallbackConfig.Type.ASYNC || StringUtils.isNotBlank(cc.respBody)) { + return directResponse(exchange, cc); + } else { + return callbackService.requestBackends(exchange, headers, body, cc, service2instMap); + } } return chain.filter(exchange); } diff --git a/fizz-core/src/main/java/we/filter/FilterExceptionHandlerConfig.java b/fizz-core/src/main/java/we/filter/FilterExceptionHandlerConfig.java index e1bfd49..bdcfb82 100644 --- a/fizz-core/src/main/java/we/filter/FilterExceptionHandlerConfig.java +++ b/fizz-core/src/main/java/we/filter/FilterExceptionHandlerConfig.java @@ -22,6 +22,7 @@ import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.annotation.Order; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; @@ -35,9 +36,7 @@ import we.exception.StopAndResponseException; import we.fizz.exception.FizzRuntimeException; import we.flume.clients.log4j2appender.LogService; import we.legacy.RespEntity; -import we.util.JacksonUtils; -import we.util.ThreadContext; -import we.util.WebUtils; +import we.util.*; import java.net.URI; @@ -54,7 +53,7 @@ public class FilterExceptionHandlerConfig { @Override public Mono handle(ServerWebExchange exchange, Throwable t) { ServerHttpResponse resp = exchange.getResponse(); - if (t instanceof StopAndResponseException) { + if (t instanceof StopAndResponseException) { StopAndResponseException ex = (StopAndResponseException) t; if (ex.getData() != null) { resp.getHeaders().add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE); @@ -83,8 +82,8 @@ public class FilterExceptionHandlerConfig { } } if (t instanceof FizzRuntimeException) { - FizzRuntimeException ex = (FizzRuntimeException) t; - log.error(ex.getMessage(), LogService.BIZ_ID, exchange.getRequest().getId(), ex); + FizzRuntimeException ex = (FizzRuntimeException) t; + log.error(ex.getMessage(), LogService.BIZ_ID, exchange.getRequest().getId(), ex); resp.getHeaders().add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE); RespEntity rs = null; String reqId = exchange.getRequest().getId(); diff --git a/fizz-core/src/main/java/we/filter/PreprocessFilter.java b/fizz-core/src/main/java/we/filter/PreprocessFilter.java index 82e9dfd..0da67de 100644 --- a/fizz-core/src/main/java/we/filter/PreprocessFilter.java +++ b/fizz-core/src/main/java/we/filter/PreprocessFilter.java @@ -17,12 +17,24 @@ package we.filter; +import com.google.common.collect.BoundType; +import io.netty.buffer.ByteBuf; +import io.netty.util.ReferenceCountUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.core.annotation.Order; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferUtils; +import org.springframework.core.io.buffer.DefaultDataBufferFactory; +import org.springframework.core.io.buffer.NettyDataBuffer; import org.springframework.http.HttpStatus; +import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.stereotype.Component; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebFilterChain; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import we.flume.clients.log4j2appender.LogService; import we.plugin.FixedPluginFilter; import we.plugin.FizzPluginFilterChain; import we.plugin.PluginFilter; @@ -30,10 +42,14 @@ import we.plugin.auth.ApiConfig; import we.plugin.auth.ApiConfigService; import we.plugin.auth.AuthPluginFilter; import we.plugin.stat.StatPluginFilter; +import we.util.ConvertedRequestBodyDataBufferWrapper; +import we.util.NettyDataBufferUtils; import we.util.ReactorUtils; import we.util.WebUtils; import javax.annotation.Resource; +import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -47,6 +63,8 @@ import java.util.function.Function; @Order(10) public class PreprocessFilter extends FizzWebFilter { + private static final Logger log = LoggerFactory.getLogger(PreprocessFilter.class); + public static final String PREPROCESS_FILTER = "preprocessFilter"; private static final FilterResult succFr = FilterResult.SUCCESS(PREPROCESS_FILTER); @@ -65,7 +83,42 @@ public class PreprocessFilter extends FizzWebFilter { Map eas = exchange.getAttributes(); eas.put(WebUtils.FILTER_CONTEXT, fc); eas.put(WebUtils.APPEND_HEADERS, appendHdrs); - Mono vm = statPluginFilter.filter(exchange, null, null); + ServerHttpRequest req = exchange.getRequest(); + return NettyDataBufferUtils.join(req.getBody()).defaultIfEmpty(WebUtils.EMPTY_BODY) + .flatMap( + body -> { + if (body != WebUtils.EMPTY_BODY && body.readableByteCount() > 0) { + try { + byte[] bytes = new byte[body.readableByteCount()]; + body.read(bytes); + DataBuffer retain = NettyDataBufferUtils.from(bytes); + eas.put(WebUtils.REQUEST_BODY, retain); + } finally { + NettyDataBufferUtils.release(body); + } + } + Mono vm = statPluginFilter.filter(exchange, null, null); + return process(exchange, chain, eas, vm); + } + ) + .doFinally( + s -> { + Object convertedRequestBody = WebUtils.getConvertedRequestBody(exchange); + if (convertedRequestBody instanceof ConvertedRequestBodyDataBufferWrapper) { + DataBuffer b = ((ConvertedRequestBodyDataBufferWrapper) convertedRequestBody).body; + if (b != null) { + boolean release = NettyDataBufferUtils.release(req.getId(), b); + if (log.isDebugEnabled()) { + log.debug("release converted request body databuffer " + release, LogService.BIZ_ID, req.getId()); + } + } + } + } + ); + } + + // TODO + private Mono process(ServerWebExchange exchange, WebFilterChain chain, Map eas, Mono vm) { return chain(exchange, vm, authPluginFilter).defaultIfEmpty(ReactorUtils.NULL) .flatMap( v -> { diff --git a/fizz-core/src/main/java/we/filter/RouteFilter.java b/fizz-core/src/main/java/we/filter/RouteFilter.java index dc2e302..49b792a 100644 --- a/fizz-core/src/main/java/we/filter/RouteFilter.java +++ b/fizz-core/src/main/java/we/filter/RouteFilter.java @@ -112,7 +112,11 @@ public class RouteFilter extends FizzWebFilter { String uri = ThreadContext.getStringBuilder().append(ac.getNextHttpHostPort()) .append(WebUtils.appendQuery(WebUtils.getBackendPath(exchange), exchange)) .toString(); - return fizzWebClient.send(rid, req.getMethod(), uri, hdrs, req.getBody()).flatMap(genServerResponse(exchange)); + Object requestBody = WebUtils.getConvertedRequestBody(exchange); + if (requestBody == null) { + requestBody = WebUtils.getRequestBody(exchange); + } + return fizzWebClient.send(rid, req.getMethod(), uri, hdrs, requestBody).flatMap(genServerResponse(exchange)); } else if (ac.type == ApiConfig.Type.DUBBO) { return dubboRpc(exchange, ac); @@ -128,7 +132,11 @@ public class RouteFilter extends FizzWebFilter { private Mono send(ServerWebExchange exchange, String service, String relativeUri, HttpHeaders hdrs) { ServerHttpRequest clientReq = exchange.getRequest(); - return fizzWebClient.send2service(clientReq.getId(), clientReq.getMethod(), service, relativeUri, hdrs, clientReq.getBody()).flatMap(genServerResponse(exchange)); + Object requestBody = WebUtils.getConvertedRequestBody(exchange); + if (requestBody == null) { + requestBody = WebUtils.getRequestBody(exchange); + } + return fizzWebClient.send2service(clientReq.getId(), clientReq.getMethod(), service, relativeUri, hdrs, requestBody).flatMap(genServerResponse(exchange)); } private Function> genServerResponse(ServerWebExchange exchange) { @@ -171,64 +179,52 @@ public class RouteFilter extends FizzWebFilter { } private Mono dubboRpc(ServerWebExchange exchange, ApiConfig ac) { - DataBuffer[] body = {null}; - return DataBufferUtils.join(exchange.getRequest().getBody()).defaultIfEmpty(WebUtils.EMPTY_BODY) - .flatMap( - b -> { - HashMap parameters = null; - if (b != WebUtils.EMPTY_BODY) { - body[0] = b; - String json = body[0].toString(StandardCharsets.UTF_8).trim(); - if (json.charAt(0) == '[') { - ArrayList lst = JacksonUtils.readValue(json, ArrayList.class); - parameters = new HashMap<>(); - for (int i = 0; i < lst.size(); i++) { - parameters.put("p" + (i + 1), lst.get(i)); - } - } else { - parameters = JacksonUtils.readValue(json, HashMap.class); - } - } - DubboInterfaceDeclaration declaration = new DubboInterfaceDeclaration(); - declaration.setServiceName(ac.backendService); - declaration.setVersion(ac.rpcVersion); - declaration.setGroup(ac.rpcGroup); - declaration.setMethod(ac.rpcMethod); - declaration.setParameterTypes(ac.rpcParamTypes); - int t = 20_000; - if (ac.timeout != 0) { - t = (int) ac.timeout; - } - declaration.setTimeout(t); + DataBuffer b = WebUtils.getRequestBody(exchange); + HashMap parameters = null; + String json = Constants.Symbol.EMPTY; + if (b != null) { + json = b.toString(StandardCharsets.UTF_8).trim(); + if (json.charAt(0) == '[') { + ArrayList lst = JacksonUtils.readValue(json, ArrayList.class); + parameters = new HashMap<>(); + for (int i = 0; i < lst.size(); i++) { + parameters.put("p" + (i + 1), lst.get(i)); + } + } else { + parameters = JacksonUtils.readValue(json, HashMap.class); + } + } + String finalJson = json; - Map attachments = Collections.singletonMap(CommonConstants.HEADER_TRACE_ID, WebUtils.getTraceId(exchange)); - return dubboGenericService.send(parameters, declaration, attachments); - } - ) - .flatMap( - dubboRpcResponseBody -> { - Mono m = WebUtils.buildJsonDirectResponse(exchange, HttpStatus.OK, null, JacksonUtils.writeValueAsString(dubboRpcResponseBody)); - return m; - } - ) - .doOnError( - t -> { - StringBuilder b = ThreadContext.getStringBuilder(); - WebUtils.request2stringBuilder(exchange, b); - if (body[0] != null) { - b.append('\n').append(body[0].toString(StandardCharsets.UTF_8)); - } - log.error(b.toString(), LogService.BIZ_ID, exchange.getRequest().getId(), t); - } - ) - .doFinally( - s -> { - if (body[0] != null) { - DataBufferUtils.release(body[0]); - } - } - ) - ; + DubboInterfaceDeclaration declaration = new DubboInterfaceDeclaration(); + declaration.setServiceName(ac.backendService); + declaration.setVersion(ac.rpcVersion); + declaration.setGroup(ac.rpcGroup); + declaration.setMethod(ac.rpcMethod); + declaration.setParameterTypes(ac.rpcParamTypes); + int t = 20_000; + if (ac.timeout != 0) { + t = (int) ac.timeout; + } + declaration.setTimeout(t); + + Map attachments = Collections.singletonMap(CommonConstants.HEADER_TRACE_ID, WebUtils.getTraceId(exchange)); + return dubboGenericService.send(parameters, declaration, attachments) + .flatMap( + dubboRpcResponseBody -> { + Mono m = WebUtils.buildJsonDirectResponse(exchange, HttpStatus.OK, null, JacksonUtils.writeValueAsString(dubboRpcResponseBody)); + return m; + } + ) + .doOnError( + e -> { + StringBuilder sb = ThreadContext.getStringBuilder(); + WebUtils.request2stringBuilder(exchange, sb); + sb.append('\n').append(finalJson); + log.error(sb.toString(), LogService.BIZ_ID, exchange.getRequest().getId(), e); + } + ) + ; } } diff --git a/fizz-core/src/main/java/we/plugin/auth/ApiConfigService.java b/fizz-core/src/main/java/we/plugin/auth/ApiConfigService.java index 9e174e4..13fb5f7 100644 --- a/fizz-core/src/main/java/we/plugin/auth/ApiConfigService.java +++ b/fizz-core/src/main/java/we/plugin/auth/ApiConfigService.java @@ -296,19 +296,18 @@ public class ApiConfigService { // TODO: improve ... private Mono canAccess(ServerWebExchange exchange, String app, String ip, String timestamp, String sign, String service, HttpMethod method, String path) { + if (!systemConfig.isAggregateTestAuth()) { + if (SystemConfig.DEFAULT_GATEWAY_TEST_PREFIX0.equals(WebUtils.getClientReqPathPrefix(exchange))) { + return Mono.just(Access.YES); + } + } + ApiConfig ac = getApiConfig(app, service, method, path); if (ac == null) { String authMsg = (String) ThreadContext.remove(AUTH_MSG); if (authMsg == null) { authMsg = deny; } - if (SystemConfig.DEFAULT_GATEWAY_TEST_PREFIX0.equals(WebUtils.getClientReqPathPrefix(exchange))) { - if (systemConfig.isAggregateTestAuth()) { - return logAndResult(authMsg); - } else { - return Mono.just(Access.YES); - } - } if (!apiConfigServiceProperties.isNeedAuth()) { return Mono.just(Access.YES); } else { diff --git a/fizz-core/src/main/java/we/proxy/CallbackService.java b/fizz-core/src/main/java/we/proxy/CallbackService.java index 8873bcc..8cd24c6 100644 --- a/fizz-core/src/main/java/we/proxy/CallbackService.java +++ b/fizz-core/src/main/java/we/proxy/CallbackService.java @@ -78,7 +78,7 @@ public class CallbackService { aggrConfigPrefix = systemConfig.getGatewayPrefix() + '/'; } - public Mono requestBackends(ServerWebExchange exchange, HttpHeaders headers, DataBuffer body, CallbackConfig cc, Map service2instMap) { + public Mono requestBackends(ServerWebExchange exchange, HttpHeaders headers, DataBuffer body, CallbackConfig cc, Map service2instMap) { ServerHttpRequest req = exchange.getRequest(); String reqId = req.getId(); HttpMethod method = req.getMethod(); @@ -172,7 +172,7 @@ public class CallbackService { return b.toString(); } - private Mono genServerResponse(ServerWebExchange exchange, ClientResponse remoteResp) { + private Mono genServerResponse(ServerWebExchange exchange, ClientResponse remoteResp) { ServerHttpResponse clientResp = exchange.getResponse(); clientResp.setStatusCode(remoteResp.statusCode()); HttpHeaders clientRespHeaders = clientResp.getHeaders(); diff --git a/fizz-core/src/main/java/we/proxy/FizzWebClient.java b/fizz-core/src/main/java/we/proxy/FizzWebClient.java index 7649e1b..ce82d32 100644 --- a/fizz-core/src/main/java/we/proxy/FizzWebClient.java +++ b/fizz-core/src/main/java/we/proxy/FizzWebClient.java @@ -21,6 +21,7 @@ import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.NettyDataBuffer; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; import org.springframework.lang.Nullable; @@ -34,9 +35,7 @@ import reactor.core.publisher.Mono; import we.config.ProxyWebClientConfig; import we.config.SystemConfig; import we.flume.clients.log4j2appender.LogService; -import we.util.Constants; -import we.util.ThreadContext; -import we.util.WebUtils; +import we.util.*; import javax.annotation.Resource; import java.time.Duration; @@ -117,19 +116,25 @@ public class FizzWebClient { } ); + boolean b = false; + DataBuffer d = null; if (body != null) { if (body instanceof BodyInserter) { req.body((BodyInserter) body); } else if (body instanceof Flux) { Flux db = (Flux) body; req.body(BodyInserters.fromDataBuffers(db)); - } else if (body instanceof String) { - String s = (String) body; - req.body(Mono.just(s), String.class); } else { + if (body instanceof ConvertedRequestBodyDataBufferWrapper) { + d = ((ConvertedRequestBodyDataBufferWrapper) body).body; + body = d; + b = true; + } req.bodyValue(body); } } + boolean finalB = b; + DataBuffer finalD = d; Mono cr = req.exchange(); if (timeout == 0) { @@ -140,7 +145,17 @@ public class FizzWebClient { if (timeout != 0) { cr = cr.timeout(Duration.ofMillis(timeout)); } - return cr; + + return cr.doFinally( + s -> { + if (finalB) { + boolean release = NettyDataBufferUtils.release(clientReqId, finalD); + if (log.isDebugEnabled()) { + log.debug("release converted request body databuffer " + release, LogService.BIZ_ID, clientReqId); + } + } + } + ); } private void setHostHeader(String uri, HttpHeaders headers) { diff --git a/fizz-core/src/main/java/we/util/WebUtils.java b/fizz-core/src/main/java/we/util/WebUtils.java index 5f190e0..41a7972 100644 --- a/fizz-core/src/main/java/we/util/WebUtils.java +++ b/fizz-core/src/main/java/we/util/WebUtils.java @@ -22,6 +22,7 @@ import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; 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; @@ -56,7 +57,7 @@ public abstract class WebUtils { private static final Logger log = LoggerFactory.getLogger(WebUtils.class); - private static final String clientService = "clientService"; + private static final String clientService = "@cs"; private static final String xForwardedFor = "X-FORWARDED-FOR"; @@ -66,17 +67,17 @@ public abstract class WebUtils { private static final String binaryAddress = "0:0:0:0:0:0:0:1"; - private static final String directResponse = "directResponse"; + private static final String directResponse = "@dr"; private static final String response = " response "; - private static final String originIp = "originIp"; + private static final String originIp = "@oi"; - private static final String clientRequestPath = "clientRequestPath"; + private static final String clientRequestPath = "@crp"; - private static final String clientRequestPathPrefix = "clientRequestPathPrefix"; + private static final String clientRequestPathPrefix = "@crpp"; - private static final String clientRequestQuery = "clientRequestQuery"; + private static final String clientRequestQuery = "@crq"; private static final String traceId = "traceId"; @@ -86,23 +87,57 @@ public abstract class WebUtils { private static final String app = "app"; - public static final String BACKEND_SERVICE = "backendService"; + public static final String BACKEND_SERVICE = "@bs"; - public static final String FILTER_CONTEXT = "filterContext"; + public static final String FILTER_CONTEXT = "@fc"; - public static final String APPEND_HEADERS = "appendHeaders"; + public static final String APPEND_HEADERS = "@ahs"; - public static final String PREV_FILTER_RESULT = "prevFilterResult"; + public static final String PREV_FILTER_RESULT = "@pfr"; - public static final String BACKEND_PATH = "backendPath"; + public static final String BACKEND_PATH = "@bp"; public static boolean LOG_RESPONSE_BODY = false; public static Set LOG_HEADER_SET = Collections.EMPTY_SET; - public static final DataBuffer EMPTY_BODY = new NettyDataBufferFactory(new UnpooledByteBufAllocator(false, true)).wrap(Constants.Symbol.EMPTY.getBytes()); + public static final DataBuffer EMPTY_BODY = NettyDataBufferUtils.from(Constants.Symbol.EMPTY); + + public static final String REQUEST_BODY = "@rb"; + + public static final String CONVERTED_REQUEST_BODY = "@crb"; + public static DataBuffer getRequestBody(ServerWebExchange exchange) { + return exchange.getAttribute(REQUEST_BODY); + } + + /** + * @param convertedRequestBody can be DataBuffer or String type + */ + public static void setConvertedRequestBody(ServerWebExchange exchange, Object convertedRequestBody) { + Object prev = exchange.getAttribute(CONVERTED_REQUEST_BODY); + if (prev instanceof ConvertedRequestBodyDataBufferWrapper) { + ConvertedRequestBodyDataBufferWrapper p = (ConvertedRequestBodyDataBufferWrapper) prev; + if (p.body != null) { + DataBufferUtils.release(p.body); + } + } + if (convertedRequestBody instanceof DataBuffer) { + DataBuffer d = (DataBuffer) convertedRequestBody; + DataBufferUtils.retain(d); + exchange.getAttributes().put(CONVERTED_REQUEST_BODY, new ConvertedRequestBodyDataBufferWrapper(d)); + } else { + exchange.getAttributes().put(CONVERTED_REQUEST_BODY, convertedRequestBody); + } + } + + /** + * @return result may be String or ConvertedRequestBodyDataBufferWrapper type + */ + public static Object getConvertedRequestBody(ServerWebExchange exchange) { + return exchange.getAttribute(CONVERTED_REQUEST_BODY); + } public static void setGatewayPrefix(String p) { gatewayPrefix = p; @@ -383,10 +418,14 @@ public abstract class WebUtils { if (appendHeaders.isEmpty()) { return req.getHeaders(); } + boolean b = appendHeaders.containsKey(HttpHeaders.CONTENT_TYPE); HttpHeaders hdrs = new HttpHeaders(); req.getHeaders().forEach( (h, vs) -> { - hdrs.addAll(h, vs); + if (b && h.equals(HttpHeaders.CONTENT_TYPE)) { + } else { + hdrs.addAll(h, vs); + } } ); appendHeaders.forEach( diff --git a/fizz-plugin/pom.xml b/fizz-plugin/pom.xml index facf940..e41a364 100644 --- a/fizz-plugin/pom.xml +++ b/fizz-plugin/pom.xml @@ -5,7 +5,7 @@ fizz-gateway-community com.fizzgate - 2.2.3 + 2.2.4-SNAPSHOT ../pom.xml 4.0.0 diff --git a/fizz-spring-boot-starter/pom.xml b/fizz-spring-boot-starter/pom.xml index 73fc70c..ca7f550 100644 --- a/fizz-spring-boot-starter/pom.xml +++ b/fizz-spring-boot-starter/pom.xml @@ -5,7 +5,7 @@ fizz-gateway-community com.fizzgate - 2.2.3 + 2.2.4-SNAPSHOT ../pom.xml 4.0.0 diff --git a/pom.xml b/pom.xml index 5f5d51a..1c97a0a 100644 --- a/pom.xml +++ b/pom.xml @@ -7,7 +7,7 @@ 2.2.13.RELEASE 5.2.13.RELEASE - Dysprosium-SR21 + Dysprosium-SR22 5.3.7.RELEASE 2.2.5.RELEASE 4.1.66.Final @@ -33,7 +33,7 @@ fizz-gateway-community ${project.artifactId} fizz gateway community - 2.2.3 + 2.2.4-SNAPSHOT pom fizz-common @@ -133,7 +133,7 @@ io.projectreactor reactor-test - 3.3.18.RELEASE + 3.3.19.RELEASE test @@ -502,5 +502,4 @@ -