支持接口转发
This commit is contained in:
parent
d495905584
commit
b0fce77c05
@ -43,6 +43,7 @@ import io.vertx.core.json.JsonObject;
|
||||
import io.vertx.core.net.NetSocket;
|
||||
import io.vertx.core.net.SocketAddress;
|
||||
import io.vertx.ext.web.RoutingContext;
|
||||
import io.vertx.ext.web.client.HttpRequest;
|
||||
import io.vertx.ext.web.client.WebClient;
|
||||
import io.vertx.ext.web.handler.HttpException;
|
||||
import io.vertx.httpproxy.Body;
|
||||
@ -242,11 +243,10 @@ public class ReverseProxy implements HttpProxy {
|
||||
|
||||
private Future<ProxyResponse> breakerAndSecurity(CircuitBreaker circuitBreaker, ProxyRequest proxyRequest) {
|
||||
String appCode = proxyRequest.headers().get(AppConfigHandler.getAppCodeHeaderKey());
|
||||
SocketAddress socketAddress = ProxyTool.resolveOriginAddress(proxyRequest.proxiedRequest());
|
||||
return Future.future(p -> {
|
||||
circuitBreaker.executeWithFallback(promise -> {
|
||||
mainWebClient.post(socketAddress.port(), socketAddress.host(), proxyRequest.getURI())
|
||||
.putHeaders(proxyRequest.headers())
|
||||
HttpRequest<Buffer> requestBuffer = methodGetRequestBuffer(proxyRequest);
|
||||
requestBuffer.putHeaders(proxyRequest.headers())
|
||||
.sendJson(bodyDecrypt(ctx.getBodyAsString(), appCode), h -> {
|
||||
if (h.succeeded()) {
|
||||
log.info("==========uri:{},response http code:{}", proxyRequest.getURI(),
|
||||
@ -305,32 +305,31 @@ public class ReverseProxy implements HttpProxy {
|
||||
}
|
||||
|
||||
private Future<ProxyResponse> breaker(CircuitBreaker circuitBreaker, ProxyRequest proxyRequest) {
|
||||
SocketAddress socketAddress = ProxyTool.resolveOriginAddress(proxyRequest.proxiedRequest());
|
||||
return Future.future(p -> {
|
||||
circuitBreaker.executeWithFallback(promise -> {
|
||||
mainWebClient.post(socketAddress.port(), socketAddress.host(), proxyRequest.getURI())
|
||||
.putHeaders(proxyRequest.headers()).sendJson(ctx.getBodyAsString(), h -> {
|
||||
if (h.succeeded()) {
|
||||
log.info("==========uri:{},response http code:{}", proxyRequest.getURI(),
|
||||
h.result().statusCode());
|
||||
if (h.result().statusCode() == 200) {
|
||||
// promise.complete();
|
||||
promise.complete("1");
|
||||
// 释放资源
|
||||
proxyRequest.release();
|
||||
JsonObject responseData = h.result().bodyAsJsonObject();
|
||||
log.info("responseData:{}", responseData);
|
||||
ProxyResponse proxyResponse = proxyRequest.response().setStatusCode(200)
|
||||
.putHeader("content-type", "application/json")
|
||||
.setBody(Body.body(responseData.toBuffer()));
|
||||
p.complete(proxyResponse);
|
||||
} else {
|
||||
promise.fail("2");
|
||||
}
|
||||
} else {
|
||||
promise.fail("2");
|
||||
}
|
||||
});
|
||||
HttpRequest<Buffer> requestBuffer = methodGetRequestBuffer(proxyRequest);
|
||||
requestBuffer.putHeaders(proxyRequest.headers()).sendJson(ctx.getBodyAsString(), h -> {
|
||||
if (h.succeeded()) {
|
||||
log.info("==========uri:{},response http code:{}", proxyRequest.getURI(),
|
||||
h.result().statusCode());
|
||||
if (h.result().statusCode() == 200) {
|
||||
// promise.complete();
|
||||
promise.complete("1");
|
||||
// 释放资源
|
||||
proxyRequest.release();
|
||||
JsonObject responseData = h.result().bodyAsJsonObject();
|
||||
log.info("responseData:{}", responseData);
|
||||
ProxyResponse proxyResponse = proxyRequest.response().setStatusCode(200)
|
||||
.putHeader("content-type", "application/json")
|
||||
.setBody(Body.body(responseData.toBuffer()));
|
||||
p.complete(proxyResponse);
|
||||
} else {
|
||||
promise.fail("2");
|
||||
}
|
||||
} else {
|
||||
promise.fail("2");
|
||||
}
|
||||
});
|
||||
}, v -> {
|
||||
// 需要传递当前状态half-open , close, 还是统计失败次数
|
||||
log.info(circuitBreaker.name() + " executed when the circuit is opened:{}", v.getMessage());
|
||||
@ -353,10 +352,10 @@ public class ReverseProxy implements HttpProxy {
|
||||
|
||||
private Future<ProxyResponse> security(CircuitBreaker circuitBreaker, ProxyRequest proxyRequest) {
|
||||
String appCode = proxyRequest.headers().get(AppConfigHandler.getAppCodeHeaderKey());
|
||||
SocketAddress socketAddress = ProxyTool.resolveOriginAddress(proxyRequest.proxiedRequest());
|
||||
return Future.future(p -> {
|
||||
mainWebClient.post(socketAddress.port(), socketAddress.host(), proxyRequest.getURI())
|
||||
.putHeaders(proxyRequest.headers()).sendJson(bodyDecrypt(ctx.getBodyAsString(), appCode), h -> {
|
||||
HttpRequest<Buffer> requestBuffer = methodGetRequestBuffer(proxyRequest);
|
||||
requestBuffer.putHeaders(proxyRequest.headers()).sendJson(bodyDecrypt(ctx.getBodyAsString(), appCode),
|
||||
h -> {
|
||||
if (h.succeeded()) {
|
||||
log.info("==========uri:{},response http code:{}", proxyRequest.getURI(),
|
||||
h.result().statusCode());
|
||||
@ -380,6 +379,30 @@ public class ReverseProxy implements HttpProxy {
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private HttpRequest<Buffer> methodGetRequestBuffer(ProxyRequest proxyRequest){
|
||||
SocketAddress socketAddress = ProxyTool.resolveOriginAddress(proxyRequest.proxiedRequest());
|
||||
HttpRequest<Buffer> requestBuffer = null;
|
||||
switch (proxyRequest.getMethod().name()) {
|
||||
case "PUT":
|
||||
requestBuffer = mainWebClient.put(socketAddress.port(), socketAddress.host(),
|
||||
proxyRequest.getURI());
|
||||
break;
|
||||
case "DELETE":
|
||||
requestBuffer = mainWebClient.delete(socketAddress.port(), socketAddress.host(),
|
||||
proxyRequest.getURI());
|
||||
break;
|
||||
case "GET":
|
||||
requestBuffer = mainWebClient.get(socketAddress.port(), socketAddress.host(),
|
||||
proxyRequest.getURI());
|
||||
break;
|
||||
default:
|
||||
requestBuffer = mainWebClient.post(socketAddress.port(), socketAddress.host(),
|
||||
proxyRequest.getURI());
|
||||
break;
|
||||
}
|
||||
return requestBuffer;
|
||||
}
|
||||
|
||||
private Future<ProxyResponse> sendProxyRequest(ProxyRequest proxyRequest) {
|
||||
// 判断<br/>
|
||||
|
Loading…
x
Reference in New Issue
Block a user