This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new a86eac9f8 [ISSUE #3731]Invoke method(s) only conditionally.
a86eac9f8 is described below
commit a86eac9f8a56d07a247a3f8d7f1c2b0faf95e57a
Author: pandaapo <[email protected]>
AuthorDate: Wed Apr 19 11:09:03 2023 +0800
[ISSUE #3731]Invoke method(s) only conditionally.
* Fix issue: invoke method(s) only conditionally.
* Invoke method(s) only conditionally.
---
.../core/protocol/http/processor/HandlerService.java | 7 ++++---
.../http/processor/LocalSubscribeEventProcessor.java | 20 +++++++++-----------
.../processor/LocalUnSubscribeEventProcessor.java | 6 +++---
.../processor/RemoteSubscribeEventProcessor.java | 12 +++++-------
.../processor/RemoteUnSubscribeEventProcessor.java | 6 +++---
5 files changed, 24 insertions(+), 27 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
index f751cb9cf..f4c92bb64 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.runtime.boot.HTTPTrace;
import org.apache.eventmesh.runtime.boot.HTTPTrace.TraceOperation;
import org.apache.eventmesh.runtime.common.EventMeshTrace;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
import org.apache.eventmesh.runtime.util.HttpResponseUtils;
@@ -68,7 +69,7 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class HandlerService {
- private final Logger httpLogger = LoggerFactory.getLogger("http");
+ private final Logger httpLogger =
LoggerFactory.getLogger(EventMeshConstants.PROTOCOL_HTTP);
private final Map<String, ProcessorWrapper> httpProcessorMap = new
ConcurrentHashMap<>();
@@ -336,8 +337,8 @@ public class HandlerService {
Map<String, Object> traceMap) {
this.traceMap = traceMap;
try {
- responseBodyMap.put("retCode", retCode.getRetCode());
- responseBodyMap.put("retMsg", retCode.getErrMsg());
+ responseBodyMap.put(EventMeshConstants.RET_CODE,
retCode.getRetCode());
+ responseBodyMap.put(EventMeshConstants.RET_MSG,
retCode.getErrMsg());
HttpEventWrapper responseWrapper =
asyncContext.getRequest().createHttpResponse(responseHeaderMap,
responseBodyMap);
asyncContext.onComplete(responseWrapper);
this.exception = new RuntimeException(retCode.getErrMsg());
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
index b761a2873..b6e5f7f44 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
@@ -67,15 +67,14 @@ public class LocalSubscribeEventProcessor extends
AbstractEventProcessor {
final Channel channel = handlerSpecific.getCtx().channel();
final HttpEventWrapper requestWrapper =
handlerSpecific.getAsyncContext().getRequest();
String localAddress = IPUtils.getLocalAddress();
+ String remoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
if (log.isInfoEnabled()) {
log.info("uri={}|{}|client2eventMesh|from={}|to={}",
requestWrapper.getRequestURI(),
- EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(channel),
- localAddress);
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
}
// user request header
- requestWrapper.getHeaderMap().put(ProtocolKey.ClientInstanceKey.IP,
- RemotingHelper.parseChannelRemoteAddr(channel));
+ requestWrapper.getHeaderMap().put(ProtocolKey.ClientInstanceKey.IP,
remoteAddr);
// build sys header
requestWrapper.buildSysHeaderForClient();
@@ -118,11 +117,10 @@ public class LocalSubscribeEventProcessor extends
AbstractEventProcessor {
if
(eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable())
{
for (final SubscriptionItem item : subscriptionList) {
try {
-
this.acl.doAclCheckInHttpReceive(RemotingHelper.parseChannelRemoteAddr(channel),
-
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.USERNAME).toString(),
-
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PASSWD).toString(),
-
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.SYS).toString(),
- item.getTopic(),
+ String user =
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.USERNAME).toString();
+ String pass =
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PASSWD).toString();
+ String subsystem =
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.SYS).toString();
+ this.acl.doAclCheckInHttpReceive(remoteAddr, user, pass,
subsystem, item.getTopic(),
requestWrapper.getRequestURI());
} catch (Exception e) {
if (log.isWarnEnabled()) {
@@ -180,8 +178,8 @@ public class LocalSubscribeEventProcessor extends
AbstractEventProcessor {
// subscription relationship change notification
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().get(consumerGroup));
- responseBodyMap.put("retCode",
EventMeshRetCode.SUCCESS.getRetCode());
- responseBodyMap.put("retMsg",
EventMeshRetCode.SUCCESS.getErrMsg());
+ responseBodyMap.put(EventMeshConstants.RET_CODE,
EventMeshRetCode.SUCCESS.getRetCode());
+ responseBodyMap.put(EventMeshConstants.RET_MSG,
EventMeshRetCode.SUCCESS.getErrMsg());
handlerSpecific.sendResponse(responseHeaderMap,
responseBodyMap);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
index 881b34185..42512167d 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
@@ -76,14 +76,14 @@ public class LocalUnSubscribeEventProcessor extends
AbstractEventProcessor {
final HttpEventWrapper requestWrapper = asyncContext.getRequest();
String localAddress = IPUtils.getLocalAddress();
+ String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
if (log.isInfoEnabled()) {
log.info("uri={}|{}|client2eventMesh|from={}|to={}",
requestWrapper.getRequestURI(),
- EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
localAddress);
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
}
// user request header
- requestWrapper.getHeaderMap().put(ProtocolKey.ClientInstanceKey.IP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
+ requestWrapper.getHeaderMap().put(ProtocolKey.ClientInstanceKey.IP,
remoteAddr);
// build sys header
requestWrapper.buildSysHeaderForClient();
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java
index a04f5c0a2..dc2a9bed8 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java
@@ -77,14 +77,14 @@ public class RemoteSubscribeEventProcessor extends
AbstractEventProcessor {
HttpEventWrapper requestWrapper = asyncContext.getRequest();
String localAddress = IPUtils.getLocalAddress();
+ String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
httpLogger.info("uri={}|{}|client2eventMesh|from={}|to={}",
requestWrapper.getRequestURI(),
- EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress
);
// user request header
Map<String, Object> userRequestHeaderMap =
requestWrapper.getHeaderMap();
- String requestIp =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- userRequestHeaderMap.put(ProtocolKey.ClientInstanceKey.IP, requestIp);
+ userRequestHeaderMap.put(ProtocolKey.ClientInstanceKey.IP, remoteAddr);
// build sys header
requestWrapper.buildSysHeaderForClient();
@@ -131,7 +131,6 @@ public class RemoteSubscribeEventProcessor extends
AbstractEventProcessor {
//do acl check
EventMeshHTTPConfiguration eventMeshHttpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
if (eventMeshHttpConfiguration.isEventMeshServerSecurityEnable()) {
- String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
String user =
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.USERNAME).toString();
String pass =
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PASSWD).toString();
String subsystem =
sysHeaderMap.get(ProtocolKey.ClientInstanceKey.SYS).toString();
@@ -162,8 +161,9 @@ public class RemoteSubscribeEventProcessor extends
AbstractEventProcessor {
return;
}
+ CloseableHttpClient closeableHttpClient =
eventMeshHTTPServer.getHttpClientPool().getClient();
// obtain webhook delivery agreement for Abuse Protection
- boolean isWebhookAllowed =
WebhookUtil.obtainDeliveryAgreement(eventMeshHTTPServer.getHttpClientPool().getClient(),
+ boolean isWebhookAllowed =
WebhookUtil.obtainDeliveryAgreement(closeableHttpClient,
url, eventMeshHttpConfiguration.getEventMeshWebhookOrigin());
if (!isWebhookAllowed) {
@@ -192,8 +192,6 @@ public class RemoteSubscribeEventProcessor extends
AbstractEventProcessor {
targetMesh = meshAddress;
}
- CloseableHttpClient closeableHttpClient =
eventMeshHTTPServer.getHttpClientPool().getClient();
-
String remoteResult = post(closeableHttpClient, targetMesh,
builderRemoteHeaderMap(localAddress), remoteBodyMap,
response -> EntityUtils.toString(response.getEntity(),
Constants.DEFAULT_CHARSET));
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
index 5090acb80..31dbb1912 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
@@ -75,14 +75,14 @@ public class RemoteUnSubscribeEventProcessor extends
AbstractEventProcessor {
HttpEventWrapper requestWrapper = asyncContext.getRequest();
String localAddress = IPUtils.getLocalAddress();
+ String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
httpLogger.info("uri={}|{}|client2eventMesh|from={}|to={}",
requestWrapper.getRequestURI(),
- EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress
);
// user request header
Map<String, Object> userRequestHeaderMap =
requestWrapper.getHeaderMap();
- String requestIp =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- userRequestHeaderMap.put(ProtocolKey.ClientInstanceKey.IP, requestIp);
+ userRequestHeaderMap.put(ProtocolKey.ClientInstanceKey.IP, remoteAddr);
// build sys header
requestWrapper.buildSysHeaderForClient();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
