You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by ch...@apache.org on 2022/07/27 10:04:29 UTC
[incubator-eventmesh] branch webhook updated: update http processors supported trace annotation
This is an automated email from the ASF dual-hosted git repository.
chenguangsheng pushed a commit to branch webhook
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/webhook by this push:
new 3d1898b0 update http processors supported trace annotation
new 295b7e24 Merge pull request #1045 from xwm1992/webhook
3d1898b0 is described below
commit 3d1898b0815b80cd80573e69df37c1bc3b1a76c3
Author: xwm1992 <mi...@126.com>
AuthorDate: Wed Jul 27 17:53:02 2022 +0800
update http processors supported trace annotation
---
.../runtime/boot/EventMeshHTTPServer.java | 9 +-
.../apache/eventmesh/runtime/boot/HTTPTrace.java | 8 +-
.../eventmesh/runtime/common/EventMeshTrace.java | 17 ++++
.../protocol/http/processor/HandlerService.java | 8 +-
.../processor/LocalSubscribeEventProcessor.java | 97 +++++++-------------
.../processor/LocalUnSubscribeEventProcessor.java | 83 +++++++----------
.../processor/RemoteSubscribeEventProcessor.java | 101 +++++++--------------
.../processor/RemoteUnSubscribeEventProcessor.java | 73 ++++++---------
.../http/processor/SendAsyncEventProcessor.java | 2 +
.../processor/SendAsyncRemoteEventProcessor.java | 2 +
.../protocol/http/processor/WebHookProcessor.java | 2 +
.../http/processor/inf/AbstractEventProcessor.java | 2 +-
12 files changed, 162 insertions(+), 242 deletions(-)
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
index a5a1c602..1b28a4bb 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
@@ -22,7 +22,6 @@ import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
import org.apache.eventmesh.common.ThreadPoolFactory;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
-import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.metrics.api.MetricsPluginFactory;
@@ -369,19 +368,19 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
registerProcessor(RequestCode.SUBSCRIBE.getRequestCode(), subscribeProcessor, clientManageExecutor);
LocalSubscribeEventProcessor localSubscribeEventProcessor = new LocalSubscribeEventProcessor(this);
- registerProcessor(RequestURI.SUBSCRIBE_LOCAL.getRequestURI(), localSubscribeEventProcessor, clientManageExecutor);
+ handlerService.register(localSubscribeEventProcessor, clientManageExecutor);
RemoteSubscribeEventProcessor remoteSubscribeEventProcessor = new RemoteSubscribeEventProcessor(this);
- registerProcessor(RequestURI.SUBSCRIBE_REMOTE.getRequestURI(), remoteSubscribeEventProcessor, clientManageExecutor);
+ handlerService.register(remoteSubscribeEventProcessor, clientManageExecutor);
UnSubscribeProcessor unSubscribeProcessor = new UnSubscribeProcessor(this);
registerProcessor(RequestCode.UNSUBSCRIBE.getRequestCode(), unSubscribeProcessor, clientManageExecutor);
LocalUnSubscribeEventProcessor localUnSubscribeEventProcessor = new LocalUnSubscribeEventProcessor(this);
- registerProcessor(RequestURI.UNSUBSCRIBE_LOCAL.getRequestURI(), localUnSubscribeEventProcessor, clientManageExecutor);
+ handlerService.register(localUnSubscribeEventProcessor, clientManageExecutor);
RemoteUnSubscribeEventProcessor remoteUnSubscribeEventProcessor = new RemoteUnSubscribeEventProcessor(this);
- registerProcessor(RequestURI.UNSUBSCRIBE_REMOTE.getRequestURI(), remoteUnSubscribeEventProcessor, clientManageExecutor);
+ handlerService.register(remoteUnSubscribeEventProcessor, clientManageExecutor);
ReplyMessageProcessor replyMessageProcessor = new ReplyMessageProcessor(this);
registerProcessor(RequestCode.REPLY_MESSAGE.getRequestCode(), replyMessageProcessor, replyMsgExecutor);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPTrace.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPTrace.java
index ee340235..afc1c519 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPTrace.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPTrace.java
@@ -45,12 +45,12 @@ public class HTTPTrace {
this.useTrace = useTrace;
}
- public TraceOperation getTraceOperation(HttpRequest httpRequest, Channel channel) {
+ public TraceOperation getTraceOperation(HttpRequest httpRequest, Channel channel, boolean traceEnabled) {
final Map<String, Object> headerMap = parseHttpHeader(httpRequest);
Span span = TraceUtils.prepareServerSpan(headerMap, EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- return new TraceOperation(span, null);
+ return new TraceOperation(span, null, traceEnabled);
}
private Map<String, Object> parseHttpHeader(HttpRequest fullReq) {
@@ -74,6 +74,8 @@ public class HTTPTrace {
private TraceOperation childTraceOperation;
+ private boolean traceEnabled;
+
public void endTrace(CloudEvent ce) {
if (!HTTPTrace.this.useTrace) {
return;
@@ -119,7 +121,7 @@ public class HTTPTrace {
}
public TraceOperation createClientTraceOperation(Map<String, Object> map, String spanName, boolean isSpanFinishInOtherThread) {
- TraceOperation traceOperation = new TraceOperation(TraceUtils.prepareClientSpan(map, spanName, isSpanFinishInOtherThread), null);
+ TraceOperation traceOperation = new TraceOperation(TraceUtils.prepareClientSpan(map, spanName, isSpanFinishInOtherThread), null, this.traceEnabled);
this.setChildTraceOperation(traceOperation);
return traceOperation;
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/common/EventMeshTrace.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/common/EventMeshTrace.java
new file mode 100644
index 00000000..27a43a9f
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/common/EventMeshTrace.java
@@ -0,0 +1,17 @@
+package org.apache.eventmesh.runtime.common;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE})
+public @interface EventMeshTrace {
+ /**
+ * If true, enable the trace
+ */
+ boolean isEnable() default false;
+}
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 3b19d5f7..e5bda60a 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
@@ -22,6 +22,7 @@ import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
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.core.protocol.http.async.AsyncContext;
import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
import org.apache.eventmesh.runtime.trace.TraceUtils;
@@ -105,6 +106,7 @@ public class HandlerService {
processorWrapper.async = (AsyncHttpProcessor) httpProcessor;
}
processorWrapper.httpProcessor = httpProcessor;
+ processorWrapper.traceEnabled = httpProcessor.getClass().getAnnotation(EventMeshTrace.class).isEnable();
httpProcessorMap.put(path, processorWrapper);
httpServerLogger.info("path is {} processor name is {}", path, httpProcessor.getClass().getSimpleName());
}
@@ -128,14 +130,12 @@ public class HandlerService {
*/
public void handler(ChannelHandlerContext ctx, HttpRequest httpRequest, ThreadPoolExecutor asyncContextCompleteHandler) {
-
- TraceOperation traceOperation = httpTrace.getTraceOperation(httpRequest, ctx.channel());
-
ProcessorWrapper processorWrapper = getProcessorWrapper(httpRequest);
if (Objects.isNull(processorWrapper)) {
this.sendResponse(ctx, httpRequest, HttpResponseUtils.createNotFound());
return;
}
+ TraceOperation traceOperation = httpTrace.getTraceOperation(httpRequest, ctx.channel(), processorWrapper.traceEnabled);
try {
HandlerSpecific handlerSpecific = new HandlerSpecific();
handlerSpecific.request = httpRequest;
@@ -363,6 +363,8 @@ public class HandlerService {
private HttpProcessor httpProcessor;
private AsyncHttpProcessor async;
+
+ private boolean traceEnabled;
}
}
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 446e4d11..4c1810f2 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
@@ -21,18 +21,18 @@ import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.common.EventMeshTrace;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicConf;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
-import org.apache.eventmesh.runtime.core.protocol.http.async.CompleteHandler;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.AbstractEventProcessor;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.Client;
-import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.eventmesh.runtime.util.WebhookUtil;
@@ -51,23 +51,28 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
import com.fasterxml.jackson.core.type.TypeReference;
-public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
+
+@EventMeshTrace(isEnable = false)
+public class LocalSubscribeEventProcessor extends AbstractEventProcessor implements AsyncHttpProcessor {
public Logger httpLogger = LoggerFactory.getLogger("http");
public Logger aclLogger = LoggerFactory.getLogger("acl");
-
public LocalSubscribeEventProcessor(EventMeshHTTPServer eventMeshHTTPServer) {
super(eventMeshHTTPServer);
}
@Override
- public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpEventWrapper> asyncContext)
- throws Exception {
+ public void handler(HandlerService.HandlerSpecific handlerSpecific, HttpRequest httpRequest) throws Exception {
+
+ AsyncContext<HttpEventWrapper> asyncContext = handlerSpecific.getAsyncContext();
+
+ ChannelHandlerContext ctx = handlerSpecific.getCtx();
HttpEventWrapper requestWrapper = asyncContext.getRequest();
@@ -98,34 +103,28 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
Map<String, Object> sysHeaderMap = requestWrapper.getSysHeaderMap();
+ Map<String, Object> responseBodyMap = new HashMap<>();
+
//validate header
if (StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.IDC).toString())
|| StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString())
|| !StringUtils.isNumeric(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString())
|| StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.SYS).toString())) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg());
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
-
//validate body
byte[] requestBody = requestWrapper.getBody();
Map<String, Object> requestBodyMap = JsonUtils.deserialize(new String(requestBody), new TypeReference<HashMap<String, Object>>() {
});
-
if (requestBodyMap.get("url") == null || requestBodyMap.get("topic") == null || requestBodyMap.get("consumerGroup") == null) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg());
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -133,7 +132,6 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
String consumerGroup = requestBodyMap.get("consumerGroup").toString();
String topic = JsonUtils.serialize(requestBodyMap.get("topic"));
-
// SubscriptionItem
List<SubscriptionItem> subscriptionList = JsonUtils.deserialize(topic, new TypeReference<List<SubscriptionItem>>() {
});
@@ -149,13 +147,9 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
Acl.doAclCheckInHttpReceive(remoteAddr, user, pass, subsystem, item.getTopic(),
requestWrapper.getRequestURI());
} catch (Exception e) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_ACL_ERR.getErrMsg());
-
- responseWrapper = asyncContext.getRequest().createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
aclLogger.warn("CLIENT HAS NO PERMISSION,SubscribeProcessor subscribe failed", e);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_ACL_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
}
@@ -166,20 +160,14 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
if (!IPUtils.isValidDomainOrIp(url, eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIpv4BlackList,
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIpv6BlackList)) {
httpLogger.error("subscriber url {} is not valid", url);
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " invalid URL: " + url);
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
} catch (Exception e) {
httpLogger.error("subscriber url {} is not valid, error {}", url, e.getMessage());
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " invalid URL: " + url);
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -189,11 +177,8 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
if (!isWebhookAllowed) {
httpLogger.error("subscriber url {} is not allowed by the target system", url);
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + "unauthorized webhook URL: " + url);
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -274,45 +259,29 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
eventMeshHTTPServer.localConsumerGroupMapping.get(consumerGroup));
- final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper -> {
- try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpEventWrapper);
- }
- eventMeshHTTPServer.sendResponse(ctx, httpEventWrapper.httpResponse());
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
- System.currentTimeMillis() - requestWrapper.getReqTime());
- } catch (Exception ex) {
- // ignore
- }
- };
+ responseBodyMap.put("retCode", EventMeshRetCode.SUCCESS.getRetCode());
+ responseBodyMap.put("retMsg", EventMeshRetCode.SUCCESS.getErrMsg());
+
+ handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
- responseWrapper = requestWrapper.createHttpResponse(EventMeshRetCode.SUCCESS);
- asyncContext.onComplete(responseWrapper, handler);
} catch (Exception e) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2));
- HttpEventWrapper err = asyncContext.getRequest().createHttpResponse(
- responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(err);
long endTime = System.currentTimeMillis();
httpLogger.error(
"message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}"
+ "|bizSeqNo={}|uniqueId={}", endTime - startTime,
JsonUtils.serialize(subscriptionList), url, e);
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR, responseHeaderMap, responseBodyMap, null);
}
// Update service metadata
updateMetadata();
}
+
}
@Override
- public boolean rejectRequest() {
- return false;
+ public String[] paths() {
+ return new String[] {RequestURI.SUBSCRIBE_LOCAL.getRequestURI()};
}
private void registerClient(HttpEventWrapper requestWrapper, String consumerGroup,
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 8b3c7b9a..3f9c858d 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
@@ -20,17 +20,17 @@ package org.apache.eventmesh.runtime.core.protocol.http.processor;
import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.common.EventMeshTrace;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicConf;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
-import org.apache.eventmesh.runtime.core.protocol.http.async.CompleteHandler;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.AbstractEventProcessor;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.Client;
-import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.commons.lang3.StringUtils;
@@ -48,10 +48,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
import com.fasterxml.jackson.core.type.TypeReference;
-public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
+@EventMeshTrace(isEnable = false)
+public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor implements AsyncHttpProcessor {
public Logger httpLogger = LoggerFactory.getLogger("http");
@@ -60,8 +62,11 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
}
@Override
- public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpEventWrapper> asyncContext)
- throws Exception {
+ public void handler(HandlerService.HandlerSpecific handlerSpecific, HttpRequest httpRequest) throws Exception {
+
+ AsyncContext<HttpEventWrapper> asyncContext = handlerSpecific.getAsyncContext();
+
+ ChannelHandlerContext ctx = handlerSpecific.getCtx();
HttpEventWrapper requestWrapper = asyncContext.getRequest();
@@ -92,17 +97,16 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
Map<String, Object> sysHeaderMap = requestWrapper.getSysHeaderMap();
+ Map<String, Object> responseBodyMap = new HashMap<>();
+
//validate header
if (StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.IDC).toString())
|| StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString())
|| !StringUtils.isNumeric(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString())
|| StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.SYS).toString())) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg());
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -113,11 +117,8 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
});
if (requestBodyMap.get("url") == null || requestBodyMap.get("topic") == null || requestBodyMap.get("consumerGroup") == null) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg());
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -135,19 +136,6 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
String ip = sysHeaderMap.get(ProtocolKey.ClientInstanceKey.IP).toString();
String pid = sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString();
- final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper -> {
- try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpEventWrapper);
- }
- eventMeshHTTPServer.sendResponse(ctx, httpEventWrapper.httpResponse());
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
- System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
- } catch (Exception ex) {
- // ignore
- }
- };
-
synchronized (eventMeshHTTPServer.localClientInfoMapping) {
boolean isChange = true;
@@ -219,33 +207,29 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
eventMeshHTTPServer.localConsumerGroupMapping.get(consumerGroup));
- responseWrapper =
- asyncContext.getRequest().createHttpResponse(EventMeshRetCode.SUCCESS);
- asyncContext.onComplete(responseWrapper, handler);
+ responseBodyMap.put("retCode", EventMeshRetCode.SUCCESS.getRetCode());
+ responseBodyMap.put("retMsg", EventMeshRetCode.SUCCESS.getErrMsg());
+
+ handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
} catch (Exception e) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2));
- HttpEventWrapper err = asyncContext.getRequest().createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(err);
long endTime = System.currentTimeMillis();
httpLogger.error(
"message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
+ "|topic={}|bizSeqNo={}|uniqueId={}", endTime - startTime,
JsonUtils.serialize(unSubTopicList), unSubscribeUrl, e);
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
- eventMeshHTTPServer.metrics.getSummaryMetrics()
- .recordSendMsgCost(endTime - startTime);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR, responseHeaderMap,
+ responseBodyMap, null);
}
} else {
//remove
try {
eventMeshHTTPServer.getConsumerManager()
.notifyConsumerManager(consumerGroup, null);
- responseWrapper =
- asyncContext.getRequest().createHttpResponse(EventMeshRetCode.SUCCESS);
- asyncContext.onComplete(responseWrapper, handler);
+ responseBodyMap.put("retCode", EventMeshRetCode.SUCCESS.getRetCode());
+ responseBodyMap.put("retMsg", EventMeshRetCode.SUCCESS.getErrMsg());
+
+ handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
// clean ClientInfo
eventMeshHTTPServer.localClientInfoMapping.keySet()
.removeIf(s -> StringUtils.contains(s, consumerGroup));
@@ -253,19 +237,13 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
eventMeshHTTPServer.localConsumerGroupMapping.keySet()
.removeIf(s -> StringUtils.equals(consumerGroup, s));
} catch (Exception e) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2));
- HttpEventWrapper err = asyncContext.getRequest().createHttpResponse(
- responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(err);
long endTime = System.currentTimeMillis();
httpLogger.error(
"message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
+ "|topic={}|bizSeqNo={}|uniqueId={}", endTime - startTime,
JsonUtils.serialize(unSubTopicList), unSubscribeUrl, e);
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR, responseHeaderMap,
+ responseBodyMap, null);
}
}
@@ -275,10 +253,11 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
}
@Override
- public boolean rejectRequest() {
- return false;
+ public String[] paths() {
+ return new String[] {RequestURI.UNSUBSCRIBE_LOCAL.getRequestURI()};
}
+
private void registerClient(HttpEventWrapper requestWrapper,
String consumerGroup,
List<String> topicList, String url) {
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 21a66a5b..f1a9d9c3 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
@@ -28,11 +28,10 @@ import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+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.core.protocol.http.async.CompleteHandler;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.AbstractEventProcessor;
-import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.eventmesh.runtime.util.WebhookUtil;
@@ -56,11 +55,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.base.Preconditions;
-public class RemoteSubscribeEventProcessor extends AbstractEventProcessor {
+@EventMeshTrace(isEnable = false)
+public class RemoteSubscribeEventProcessor extends AbstractEventProcessor implements AsyncHttpProcessor {
public Logger httpLogger = LoggerFactory.getLogger("http");
@@ -72,8 +73,11 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor {
}
@Override
- public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpEventWrapper> asyncContext)
- throws Exception {
+ public void handler(HandlerService.HandlerSpecific handlerSpecific, HttpRequest httpRequest) throws Exception {
+
+ AsyncContext<HttpEventWrapper> asyncContext = handlerSpecific.getAsyncContext();
+
+ ChannelHandlerContext ctx = handlerSpecific.getCtx();
HttpEventWrapper requestWrapper = asyncContext.getRequest();
@@ -101,17 +105,16 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor {
Map<String, Object> sysHeaderMap = requestWrapper.getSysHeaderMap();
+ Map<String, Object> responseBodyMap = new HashMap<>();
+
//validate header
if (StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.IDC).toString())
|| StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString())
|| !StringUtils.isNumeric(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString())
|| StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.SYS).toString())) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg());
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -124,11 +127,8 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor {
if (requestBodyMap.get("url") == null || requestBodyMap.get("topic") == null || requestBodyMap.get("consumerGroup") == null) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg());
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -152,13 +152,10 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor {
Acl.doAclCheckInHttpReceive(remoteAddr, user, pass, subsystem, item.getTopic(),
requestWrapper.getRequestURI());
} catch (Exception e) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_ACL_ERR.getErrMsg());
-
- responseWrapper = asyncContext.getRequest().createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
aclLogger.warn("CLIENT HAS NO PERMISSION,SubscribeProcessor subscribe failed", e);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_ACL_ERR, responseHeaderMap,
+ responseBodyMap, null);
+
return;
}
}
@@ -169,20 +166,14 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor {
if (!IPUtils.isValidDomainOrIp(url, eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIpv4BlackList,
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIpv6BlackList)) {
httpLogger.error("subscriber url {} is not valid", url);
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " invalid URL: " + url);
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
} catch (Exception e) {
httpLogger.error("subscriber url {} is not valid, error {}", url, e.getMessage());
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " invalid URL: " + url);
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -192,11 +183,8 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor {
if (!isWebhookAllowed) {
httpLogger.error("subscriber url {} is not allowed by the target system", url);
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + "unauthorized webhook URL: " + url);
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -248,51 +236,30 @@ public class RemoteSubscribeEventProcessor extends AbstractEventProcessor {
Map<String, String> remoteResultMap = JsonUtils.deserialize(remoteResult, new TypeReference<Map<String, String>>() {
});
- final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper -> {
- try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpEventWrapper);
- }
- eventMeshHTTPServer.sendResponse(ctx, httpEventWrapper.httpResponse());
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
- System.currentTimeMillis() - requestWrapper.getReqTime());
- } catch (Exception ex) {
- // ignore
- }
- };
-
if (String.valueOf(EventMeshRetCode.SUCCESS.getRetCode()).equals(remoteResultMap.get("retCode"))) {
- responseWrapper = requestWrapper.createHttpResponse(EventMeshRetCode.SUCCESS);
- asyncContext.onComplete(responseWrapper, handler);
+ responseBodyMap.put("retCode", EventMeshRetCode.SUCCESS.getRetCode());
+ responseBodyMap.put("retMsg", EventMeshRetCode.SUCCESS.getErrMsg());
+
+ handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
} else {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg());
- HttpEventWrapper err = asyncContext.getRequest().createHttpResponse(
- responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(err);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR, responseHeaderMap,
+ responseBodyMap, null);
}
} catch (Exception e) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2));
- HttpEventWrapper err = asyncContext.getRequest().createHttpResponse(
- responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(err);
long endTime = System.currentTimeMillis();
httpLogger.error(
"message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}"
+ "|bizSeqNo={}|uniqueId={}", endTime - startTime,
JsonUtils.serialize(subscriptionList), url, e);
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR, responseHeaderMap,
+ responseBodyMap, null);
}
}
@Override
- public boolean rejectRequest() {
- return false;
+ public String[] paths() {
+ return new String[] {RequestURI.SUBSCRIBE_REMOTE.getRequestURI()};
}
public static String post(CloseableHttpClient client, String uri,
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 e7950bdd..9c3d9ff7 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
@@ -27,11 +27,10 @@ import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+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.core.protocol.http.async.CompleteHandler;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.AbstractEventProcessor;
-import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.commons.collections4.MapUtils;
@@ -55,11 +54,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.base.Preconditions;
-public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor {
+@EventMeshTrace(isEnable = false)
+public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor implements AsyncHttpProcessor {
public Logger httpLogger = LoggerFactory.getLogger("http");
@@ -70,8 +71,11 @@ public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor {
}
@Override
- public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpEventWrapper> asyncContext)
- throws Exception {
+ public void handler(HandlerService.HandlerSpecific handlerSpecific, HttpRequest httpRequest) throws Exception {
+
+ AsyncContext<HttpEventWrapper> asyncContext = handlerSpecific.getAsyncContext();
+
+ ChannelHandlerContext ctx = handlerSpecific.getCtx();
HttpEventWrapper requestWrapper = asyncContext.getRequest();
@@ -99,17 +103,15 @@ public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor {
Map<String, Object> sysHeaderMap = requestWrapper.getSysHeaderMap();
+ Map<String, Object> responseBodyMap = new HashMap<>();
+
//validate header
if (StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.IDC).toString())
|| StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString())
|| !StringUtils.isNumeric(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.PID).toString())
|| StringUtils.isBlank(sysHeaderMap.get(ProtocolKey.ClientInstanceKey.SYS).toString())) {
-
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg());
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -122,11 +124,8 @@ public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor {
if (requestBodyMap.get("url") == null || requestBodyMap.get("topic") == null || requestBodyMap.get("consumerGroup") == null) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg());
- responseWrapper = requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(responseWrapper);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
+ responseBodyMap, null);
return;
}
@@ -189,51 +188,31 @@ public class RemoteUnSubscribeEventProcessor extends AbstractEventProcessor {
Map<String, String> remoteResultMap = JsonUtils.deserialize(remoteResult, new TypeReference<Map<String, String>>() {
});
- final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper -> {
- try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpEventWrapper);
- }
- eventMeshHTTPServer.sendResponse(ctx, httpEventWrapper.httpResponse());
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
- System.currentTimeMillis() - requestWrapper.getReqTime());
- } catch (Exception ex) {
- // ignore
- }
- };
-
if (String.valueOf(EventMeshRetCode.SUCCESS.getRetCode()).equals(remoteResultMap.get("retCode"))) {
- responseWrapper = requestWrapper.createHttpResponse(EventMeshRetCode.SUCCESS);
- asyncContext.onComplete(responseWrapper, handler);
+ responseBodyMap.put("retCode", EventMeshRetCode.SUCCESS.getRetCode());
+ responseBodyMap.put("retMsg", EventMeshRetCode.SUCCESS.getErrMsg());
+
+ handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
} else {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg());
- HttpEventWrapper err = asyncContext.getRequest().createHttpResponse(
- responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(err);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR, responseHeaderMap,
+ responseBodyMap, null);
}
} catch (Exception e) {
- Map<String, Object> responseBodyMap = new HashMap<>();
- responseBodyMap.put("retCode", EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode());
- responseBodyMap.put("retMsg", EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2));
- HttpEventWrapper err = asyncContext.getRequest().createHttpResponse(
- responseHeaderMap, responseBodyMap);
- asyncContext.onComplete(err);
long endTime = System.currentTimeMillis();
httpLogger.error(
"message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}"
+ "|bizSeqNo={}|uniqueId={}", endTime - startTime,
topic, unSubscribeUrl, e);
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
- eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
+ handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR, responseHeaderMap,
+ responseBodyMap, null);
}
+
}
@Override
- public boolean rejectRequest() {
- return false;
+ public String[] paths() {
+ return new String[] {RequestURI.UNSUBSCRIBE_REMOTE.getRequestURI()};
}
public static String post(CloseableHttpClient client, String uri,
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java
index 05302eb1..0634f1e7 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java
@@ -32,6 +32,7 @@ import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+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.core.protocol.http.producer.EventMeshProducer;
@@ -56,6 +57,7 @@ import io.cloudevents.core.builder.CloudEventBuilder;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.HttpRequest;
+@EventMeshTrace(isEnable = true)
public class SendAsyncEventProcessor implements AsyncHttpProcessor {
public Logger messageLogger = LoggerFactory.getLogger("message");
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
index 6e40fd27..d0c4982a 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
@@ -33,6 +33,7 @@ import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+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.core.protocol.http.producer.EventMeshProducer;
@@ -60,6 +61,7 @@ import io.netty.handler.codec.http.HttpRequest;
import com.fasterxml.jackson.core.type.TypeReference;
+@EventMeshTrace(isEnable = true)
public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
public Logger messageLogger = LoggerFactory.getLogger("message");
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/WebHookProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/WebHookProcessor.java
index 8b907b6f..2bc878fa 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/WebHookProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/WebHookProcessor.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
+import org.apache.eventmesh.runtime.common.EventMeshTrace;
import org.apache.eventmesh.runtime.util.HttpResponseUtils;
import org.apache.eventmesh.webhook.receive.WebHookController;
@@ -29,6 +30,7 @@ import io.netty.handler.codec.http.HttpResponse;
import lombok.Setter;
+@EventMeshTrace(isEnable = true)
public class WebHookProcessor implements HttpProcessor {
@Setter
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
index 2af9bc0b..2fe7f25b 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
@@ -42,7 +42,7 @@ import org.slf4j.LoggerFactory;
/**
* EventProcessor
*/
-public abstract class AbstractEventProcessor implements EventProcessor {
+public class AbstractEventProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger("AbstractEventProcessor");
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org