This is an automated email from the ASF dual-hosted git repository.
xiatian 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 9a3912a35 [ISSUE #4731] HttpRequestProcessor enhancement (#4732)
9a3912a35 is described below
commit 9a3912a35b1dab6d77d6218fac926ac88c28e2bf
Author: Karson <[email protected]>
AuthorDate: Sun Feb 18 17:05:15 2024 +0800
[ISSUE #4731] HttpRequestProcessor enhancement (#4732)
* resolve conflicts.
* resolve
---
.../eventmesh/runtime/boot/AbstractHTTPServer.java | 84 ++++++++++++----------
.../runtime/boot/EventMeshHTTPServer.java | 57 ++++++---------
...ssor.java => AbstractHttpRequestProcessor.java} | 7 +-
.../http/processor/AdminMetricsProcessor.java | 11 ++-
.../http/processor/AdminShutdownProcessor.java | 12 +++-
.../http/processor/BatchSendMessageProcessor.java | 9 ++-
.../processor/BatchSendMessageV2Processor.java | 9 ++-
.../http/processor/CreateTopicProcessor.java | 6 ++
.../http/processor/DeleteTopicProcessor.java | 6 ++
.../protocol/http/processor/HandlerService.java | 17 +++--
.../http/processor/HeartBeatProcessor.java | 9 ++-
.../protocol/http/processor/HttpProcessor.java | 9 +++
.../processor/LocalSubscribeEventProcessor.java | 8 ++-
.../processor/LocalUnSubscribeEventProcessor.java | 11 ++-
.../http/processor/QuerySubscriptionProcessor.java | 6 ++
.../processor/RemoteSubscribeEventProcessor.java | 8 ++-
.../processor/RemoteUnSubscribeEventProcessor.java | 8 ++-
.../http/processor/ReplyMessageProcessor.java | 9 ++-
.../http/processor/SendAsyncEventProcessor.java | 8 ++-
.../http/processor/SendAsyncMessageProcessor.java | 11 ++-
.../processor/SendAsyncRemoteEventProcessor.java | 8 ++-
.../http/processor/SendSyncMessageProcessor.java | 12 +++-
.../http/processor/ShortHttpProcessor.java | 1 +
.../http/processor/SubscribeProcessor.java | 13 +++-
.../http/processor/UnSubscribeProcessor.java | 82 ++++++++++-----------
.../protocol/http/processor/WebHookProcessor.java | 2 +-
.../http/processor/inf/HttpRequestProcessor.java | 6 ++
27 files changed, 274 insertions(+), 155 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
index 4ff086871..f978f25a8 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
@@ -27,7 +27,6 @@ import
org.apache.eventmesh.common.protocol.http.common.ProtocolVersion;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.protocol.http.header.Header;
import org.apache.eventmesh.common.utils.AssertUtils;
-import org.apache.eventmesh.runtime.common.Pair;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
@@ -117,7 +116,7 @@ public abstract class AbstractHTTPServer extends
AbstractRemotingServer {
/**
* key: request code
*/
- protected final transient Map<String, Pair<HttpRequestProcessor,
ThreadPoolExecutor>> httpRequestProcessorTable =
+ protected final transient Map<String, HttpRequestProcessor>
httpRequestProcessorTable =
new ConcurrentHashMap<>(64);
private HttpConnectionHandler httpConnectionHandler;
@@ -198,11 +197,10 @@ public abstract class AbstractHTTPServer extends
AbstractRemotingServer {
/**
* Registers the processors required by the runtime module
*/
- public void registerProcessor(final Integer requestCode, final
HttpRequestProcessor processor, final ThreadPoolExecutor executor) {
+ public void registerProcessor(final Integer requestCode, final
HttpRequestProcessor processor) {
AssertUtils.notNull(requestCode, "requestCode can't be null");
AssertUtils.notNull(processor, "processor can't be null");
- AssertUtils.notNull(executor, "executor can't be null");
- this.httpRequestProcessorTable.put(requestCode.toString(), new
Pair<>(processor, executor));
+ this.httpRequestProcessorTable.putIfAbsent(requestCode.toString(),
processor);
}
/**
@@ -400,46 +398,54 @@ public abstract class AbstractHTTPServer extends
AbstractRemotingServer {
private void processHttpCommandRequest(final ChannelHandlerContext
ctx, final AsyncContext<HttpCommand> asyncContext) {
final HttpCommand request = asyncContext.getRequest();
- final Pair<HttpRequestProcessor, ThreadPoolExecutor> choosed =
httpRequestProcessorTable.get(request.getRequestCode());
- try {
- choosed.getObject2().submit(() -> {
- try {
- final HttpRequestProcessor processor =
choosed.getObject1();
- if (processor.rejectRequest()) {
- final HttpCommand responseCommand =
-
request.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_REJECT_BY_PROCESSOR_ERROR);
- asyncContext.onComplete(responseCommand);
-
- if (asyncContext.isComplete()) {
- sendResponse(ctx,
responseCommand.httpResponse());
- log.debug("{}", asyncContext.getResponse());
- final Map<String, Object> traceMap =
asyncContext.getRequest().getHeader().toMap();
-
TraceUtils.finishSpanWithException(TraceUtils.prepareServerSpan(traceMap,
-
EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
- false),
- traceMap,
-
EventMeshRetCode.EVENTMESH_REJECT_BY_PROCESSOR_ERROR.getErrMsg(), null);
- }
-
- return;
+ final HttpRequestProcessor choosed =
httpRequestProcessorTable.get(request.getRequestCode());
+ Runnable runnable = () -> {
+ try {
+ final HttpRequestProcessor processor = choosed;
+ if (processor.rejectRequest()) {
+ final HttpCommand responseCommand =
+
request.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_REJECT_BY_PROCESSOR_ERROR);
+ asyncContext.onComplete(responseCommand);
+
+ if (asyncContext.isComplete()) {
+ sendResponse(ctx, responseCommand.httpResponse());
+ log.debug("{}", asyncContext.getResponse());
+ final Map<String, Object> traceMap =
asyncContext.getRequest().getHeader().toMap();
+
TraceUtils.finishSpanWithException(TraceUtils.prepareServerSpan(traceMap,
+
+
EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
+ false),
+ traceMap,
+
EventMeshRetCode.EVENTMESH_REJECT_BY_PROCESSOR_ERROR.getErrMsg(), null);
}
- processor.processRequest(ctx, asyncContext);
- if (!asyncContext.isComplete()) {
- return;
- }
+ return;
+ }
+
+ processor.processRequest(ctx, asyncContext);
+ if (!asyncContext.isComplete()) {
+ return;
+ }
- metrics.getSummaryMetrics()
-
.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
+ log.debug("{}", asyncContext.getResponse());
+ metrics.getSummaryMetrics()
+ .recordHTTPReqResTimeCost(System.currentTimeMillis() -
request.getReqTime());
+ sendResponse(ctx,
asyncContext.getResponse().httpResponse());
- log.debug("{}", asyncContext.getResponse());
+ } catch (Exception e) {
+ log.error("process error", e);
+ }
+ };
- sendResponse(ctx,
asyncContext.getResponse().httpResponse());
+ try {
+ if (Objects.nonNull(choosed.executor())) {
+ choosed.executor().execute(() -> {
+ runnable.run();
+ });
+ } else {
+ runnable.run();
+ }
- } catch (Exception e) {
- log.error("process error", e);
- }
- });
} catch (RejectedExecutionException re) {
asyncContext.onComplete(request.createHttpCommandResponse(EventMeshRetCode.OVERLOAD));
metrics.getSummaryMetrics().recordHTTPDiscard();
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 05e90d482..b43931370 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
@@ -62,7 +62,6 @@ import org.apache.commons.lang3.StringUtils;
import java.util.List;
import java.util.Optional;
-import java.util.concurrent.ThreadPoolExecutor;
import org.assertj.core.util.Lists;
@@ -71,6 +70,7 @@ import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
+
/**
* Add multiple managers to the underlying server
*/
@@ -84,22 +84,16 @@ public class EventMeshHTTPServer extends AbstractHTTPServer
{
private final Acl acl;
private final EventBus eventBus = new EventBus();
-
+ private final transient HTTPClientPool httpClientPool = new
HTTPClientPool(10);
private ConsumerManager consumerManager;
private ProducerManager producerManager;
private SubscriptionManager subscriptionManager;
-
private FilterEngine filterEngine;
-
private TransformerEngine transformerEngine;
-
private HttpRetryer httpRetryer;
-
private transient RateLimiter msgRateLimiter;
private transient RateLimiter batchRateLimiter;
- private final transient HTTPClientPool httpClientPool = new
HTTPClientPool(10);
-
public EventMeshHTTPServer(final EventMeshServer eventMeshServer, final
EventMeshHTTPConfiguration eventMeshHttpConfiguration) {
super(eventMeshHttpConfiguration.getHttpServerPort(),
@@ -239,68 +233,59 @@ public class EventMeshHTTPServer extends
AbstractHTTPServer {
}
private void registerHTTPRequestProcessor() throws Exception {
- HTTPThreadPoolGroup httpThreadPoolGroup =
super.getHttpThreadPoolGroup();
-
- ThreadPoolExecutor batchMsgExecutor =
httpThreadPoolGroup.getBatchMsgExecutor();
final BatchSendMessageProcessor batchSendMessageProcessor = new
BatchSendMessageProcessor(this);
- registerProcessor(RequestCode.MSG_BATCH_SEND.getRequestCode(),
batchSendMessageProcessor, batchMsgExecutor);
+ registerProcessor(RequestCode.MSG_BATCH_SEND.getRequestCode(),
batchSendMessageProcessor);
final BatchSendMessageV2Processor batchSendMessageV2Processor = new
BatchSendMessageV2Processor(this);
- registerProcessor(RequestCode.MSG_BATCH_SEND_V2.getRequestCode(),
batchSendMessageV2Processor,
- batchMsgExecutor);
+ registerProcessor(RequestCode.MSG_BATCH_SEND_V2.getRequestCode(),
batchSendMessageV2Processor);
- ThreadPoolExecutor sendMsgExecutor =
httpThreadPoolGroup.getSendMsgExecutor();
final SendSyncMessageProcessor sendSyncMessageProcessor = new
SendSyncMessageProcessor(this);
- registerProcessor(RequestCode.MSG_SEND_SYNC.getRequestCode(),
sendSyncMessageProcessor, sendMsgExecutor);
+ registerProcessor(RequestCode.MSG_SEND_SYNC.getRequestCode(),
sendSyncMessageProcessor);
final SendAsyncMessageProcessor sendAsyncMessageProcessor = new
SendAsyncMessageProcessor(this);
- registerProcessor(RequestCode.MSG_SEND_ASYNC.getRequestCode(),
sendAsyncMessageProcessor, sendMsgExecutor);
+ registerProcessor(RequestCode.MSG_SEND_ASYNC.getRequestCode(),
sendAsyncMessageProcessor);
final SendAsyncEventProcessor sendAsyncEventProcessor = new
SendAsyncEventProcessor(this);
- this.getHandlerService().register(sendAsyncEventProcessor,
sendMsgExecutor);
+ this.getHandlerService().register(sendAsyncEventProcessor);
- ThreadPoolExecutor remoteMsgExecutor =
httpThreadPoolGroup.getRemoteMsgExecutor();
final SendAsyncRemoteEventProcessor sendAsyncRemoteEventProcessor =
new SendAsyncRemoteEventProcessor(this);
- this.getHandlerService().register(sendAsyncRemoteEventProcessor,
remoteMsgExecutor);
+ this.getHandlerService().register(sendAsyncRemoteEventProcessor);
- ThreadPoolExecutor runtimeAdminExecutor =
httpThreadPoolGroup.getRuntimeAdminExecutor();
final AdminMetricsProcessor adminMetricsProcessor = new
AdminMetricsProcessor(this);
- registerProcessor(RequestCode.ADMIN_METRICS.getRequestCode(),
adminMetricsProcessor, runtimeAdminExecutor);
+ registerProcessor(RequestCode.ADMIN_METRICS.getRequestCode(),
adminMetricsProcessor);
- ThreadPoolExecutor clientManageExecutor =
httpThreadPoolGroup.getClientManageExecutor();
final HeartBeatProcessor heartProcessor = new HeartBeatProcessor(this);
- registerProcessor(RequestCode.HEARTBEAT.getRequestCode(),
heartProcessor, clientManageExecutor);
+ registerProcessor(RequestCode.HEARTBEAT.getRequestCode(),
heartProcessor);
final SubscribeProcessor subscribeProcessor = new
SubscribeProcessor(this);
- registerProcessor(RequestCode.SUBSCRIBE.getRequestCode(),
subscribeProcessor, clientManageExecutor);
+ registerProcessor(RequestCode.SUBSCRIBE.getRequestCode(),
subscribeProcessor);
final LocalSubscribeEventProcessor localSubscribeEventProcessor = new
LocalSubscribeEventProcessor(this);
- this.getHandlerService().register(localSubscribeEventProcessor,
clientManageExecutor);
+ this.getHandlerService().register(localSubscribeEventProcessor);
final RemoteSubscribeEventProcessor remoteSubscribeEventProcessor =
new RemoteSubscribeEventProcessor(this);
- this.getHandlerService().register(remoteSubscribeEventProcessor,
clientManageExecutor);
+ this.getHandlerService().register(remoteSubscribeEventProcessor);
final UnSubscribeProcessor unSubscribeProcessor = new
UnSubscribeProcessor(this);
- registerProcessor(RequestCode.UNSUBSCRIBE.getRequestCode(),
unSubscribeProcessor, clientManageExecutor);
+ registerProcessor(RequestCode.UNSUBSCRIBE.getRequestCode(),
unSubscribeProcessor);
final LocalUnSubscribeEventProcessor localUnSubscribeEventProcessor =
new LocalUnSubscribeEventProcessor(this);
- this.getHandlerService().register(localUnSubscribeEventProcessor,
clientManageExecutor);
+ this.getHandlerService().register(localUnSubscribeEventProcessor);
final RemoteUnSubscribeEventProcessor remoteUnSubscribeEventProcessor
= new RemoteUnSubscribeEventProcessor(this);
- this.getHandlerService().register(remoteUnSubscribeEventProcessor,
clientManageExecutor);
+ this.getHandlerService().register(remoteUnSubscribeEventProcessor);
- ThreadPoolExecutor replyMsgExecutor =
httpThreadPoolGroup.getReplyMsgExecutor();
final ReplyMessageProcessor replyMessageProcessor = new
ReplyMessageProcessor(this);
- registerProcessor(RequestCode.REPLY_MESSAGE.getRequestCode(),
replyMessageProcessor, replyMsgExecutor);
+ registerProcessor(RequestCode.REPLY_MESSAGE.getRequestCode(),
replyMessageProcessor);
final CreateTopicProcessor createTopicProcessor = new
CreateTopicProcessor(this);
- this.getHandlerService().register(createTopicProcessor,
clientManageExecutor);
+ this.getHandlerService().register(createTopicProcessor);
final DeleteTopicProcessor deleteTopicProcessor = new
DeleteTopicProcessor(this);
- this.getHandlerService().register(deleteTopicProcessor,
clientManageExecutor);
+ this.getHandlerService().register(deleteTopicProcessor);
final QuerySubscriptionProcessor querySubscriptionProcessor = new
QuerySubscriptionProcessor(this);
- this.getHandlerService().register(querySubscriptionProcessor,
clientManageExecutor);
+ this.getHandlerService().register(querySubscriptionProcessor);
registerWebhook();
}
@@ -370,4 +355,6 @@ public class EventMeshHTTPServer extends AbstractHTTPServer
{
public HTTPClientPool getHttpClientPool() {
return httpClientPool;
}
+
+
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AbstractHttpRequestProcessor.java
similarity index 83%
copy from
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
copy to
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AbstractHttpRequestProcessor.java
index 0c062f8f3..81a4747aa 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AbstractHttpRequestProcessor.java
@@ -17,9 +17,8 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
-/**
- * short-lived HTTP processor
- */
+import
org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
+
+public abstract class AbstractHttpRequestProcessor implements
HttpRequestProcessor {
-public interface ShortHttpProcessor extends HttpProcessor {
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
index 817d339b6..b3fbf0d6a 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
@@ -20,18 +20,25 @@ package
org.apache.eventmesh.runtime.core.protocol.http.processor;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
-import
org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
+
+import java.util.concurrent.Executor;
import io.netty.channel.ChannelHandlerContext;
import lombok.RequiredArgsConstructor;
+
@RequiredArgsConstructor
-public class AdminMetricsProcessor implements HttpRequestProcessor {
+public class AdminMetricsProcessor extends AbstractHttpRequestProcessor {
private final EventMeshHTTPServer eventMeshHTTPServer;
@Override
public void processRequest(ChannelHandlerContext ctx,
AsyncContext<HttpCommand> asyncContext) throws Exception {
}
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getRuntimeAdminExecutor();
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java
index e9e975130..4688ee106 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java
@@ -24,9 +24,10 @@ import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.runtime.boot.EventMeshServer;
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.processor.inf.HttpRequestProcessor;
import org.apache.eventmesh.runtime.util.RemotingHelper;
+import java.util.concurrent.Executor;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,7 +36,7 @@ import io.netty.channel.ChannelHandlerContext;
import lombok.RequiredArgsConstructor;
@RequiredArgsConstructor
-public class AdminShutdownProcessor implements HttpRequestProcessor {
+public class AdminShutdownProcessor extends AbstractHttpRequestProcessor {
public final Logger cmdLogger =
LoggerFactory.getLogger(EventMeshConstants.CMD);
@@ -54,4 +55,11 @@ public class AdminShutdownProcessor implements
HttpRequestProcessor {
HttpCommand responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand);
}
+
+ @Override
+ public Executor executor() {
+ return (Runnable runnable) -> {
+ runnable.run();
+ };
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
index 5baf65515..35fffd1ce 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
@@ -40,7 +40,6 @@ import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
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.processor.inf.HttpRequestProcessor;
import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
import org.apache.eventmesh.runtime.core.protocol.producer.SendMessageContext;
import org.apache.eventmesh.runtime.util.RemotingHelper;
@@ -53,6 +52,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
@@ -65,7 +65,7 @@ import io.netty.channel.ChannelHandlerContext;
import com.google.common.base.Stopwatch;
-public class BatchSendMessageProcessor implements HttpRequestProcessor {
+public class BatchSendMessageProcessor extends AbstractHttpRequestProcessor {
private static final Logger CMD_LOGGER =
LoggerFactory.getLogger(EventMeshConstants.CMD);
@@ -288,4 +288,9 @@ public class BatchSendMessageProcessor implements
HttpRequestProcessor {
SendMessageBatchResponseBody.class);
return;
}
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getBatchMsgExecutor();
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
index c3602cc6f..323b2462e 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
@@ -39,7 +39,6 @@ import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
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.processor.inf.HttpRequestProcessor;
import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
import org.apache.eventmesh.runtime.core.protocol.producer.SendMessageContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
@@ -49,6 +48,7 @@ import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.Objects;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
@@ -58,7 +58,7 @@ import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.netty.channel.ChannelHandlerContext;
-public class BatchSendMessageV2Processor implements HttpRequestProcessor {
+public class BatchSendMessageV2Processor extends AbstractHttpRequestProcessor {
private static final Logger CMD_LOGGER =
LoggerFactory.getLogger(EventMeshConstants.CMD);
@@ -254,4 +254,9 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
EventMeshRetCode.SUCCESS, null,
SendMessageBatchV2ResponseBody.class);
}
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getBatchMsgExecutor();
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
index 3da0de9e0..3f364e9fa 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
@@ -36,6 +36,7 @@ import org.apache.commons.lang3.StringUtils;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.Executor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -151,4 +152,9 @@ public class CreateTopicProcessor implements
AsyncHttpProcessor {
public String[] paths() {
return new String[] {RequestURI.CREATE_TOPIC.getRequestURI()};
}
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
index 0a4c2cc0f..be426c0fc 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
@@ -38,6 +38,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.Executor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -174,4 +175,9 @@ public class DeleteTopicProcessor implements
AsyncHttpProcessor {
public String[] paths() {
return new String[] {RequestURI.DELETE_TOPIC.getRequestURI()};
}
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
+ }
}
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 7f877aab2..58149fc42 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
@@ -42,6 +42,7 @@ import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import org.slf4j.Logger;
@@ -86,20 +87,26 @@ public class HandlerService {
log.info("HandlerService start ");
}
- public void register(HttpProcessor httpProcessor, ThreadPoolExecutor
threadPoolExecutor) {
+ public void register(HttpProcessor httpProcessor, Executor
threadPoolExecutor) {
for (String path : httpProcessor.paths()) {
this.register(path, httpProcessor, threadPoolExecutor);
}
}
- public void register(String path, HttpProcessor httpProcessor,
ThreadPoolExecutor threadPoolExecutor) {
+ public void register(HttpProcessor httpProcessor) {
+ for (String path : httpProcessor.paths()) {
+ this.register(path, httpProcessor, httpProcessor.executor());
+ }
+ }
+
+ public void register(String path, HttpProcessor httpProcessor, Executor
threadPoolExecutor) {
if (httpProcessorMap.containsKey(path)) {
throw new RuntimeException(String.format("HandlerService path %s
repeat, repeat processor is %s ",
path, httpProcessor.getClass().getSimpleName()));
}
ProcessorWrapper processorWrapper = new ProcessorWrapper();
- processorWrapper.threadPoolExecutor = threadPoolExecutor;
+ processorWrapper.executor = threadPoolExecutor;
if (httpProcessor instanceof AsyncHttpProcessor) {
processorWrapper.async = (AsyncHttpProcessor) httpProcessor;
}
@@ -140,7 +147,7 @@ public class HandlerService {
handlerSpecific.ctx = ctx;
handlerSpecific.traceOperation = traceOperation;
handlerSpecific.asyncContext = new AsyncContext<>(new
HttpEventWrapper(), null, asyncContextCompleteHandler);
- processorWrapper.threadPoolExecutor.execute(handlerSpecific);
+ processorWrapper.executor.execute(handlerSpecific);
} catch (Exception e) {
log.error(e.getMessage(), e);
this.sendResponse(ctx, httpRequest,
HttpResponseUtils.createInternalServerError());
@@ -380,7 +387,7 @@ public class HandlerService {
private static class ProcessorWrapper {
- private ThreadPoolExecutor threadPoolExecutor;
+ private Executor executor;
private HttpProcessor httpProcessor;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
index ed2673fdb..be85d0a87 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
@@ -33,7 +33,6 @@ 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.Client;
-import
org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
@@ -46,13 +45,14 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class HeartBeatProcessor implements HttpRequestProcessor {
+public class HeartBeatProcessor extends AbstractHttpRequestProcessor {
private final transient EventMeshHTTPServer eventMeshHTTPServer;
@@ -184,6 +184,11 @@ public class HeartBeatProcessor implements
HttpRequestProcessor {
}
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
+ }
+
private void supplyClientInfoList(final List<Client> tmpClientList, final
List<Client> localClientList) {
Objects.requireNonNull(tmpClientList, "tmpClientList can not be null");
Objects.requireNonNull(localClientList, "localClientList can not be
null");
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HttpProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HttpProcessor.java
index 9b74e4301..64fd431b7 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HttpProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HttpProcessor.java
@@ -17,6 +17,8 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
+import java.util.concurrent.Executor;
+
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
@@ -28,4 +30,11 @@ public interface HttpProcessor {
String[] paths();
HttpResponse handler(HttpRequest httpRequest);
+
+ /**
+ * @return {@link Executor}
+ */
+ default Executor executor() {
+ return null;
+ }
}
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 b5863e5dd..860e77335 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
@@ -40,6 +40,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.Executor;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.HttpRequest;
@@ -185,7 +186,12 @@ public class LocalSubscribeEventProcessor extends
AbstractEventProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.SUBSCRIBE_LOCAL.getRequestURI()};
+ return new String[] {RequestURI.SUBSCRIBE_LOCAL.getRequestURI()};
+ }
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
}
private ClientInfo getClientInfo(final HttpEventWrapper requestWrapper) {
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 bb9c61ed7..72ad78b68 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
@@ -48,6 +48,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.Executor;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.HttpRequest;
@@ -193,6 +194,7 @@ public class LocalUnSubscribeEventProcessor extends
AbstractEventProcessor {
} catch (Exception e) {
log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
+ "|topic={}|url={}", System.currentTimeMillis() -
startTime, JsonUtils.toJSONString(unSubTopicList), unSubscribeUrl, e);
+
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR,
responseHeaderMap,
responseBodyMap, null);
}
@@ -212,8 +214,10 @@ public class LocalUnSubscribeEventProcessor extends
AbstractEventProcessor {
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().keySet()
.removeIf(s -> StringUtils.equals(consumerGroup, s));
} catch (Exception e) {
+
log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
+ "|topic={}|url={}", System.currentTimeMillis() -
startTime, JsonUtils.toJSONString(unSubTopicList), unSubscribeUrl, e);
+
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR,
responseHeaderMap,
responseBodyMap, null);
}
@@ -226,7 +230,12 @@ public class LocalUnSubscribeEventProcessor extends
AbstractEventProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.UNSUBSCRIBE_LOCAL.getRequestURI()};
+ return new String[] {RequestURI.UNSUBSCRIBE_LOCAL.getRequestURI()};
+ }
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
}
private void registerClient(final HttpEventWrapper requestWrapper, final
String consumerGroup, final List<String> topicList, final String url) {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
index 6ffee13a6..7a8d5c17f 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
@@ -33,6 +33,7 @@ import org.apache.eventmesh.runtime.util.RemotingHelper;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.Executor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -114,4 +115,9 @@ public class QuerySubscriptionProcessor implements
AsyncHttpProcessor {
public String[] paths() {
return new String[] {RequestURI.SUBSCRIPTION_QUERY.getRequestURI()};
}
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
+ }
}
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 43f90ea66..84a5812af 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
@@ -43,6 +43,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.Executor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -217,7 +218,12 @@ public class RemoteSubscribeEventProcessor extends
AbstractEventProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.SUBSCRIBE_REMOTE.getRequestURI()};
+ return new String[] {RequestURI.SUBSCRIBE_REMOTE.getRequestURI()};
+ }
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
}
}
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 edbd9ccfa..fbd1af445 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
@@ -43,6 +43,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.Executor;
import java.util.stream.Collectors;
import org.slf4j.Logger;
@@ -184,7 +185,12 @@ public class RemoteUnSubscribeEventProcessor extends
AbstractEventProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.UNSUBSCRIBE_REMOTE.getRequestURI()};
+ return new String[] {RequestURI.UNSUBSCRIBE_REMOTE.getRequestURI()};
+ }
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
index d23a26d8c..eb5cb8786 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
@@ -39,7 +39,6 @@ import
org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
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.HttpRequestProcessor;
import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
import org.apache.eventmesh.runtime.core.protocol.producer.SendMessageContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
@@ -48,6 +47,7 @@ import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
@@ -57,7 +57,7 @@ import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.netty.channel.ChannelHandlerContext;
-public class ReplyMessageProcessor implements HttpRequestProcessor {
+public class ReplyMessageProcessor extends AbstractHttpRequestProcessor {
public static final Logger MESSAGE_LOGGER =
LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
@@ -247,4 +247,9 @@ public class ReplyMessageProcessor implements
HttpRequestProcessor {
summaryMetrics.recordReplyMsgCost(endTime - startTime);
}
}
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getReplyMsgExecutor();
+ }
}
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 2768e0e31..82cb65fc7 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
@@ -52,6 +52,7 @@ import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import io.cloudevents.CloudEvent;
@@ -308,6 +309,11 @@ public class SendAsyncEventProcessor implements
AsyncHttpProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.PUBLISH.getRequestURI()};
+ return new String[] {RequestURI.PUBLISH.getRequestURI()};
+ }
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor();
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
index 332f69ce3..6c4e60fd8 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
@@ -40,7 +40,6 @@ import
org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
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.HttpRequestProcessor;
import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
import org.apache.eventmesh.runtime.core.protocol.producer.SendMessageContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
@@ -52,6 +51,7 @@ import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.Objects;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
@@ -62,7 +62,7 @@ import io.cloudevents.core.builder.CloudEventBuilder;
import io.netty.channel.ChannelHandlerContext;
import io.opentelemetry.api.trace.Span;
-public class SendAsyncMessageProcessor implements HttpRequestProcessor {
+public class SendAsyncMessageProcessor extends AbstractHttpRequestProcessor {
private static final Logger MESSAGE_LOGGER =
LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
@@ -89,7 +89,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
HttpCommand request = asyncContext.getRequest();
String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
CMD_LOGGER.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(
- Integer.valueOf(request.getRequestCode())),
+ Integer.valueOf(request.getRequestCode())),
EventMeshConstants.PROTOCOL_HTTP,
remoteAddr, localAddress);
@@ -306,6 +306,11 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
}
}
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor();
+ }
+
private void spanWithException(CloudEvent event, String protocolVersion,
EventMeshRetCode retCode) {
Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
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 44c7cf1d4..baa93d4cb 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
@@ -50,6 +50,7 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import io.cloudevents.CloudEvent;
@@ -322,7 +323,12 @@ public class SendAsyncRemoteEventProcessor implements
AsyncHttpProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.PUBLISH_BRIDGE.getRequestURI()};
+ return new String[] {RequestURI.PUBLISH_BRIDGE.getRequestURI()};
+ }
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getRemoteMsgExecutor();
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
index 529cf6654..b687156a4 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
@@ -39,7 +39,6 @@ import
org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
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.HttpRequestProcessor;
import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
import org.apache.eventmesh.runtime.core.protocol.producer.SendMessageContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
@@ -49,6 +48,7 @@ import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.Objects;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import io.cloudevents.CloudEvent;
@@ -58,7 +58,7 @@ import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class SendSyncMessageProcessor implements HttpRequestProcessor {
+public class SendSyncMessageProcessor extends AbstractHttpRequestProcessor {
private transient EventMeshHTTPServer eventMeshHTTPServer;
@@ -215,6 +215,7 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
@Override
public void onSuccess(final CloudEvent event) {
+
log.info("message|mq2eventMesh|RSP|SYNC|rrCost={}ms|topic={}"
+ "|bizSeqNo={}|uniqueId={}",
System.currentTimeMillis() - startTime, topic, bizNo, uniqueId);
@@ -262,8 +263,10 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
asyncContext.onComplete(err, handler);
eventMeshHTTPServer.getHttpRetryer().newTimeout(sendMessageContext, 10,
TimeUnit.SECONDS);
+
log.error("message|mq2eventMesh|RSP|SYNC|rrCost={}ms|topic={}"
+ "|bizSeqNo={}|uniqueId={}",
System.currentTimeMillis() - startTime, topic, bizNo, uniqueId, e);
+
}
}, Integer.parseInt(ttl));
} catch (Exception ex) {
@@ -282,4 +285,9 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
return;
}
+
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor();
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
index 0c062f8f3..76ec80c66 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
@@ -22,4 +22,5 @@ package
org.apache.eventmesh.runtime.core.protocol.http.processor;
*/
public interface ShortHttpProcessor extends HttpProcessor {
+
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
index ddaaaf1c7..f4dda2a76 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
@@ -36,7 +36,6 @@ import org.apache.eventmesh.runtime.core.consumer.ClientInfo;
import org.apache.eventmesh.runtime.core.consumer.SubscriptionManager;
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.HttpRequestProcessor;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.eventmesh.runtime.util.WebhookUtil;
@@ -45,13 +44,14 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.List;
+import java.util.concurrent.Executor;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class SubscribeProcessor implements HttpRequestProcessor {
+public class SubscribeProcessor extends AbstractHttpRequestProcessor {
private final transient EventMeshHTTPServer eventMeshHTTPServer;
@@ -186,8 +186,10 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2),
SubscribeResponseBody.class);
final long endTime = System.currentTimeMillis();
+
log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime,
JsonUtils.toJSONString(subscribeRequestBody.getTopics()),
subscribeRequestBody.getUrl(), e);
+
summaryMetrics.recordSendMsgFailed();
summaryMetrics.recordSendMsgCost(endTime - startTime);
}
@@ -195,6 +197,11 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
}
}
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
+ }
+
private ClientInfo getClientInfo(final SubscribeRequestHeader
subscribeRequestHeader) {
ClientInfo clientInfo = new ClientInfo();
clientInfo.setEnv(subscribeRequestHeader.getEnv());
@@ -204,4 +211,6 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
clientInfo.setPid(subscribeRequestHeader.getPid());
return clientInfo;
}
+
+
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
index f536909f4..0392e1272 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
@@ -36,7 +36,6 @@ 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.Client;
-import
org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
@@ -52,13 +51,14 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class UnSubscribeProcessor implements HttpRequestProcessor {
+public class UnSubscribeProcessor extends AbstractHttpRequestProcessor {
private final transient EventMeshHTTPServer eventMeshHTTPServer;
@@ -67,39 +67,35 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
}
@Override
- public void processRequest(final ChannelHandlerContext ctx, final
AsyncContext<HttpCommand> asyncContext)
- throws Exception {
+ public void processRequest(final ChannelHandlerContext ctx, final
AsyncContext<HttpCommand> asyncContext) throws Exception {
HttpCommand responseEventMeshCommand;
final HttpCommand request = asyncContext.getRequest();
final String localAddress = IPUtils.getLocalAddress();
log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(Integer.valueOf(request.getRequestCode())),
+
EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
final UnSubscribeRequestHeader unSubscribeRequestHeader =
(UnSubscribeRequestHeader) request.getHeader();
final UnSubscribeRequestBody unSubscribeRequestBody =
(UnSubscribeRequestBody) request.getBody();
EventMeshHTTPConfiguration eventMeshHttpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
- final UnSubscribeResponseHeader unSubscribeResponseHeader =
- UnSubscribeResponseHeader
- .buildHeader(Integer.valueOf(request.getRequestCode()),
- eventMeshHttpConfiguration.getEventMeshCluster(),
- localAddress, eventMeshHttpConfiguration.getEventMeshEnv(),
- eventMeshHttpConfiguration.getEventMeshIDC());
+ final UnSubscribeResponseHeader unSubscribeResponseHeader =
UnSubscribeResponseHeader.buildHeader(Integer.valueOf(request.getRequestCode()),
+ eventMeshHttpConfiguration.getEventMeshCluster(), localAddress,
eventMeshHttpConfiguration.getEventMeshEnv(),
+ eventMeshHttpConfiguration.getEventMeshIDC());
// validate header
- if (StringUtils.isAnyBlank(unSubscribeRequestHeader.getIdc(),
unSubscribeRequestHeader.getPid(),
- unSubscribeRequestHeader.getSys())
+ if (StringUtils.isAnyBlank(unSubscribeRequestHeader.getIdc(),
unSubscribeRequestHeader.getPid(), unSubscribeRequestHeader.getSys())
|| !StringUtils.isNumeric(unSubscribeRequestHeader.getPid())) {
- completeResponse(request, asyncContext, unSubscribeResponseHeader,
- EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
UnSubscribeResponseBody.class);
+ completeResponse(request, asyncContext, unSubscribeResponseHeader,
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
+ UnSubscribeResponseBody.class);
return;
}
// validate body
- if (StringUtils.isAnyBlank(unSubscribeRequestBody.getUrl(),
unSubscribeRequestBody.getConsumerGroup())
- || CollectionUtils.isEmpty(unSubscribeRequestBody.getTopics())) {
- completeResponse(request, asyncContext, unSubscribeResponseHeader,
- EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
UnSubscribeResponseBody.class);
+ if (StringUtils.isAnyBlank(unSubscribeRequestBody.getUrl(),
unSubscribeRequestBody.getConsumerGroup()) || CollectionUtils.isEmpty(
+ unSubscribeRequestBody.getTopics())) {
+ completeResponse(request, asyncContext, unSubscribeResponseHeader,
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
+ UnSubscribeResponseBody.class);
return;
}
@@ -119,23 +115,23 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
};
SubscriptionManager subscriptionManager =
eventMeshHTTPServer.getSubscriptionManager();
- ConcurrentHashMap<String, ConsumerGroupConf> localConsumerGroupMap =
- subscriptionManager.getLocalConsumerGroupMapping();
+ ConcurrentHashMap<String, ConsumerGroupConf> localConsumerGroupMap =
subscriptionManager.getLocalConsumerGroupMapping();
synchronized (subscriptionManager.getLocalClientInfoMapping()) {
boolean isChange = true;
registerClient(unSubscribeRequestHeader, consumerGroup,
unSubTopicList, unSubscribeUrl);
for (final String unSubTopic : unSubTopicList) {
- final List<Client> groupTopicClients =
subscriptionManager.getLocalClientInfoMapping()
- .get(consumerGroup + "@" + unSubTopic);
+ final List<Client> groupTopicClients =
subscriptionManager.getLocalClientInfoMapping().get(consumerGroup + "@" +
unSubTopic);
final Iterator<Client> clientIterator =
groupTopicClients.iterator();
while (clientIterator.hasNext()) {
final Client client = clientIterator.next();
+
if (StringUtils.equals(client.getPid(), pid)
&& StringUtils.equals(client.getUrl(),
unSubscribeUrl)) {
log.warn("client {} start unsubscribe",
JsonUtils.toJSONString(client));
+
clientIterator.remove();
}
}
@@ -179,17 +175,14 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
final long startTime = System.currentTimeMillis();
if (isChange) {
try {
-
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
- localConsumerGroupMap.get(consumerGroup));
+
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
localConsumerGroupMap.get(consumerGroup));
responseEventMeshCommand =
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
- completeResponse(request, asyncContext,
unSubscribeResponseHeader,
- EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR,
- EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg()
+ EventMeshUtil.stackTrace(e, 2),
- UnSubscribeResponseBody.class);
+ completeResponse(request, asyncContext,
unSubscribeResponseHeader, EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR,
+ EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg()
+ EventMeshUtil.stackTrace(e, 2), UnSubscribeResponseBody.class);
final long endTime = System.currentTimeMillis();
log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}",
endTime - startTime,
JsonUtils.toJSONString(unSubscribeRequestBody.getTopics()),
unSubscribeRequestBody.getUrl(), e);
@@ -199,22 +192,16 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
} else {
// remove
try {
- eventMeshHTTPServer.getConsumerManager()
- .notifyConsumerManager(consumerGroup, null);
- responseEventMeshCommand =
-
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
+
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
null);
+ responseEventMeshCommand =
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
// clean ClientInfo
- subscriptionManager.getLocalClientInfoMapping().keySet()
- .removeIf(s -> StringUtils.contains(s, consumerGroup));
+
subscriptionManager.getLocalClientInfoMapping().keySet().removeIf(s ->
StringUtils.contains(s, consumerGroup));
// clean ConsumerGroupInfo
- localConsumerGroupMap.keySet()
- .removeIf(s -> StringUtils.equals(consumerGroup, s));
+ localConsumerGroupMap.keySet().removeIf(s ->
StringUtils.equals(consumerGroup, s));
} catch (Exception e) {
- completeResponse(request, asyncContext,
unSubscribeResponseHeader,
- EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR,
- EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg()
+ EventMeshUtil.stackTrace(e, 2),
- UnSubscribeResponseBody.class);
+ completeResponse(request, asyncContext,
unSubscribeResponseHeader, EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR,
+ EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg()
+ EventMeshUtil.stackTrace(e, 2), UnSubscribeResponseBody.class);
final long endTime = System.currentTimeMillis();
log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}",
endTime - startTime,
JsonUtils.toJSONString(unSubscribeRequestBody.getTopics()),
unSubscribeRequestBody.getUrl(), e);
@@ -226,10 +213,14 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
}
}
- private void registerClient(final UnSubscribeRequestHeader
unSubscribeRequestHeader,
- final String consumerGroup,
- final List<String> topicList,
- final String url) {
+ @Override
+ public Executor executor() {
+ return
eventMeshHTTPServer.getHttpThreadPoolGroup().getClientManageExecutor();
+ }
+
+
+ private void registerClient(final UnSubscribeRequestHeader
unSubscribeRequestHeader, final String consumerGroup,
+ final List<String> topicList, final String url) {
for (final String topic : topicList) {
final Client client = new Client();
client.setEnv(unSubscribeRequestHeader.getEnv());
@@ -243,8 +234,7 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
client.setLastUpTime(new Date());
final String groupTopicKey = client.getConsumerGroup() + "@" +
client.getTopic();
- ConcurrentHashMap<String, List<Client>> localClientInfoMap =
eventMeshHTTPServer.getSubscriptionManager()
- .getLocalClientInfoMapping();
+ ConcurrentHashMap<String, List<Client>> localClientInfoMap =
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping();
if (localClientInfoMap.containsKey(groupTopicKey)) {
final List<Client> localClients =
localClientInfoMap.get(groupTopicKey);
boolean isContains = false;
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 256b149b3..0e0aaa275 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
@@ -41,7 +41,7 @@ public class WebHookProcessor implements ShortHttpProcessor {
@Override
public String[] paths() {
- return new String[]{WebHookOperationConstant.CALLBACK_PATH_PREFIX};
+ return new String[] {WebHookOperationConstant.CALLBACK_PATH_PREFIX};
}
@Override
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/HttpRequestProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/HttpRequestProcessor.java
index 570f11c99..35231ef8b 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/HttpRequestProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/HttpRequestProcessor.java
@@ -27,6 +27,7 @@ import org.apache.commons.lang3.StringUtils;
import java.lang.reflect.Method;
import java.util.Objects;
+import java.util.concurrent.Executor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,4 +68,9 @@ public interface HttpRequestProcessor {
return Objects.isNull(extension) ? "" : extension.toString();
}
+ /**
+ * @return {@link Executor}
+ */
+ Executor executor();
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
