This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 02b9448e9 [ISSUE #3820]HttpRequestProcessor and its subclasses can be
refactored. (#3842)
02b9448e9 is described below
commit 02b9448e96c530ebb70f4560f7df08b7ad449ac7
Author: pandaapo <[email protected]>
AuthorDate: Fri May 5 20:48:31 2023 +0800
[ISSUE #3820]HttpRequestProcessor and its subclasses can be refactored.
(#3842)
* Refactor HttpRequestProcessor and its subclasses.
* Refactor HttpRequestProcessor's subclasses.
* Fix checkstyle error.
* Revoke controversial import static.
* Fix compile error.
---
.../http/processor/AdminMetricsProcessor.java | 16 +-
.../http/processor/AdminShutdownProcessor.java | 8 +-
.../http/processor/BatchSendMessageProcessor.java | 177 ++++++++------------
.../processor/BatchSendMessageV2Processor.java | 177 +++++++-------------
.../http/processor/HeartBeatProcessor.java | 93 +++++------
.../http/processor/ReplyMessageProcessor.java | 185 ++++++++-------------
.../http/processor/SendAsyncMessageProcessor.java | 113 +++++--------
.../http/processor/SendSyncMessageProcessor.java | 121 ++++----------
.../http/processor/SubscribeProcessor.java | 74 +++------
.../http/processor/UnSubscribeProcessor.java | 48 ++----
.../http/processor/inf/HttpRequestProcessor.java | 32 +++-
11 files changed, 382 insertions(+), 662 deletions(-)
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 a558b8704..817d339b6 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
@@ -24,24 +24,14 @@ import
org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequest
import io.netty.channel.ChannelHandlerContext;
+import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
+@RequiredArgsConstructor
public class AdminMetricsProcessor implements HttpRequestProcessor {
- private EventMeshHTTPServer eventMeshHTTPServer;
-
- public AdminMetricsProcessor(EventMeshHTTPServer eventMeshHTTPServer) {
- this.eventMeshHTTPServer = eventMeshHTTPServer;
- }
+ private final EventMeshHTTPServer eventMeshHTTPServer;
@Override
public void processRequest(ChannelHandlerContext ctx,
AsyncContext<HttpCommand> asyncContext) throws Exception {
}
-
- @Override
- public boolean rejectRequest() {
- return false;
- }
}
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 641ddd8b9..d0b11f972 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
@@ -34,6 +34,7 @@ import io.netty.channel.ChannelHandlerContext;
import lombok.RequiredArgsConstructor;
+
@RequiredArgsConstructor
public class AdminShutdownProcessor implements HttpRequestProcessor {
@@ -47,16 +48,11 @@ public class AdminShutdownProcessor implements
HttpRequestProcessor {
String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
cmdLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP, remoteAddr,
IPUtils.getLocalAddress());
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr,
IPUtils.getLocalAddress());
eventMeshServer.shutdown();
HttpCommand responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand);
}
-
- @Override
- public boolean rejectRequest() {
- return false;
- }
}
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 277d64142..22a05f37f 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
@@ -20,22 +20,24 @@ package
org.apache.eventmesh.runtime.core.protocol.http.processor;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageBatchRequestBody;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageBatchResponseBody;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
-import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
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.RequestCode;
import
org.apache.eventmesh.common.protocol.http.header.message.SendMessageBatchRequestHeader;
import
org.apache.eventmesh.common.protocol.http.header.message.SendMessageBatchResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
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.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;
@@ -44,13 +46,12 @@ import
org.apache.eventmesh.runtime.core.protocol.http.producer.SendMessageConte
import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
@@ -79,37 +80,34 @@ public class BatchSendMessageProcessor implements
HttpRequestProcessor {
this.acl = eventMeshHTTPServer.getAcl();
}
-
-
@Override
public void processRequest(ChannelHandlerContext ctx,
AsyncContext<HttpCommand> asyncContext) throws Exception {
HttpCommand responseEventMeshCommand;
String localAddress = IPUtils.getLocalAddress();
+ HttpCommand request = asyncContext.getRequest();
cmdLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(
- Integer.valueOf(asyncContext.getRequest().getRequestCode())),
+ Integer.valueOf(request.getRequestCode())),
EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
localAddress);
SendMessageBatchRequestHeader sendMessageBatchRequestHeader =
- (SendMessageBatchRequestHeader)
asyncContext.getRequest().getHeader();
+ (SendMessageBatchRequestHeader) request.getHeader();
+ EventMeshHTTPConfiguration httpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
SendMessageBatchResponseHeader sendMessageBatchResponseHeader =
-
SendMessageBatchResponseHeader.buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster(),
- localAddress,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+
SendMessageBatchResponseHeader.buildHeader(Integer.valueOf(request.getRequestCode()),
+ httpConfiguration.getEventMeshCluster(),
+ localAddress, httpConfiguration.getEventMeshEnv(),
+ httpConfiguration.getEventMeshIDC());
String protocolType = sendMessageBatchRequestHeader.getProtocolType();
ProtocolAdaptor<ProtocolTransportObject> httpCommandProtocolAdaptor =
ProtocolPluginFactory.getProtocolAdaptor(protocolType);
- List<CloudEvent> eventList =
httpCommandProtocolAdaptor.toBatchCloudEvent(asyncContext.getRequest());
+ List<CloudEvent> eventList =
httpCommandProtocolAdaptor.toBatchCloudEvent(request);
if (CollectionUtils.isEmpty(eventList)) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
SendMessageBatchResponseBody.class);
return;
}
@@ -117,89 +115,68 @@ public class BatchSendMessageProcessor implements
HttpRequestProcessor {
String producerGroup = "";
int eventSize = eventList.size();
- if (eventSize >
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventBatchSize())
{
+ if (eventSize > httpConfiguration.getEventMeshEventBatchSize()) {
batchMessageLogger.error("Event batch size exceeds the limit: {}",
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventBatchSize());
-
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- "Event batch size exceeds the limit: " +
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventBatchSize()));
- asyncContext.onComplete(responseEventMeshCommand);
+ httpConfiguration.getEventMeshEventBatchSize());
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
+ "Event batch size exceeds the limit: " +
httpConfiguration.getEventMeshEventBatchSize(),
+ SendMessageBatchResponseBody.class);
return;
}
for (CloudEvent event : eventList) {
//validate event
- if (StringUtils.isBlank(event.getId())
- || event.getSource() == null
- || event.getSpecVersion() == null
- || StringUtils.isBlank(event.getType())
- || StringUtils.isBlank(event.getSubject())) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ if (!ObjectUtils.allNotNull(event.getSource(),
event.getSpecVersion())
+ || StringUtils.isAnyBlank(event.getId(), event.getType(),
event.getSubject())) {
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SendMessageBatchResponseBody.class);
return;
}
- String content = event.getData() == null ? "" : new
String(event.getData().toBytes(), StandardCharsets.UTF_8);
- if (content.length() >
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()) {
+ String content = event.getData() == null ? "" : new
String(event.getData().toBytes(), Constants.DEFAULT_CHARSET);
+ if (content.length() > httpConfiguration.getEventMeshEventSize()) {
batchMessageLogger.error("Event size exceeds the limit: {}",
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize());
-
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
- "Event size exceeds the limit: " +
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()));
- asyncContext.onComplete(responseEventMeshCommand);
+ httpConfiguration.getEventMeshEventSize());
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR,
+ "Event size exceeds the limit: " +
httpConfiguration.getEventMeshEventSize(),
+ SendMessageBatchResponseBody.class);
return;
}
- String idc =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC)).toString();
- String pid =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID)).toString();
- String sys =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS)).toString();
+ String idc = getExtension(event,
ProtocolKey.ClientInstanceKey.IDC);
+ String pid = getExtension(event,
ProtocolKey.ClientInstanceKey.PID);
+ String sys = getExtension(event,
ProtocolKey.ClientInstanceKey.SYS);
//validate event-extension
- if (StringUtils.isBlank(idc)
- || StringUtils.isBlank(pid)
- || !StringUtils.isNumeric(pid)
- || StringUtils.isBlank(sys)) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ if (StringUtils.isAnyBlank(idc, pid, sys)
+ || !StringUtils.isNumeric(pid)) {
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SendMessageBatchResponseBody.class);
return;
}
- batchId =
Objects.requireNonNull(event.getExtension(SendMessageBatchRequestBody.BATCHID)).toString();
- producerGroup =
Objects.requireNonNull(event.getExtension(SendMessageBatchRequestBody.PRODUCERGROUP)).toString();
- eventSize =
Integer.parseInt(Objects.requireNonNull(event.getExtension(SendMessageBatchRequestBody.SIZE)).toString());
+ batchId = getExtension(event, SendMessageBatchRequestBody.BATCHID);
+ producerGroup = getExtension(event,
SendMessageBatchRequestBody.PRODUCERGROUP);
+ eventSize = Integer.parseInt(getExtension(event,
SendMessageBatchRequestBody.SIZE));
CloudEventData eventData = event.getData();
- if (eventData == null || StringUtils.isBlank(batchId)
- || StringUtils.isBlank(producerGroup)
+ if (eventData == null || StringUtils.isAnyBlank(batchId,
producerGroup)
|| eventSize != eventList.size()) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
SendMessageBatchResponseBody.class);
return;
}
}
+ HttpSummaryMetrics summaryMetrics =
eventMeshHTTPServer.getMetrics().getSummaryMetrics();
if (!eventMeshHTTPServer.getBatchRateLimiter()
.tryAcquire(eventSize,
EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getErrMsg()));
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendBatchMsgDiscard(eventSize);
- asyncContext.onComplete(responseEventMeshCommand);
+ summaryMetrics.recordSendBatchMsgDiscard(eventSize);
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR,
null, SendMessageBatchResponseBody.class);
return;
}
@@ -208,49 +185,41 @@ public class BatchSendMessageProcessor implements
HttpRequestProcessor {
batchEventMeshProducer.getMqProducerWrapper().getMeshMQProducer().setExtFields();
if (!batchEventMeshProducer.isStarted()) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR,
null, SendMessageBatchResponseBody.class);
return;
}
final long batchStartTime = System.currentTimeMillis();
String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- int requestCode =
Integer.parseInt(asyncContext.getRequest().getRequestCode());
+ int requestCode = Integer.parseInt(request.getRequestCode());
Map<String, List<CloudEvent>> topicBatchMessageMappings = new
ConcurrentHashMap<String, List<CloudEvent>>();
for (CloudEvent cloudEvent : eventList) {
- if (StringUtils.isBlank(cloudEvent.getSubject())
- || cloudEvent.getData() == null) {
+ if (StringUtils.isBlank(cloudEvent.getSubject()) ||
cloudEvent.getData() == null) {
continue;
}
- String user =
Objects.requireNonNull(cloudEvent.getExtension(ProtocolKey.ClientInstanceKey.USERNAME)).toString();
- String pass =
Objects.requireNonNull(cloudEvent.getExtension(ProtocolKey.ClientInstanceKey.PASSWD)).toString();
- String subsystem =
Objects.requireNonNull(cloudEvent.getExtension(ProtocolKey.ClientInstanceKey.SYS)).toString();
+ String user = getExtension(cloudEvent,
ProtocolKey.ClientInstanceKey.USERNAME);
+ String pass = getExtension(cloudEvent,
ProtocolKey.ClientInstanceKey.PASSWD);
+ String subsystem = getExtension(cloudEvent,
ProtocolKey.ClientInstanceKey.SYS);
//do acl check
- if
(eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable())
{
+ if (httpConfiguration.isEventMeshServerSecurityEnable()) {
try {
this.acl.doAclCheckInHttpSend(remoteAddr, user, pass,
subsystem, cloudEvent.getSubject(), requestCode);
} catch (Exception e) {
-
-
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(),
e.getMessage()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_ACL_ERR,
e.getMessage(), SendMessageBatchResponseBody.class);
aclLogger.warn("CLIENT HAS NO
PERMISSION,BatchSendMessageProcessor send failed", e);
return;
}
}
try {
- String ttl =
Objects.requireNonNull(cloudEvent.getExtension(SendMessageRequestBody.TTL)).toString();
+ String ttl = getExtension(cloudEvent,
SendMessageRequestBody.TTL);
if (StringUtils.isBlank(ttl) || !StringUtils.isNumeric(ttl)) {
cloudEvent = CloudEventBuilder.from(cloudEvent)
@@ -277,18 +246,15 @@ public class BatchSendMessageProcessor implements
HttpRequestProcessor {
}
if (CollectionUtils.isEmpty(eventList)) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
SendMessageBatchResponseBody.class);
return;
}
long delta = eventSize;
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendBatchMsg(delta);
+ summaryMetrics.recordSendBatchMsg(delta);
- if
(eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerBatchMsgBatchEnabled())
{
+ if (httpConfiguration.isEventMeshServerBatchMsgBatchEnabled()) {
for (List<CloudEvent> eventlist :
topicBatchMessageMappings.values()) {
// TODO: Implementation in API. Consider whether to put it in
the plug-in.
CloudEvent event = null;
@@ -330,21 +296,12 @@ public class BatchSendMessageProcessor implements
HttpRequestProcessor {
}
long batchEndTime = System.currentTimeMillis();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordBatchSendMsgCost(batchEndTime
- batchStartTime);
+ summaryMetrics.recordBatchSendMsgCost(batchEndTime - batchStartTime);
batchMessageLogger.debug("batchMessage|eventMesh2mq|REQ|ASYNC|batchId={}|send2MQCost={}ms|msgNum={}|topics={}",
batchId, batchEndTime - batchStartTime, eventSize,
topicBatchMessageMappings.keySet());
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
-
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(),
- EventMeshRetCode.SUCCESS.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext,
sendMessageBatchResponseHeader, EventMeshRetCode.SUCCESS, null,
+ SendMessageBatchResponseBody.class);
return;
}
-
- @Override
- public boolean rejectRequest() {
- return false;
- }
}
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 64bd8f15f..e32bfd3cb 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
@@ -20,21 +20,23 @@ package
org.apache.eventmesh.runtime.core.protocol.http.processor;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageBatchV2RequestBody;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageBatchV2ResponseBody;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
-import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
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.RequestCode;
import
org.apache.eventmesh.common.protocol.http.header.message.SendMessageBatchV2RequestHeader;
import
org.apache.eventmesh.common.protocol.http.header.message.SendMessageBatchV2ResponseHeader;
+import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
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.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;
@@ -43,9 +45,9 @@ import
org.apache.eventmesh.runtime.core.protocol.http.producer.SendMessageConte
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
+import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
-import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
@@ -77,8 +79,6 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
@Override
public void processRequest(ChannelHandlerContext ctx,
AsyncContext<HttpCommand> asyncContext)
throws Exception {
-
- HttpCommand responseEventMeshCommand;
final HttpCommand request = asyncContext.getRequest();
final Integer requestCode = Integer.valueOf(request.getRequestCode());
@@ -89,94 +89,66 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
SendMessageBatchV2RequestHeader sendMessageBatchV2RequestHeader =
- (SendMessageBatchV2RequestHeader)
asyncContext.getRequest().getHeader();
+ (SendMessageBatchV2RequestHeader) request.getHeader();
String protocolType =
sendMessageBatchV2RequestHeader.getProtocolType();
ProtocolAdaptor<ProtocolTransportObject> httpCommandProtocolAdaptor =
ProtocolPluginFactory.getProtocolAdaptor(protocolType);
- CloudEvent event =
httpCommandProtocolAdaptor.toCloudEvent(asyncContext.getRequest());
+ CloudEvent event = httpCommandProtocolAdaptor.toCloudEvent(request);
+ EventMeshHTTPConfiguration httpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
SendMessageBatchV2ResponseHeader sendMessageBatchV2ResponseHeader =
SendMessageBatchV2ResponseHeader.buildHeader(
- requestCode,
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC()
+ requestCode,
+ httpConfiguration.getEventMeshCluster(),
+ httpConfiguration.getEventMeshEnv(),
+ httpConfiguration.getEventMeshIDC()
);
// todo: use validate processor to check
//validate event
- if (StringUtils.isBlank(event.getId())
- || event.getSource() == null
- || event.getSpecVersion() == null
- || StringUtils.isBlank(event.getType())
- || StringUtils.isBlank(event.getSubject())) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ if (!ObjectUtils.allNotNull(event.getSource(), event.getSpecVersion())
+ || StringUtils.isAnyBlank(event.getId(), event.getType(),
event.getSubject())) {
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SendMessageBatchV2ResponseBody.class);
return;
}
- String idc =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC))
- .toString();
- String pid =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID))
- .toString();
- String sys =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS))
- .toString();
+ String idc = getExtension(event, ProtocolKey.ClientInstanceKey.IDC);
+ String pid = getExtension(event, ProtocolKey.ClientInstanceKey.PID);
+ String sys = getExtension(event, ProtocolKey.ClientInstanceKey.SYS);
//validate event-extension
- if (StringUtils.isBlank(idc)
- || StringUtils.isBlank(pid)
- || !StringUtils.isNumeric(pid)
- || StringUtils.isBlank(sys)) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ if (StringUtils.isAnyBlank(idc, pid, sys)
+ || !StringUtils.isNumeric(pid)) {
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SendMessageBatchV2ResponseBody.class);
return;
}
- String bizNo =
-
Objects.requireNonNull(event.getExtension(SendMessageBatchV2RequestBody.BIZSEQNO))
- .toString();
- String producerGroup =
-
Objects.requireNonNull(event.getExtension(SendMessageBatchV2RequestBody.PRODUCERGROUP))
- .toString();
+ String bizNo = getExtension(event,
SendMessageBatchV2RequestBody.BIZSEQNO);
+ String producerGroup = getExtension(event,
SendMessageBatchV2RequestBody.PRODUCERGROUP);
String topic = event.getSubject();
- if (StringUtils.isBlank(bizNo)
- || StringUtils.isBlank(topic)
- || StringUtils.isBlank(producerGroup)
+ if (StringUtils.isAnyBlank(bizNo, topic, producerGroup)
|| event.getData() == null) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
SendMessageBatchV2ResponseBody.class);
return;
}
- String content = new
String(Objects.requireNonNull(event.getData()).toBytes(),
StandardCharsets.UTF_8);
- if (content.length() >
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()) {
- batchMessageLogger.error("Event size exceeds the limit: {}",
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize());
-
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
-
SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- "Event size exceeds the limit: " +
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()));
- asyncContext.onComplete(responseEventMeshCommand);
+ String content = new
String(Objects.requireNonNull(event.getData()).toBytes(),
Constants.DEFAULT_CHARSET);
+ if (content.length() > httpConfiguration.getEventMeshEventSize()) {
+ batchMessageLogger.error("Event size exceeds the limit: {}",
httpConfiguration.getEventMeshEventSize());
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
+ "Event size exceeds the limit: " +
httpConfiguration.getEventMeshEventSize(),
+ SendMessageBatchV2ResponseBody.class);
return;
}
//do acl check
- if
(eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable())
{
+ if (httpConfiguration.isEventMeshServerSecurityEnable()) {
String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
String user = getExtension(event,
ProtocolKey.ClientInstanceKey.USERNAME);
String pass = getExtension(event,
ProtocolKey.ClientInstanceKey.PASSWD);
@@ -184,28 +156,19 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
try {
this.acl.doAclCheckInHttpSend(remoteAddr, user, pass,
subsystem, topic, requestCode);
} catch (Exception e) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(),
- e.getMessage()));
- asyncContext.onComplete(responseEventMeshCommand);
- aclLogger
- .warn("CLIENT HAS NO
PERMISSION,BatchSendMessageV2Processor send failed", e);
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
+ EventMeshRetCode.EVENTMESH_ACL_ERR, e.getMessage(),
SendMessageBatchV2ResponseBody.class);
+ aclLogger.warn("CLIENT HAS NO
PERMISSION,BatchSendMessageV2Processor send failed", e);
return;
}
}
+ HttpSummaryMetrics summaryMetrics =
eventMeshHTTPServer.getMetrics().getSummaryMetrics();
if (!eventMeshHTTPServer.getBatchRateLimiter()
-
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
- TimeUnit.MILLISECONDS)) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getErrMsg()));
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendBatchMsgDiscard(1);
- asyncContext.onComplete(responseEventMeshCommand);
+
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
+ summaryMetrics.recordSendBatchMsgDiscard(1);
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
+ EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR,
null, SendMessageBatchV2ResponseBody.class);
return;
}
@@ -213,12 +176,8 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);
batchEventMeshProducer.getMqProducerWrapper().getMeshMQProducer().setExtFields();
if (!batchEventMeshProducer.isStarted()) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
+ EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR,
null, SendMessageBatchV2ResponseBody.class);
return;
}
@@ -246,18 +205,14 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
} catch (Exception e) {
batchMessageLogger.error("msg2MQMsg err, topic:{}, msg:{}", topic,
event.getData(), e);
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg()
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader, EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR,
+ EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg()
+
- EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(responseEventMeshCommand);
+ EventMeshUtil.stackTrace(e, 2),
SendMessageBatchV2ResponseBody.class);
return;
}
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendBatchMsg(1);
+ summaryMetrics.recordSendBatchMsg(1);
final SendMessageContext sendMessageContext =
new SendMessageContext(bizNo, event, batchEventMeshProducer,
eventMeshHTTPServer);
@@ -267,7 +222,7 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
@Override
public void onSuccess(SendResult sendResult) {
long batchEndTime = System.currentTimeMillis();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordBatchSendMsgCost(batchEndTime
- batchStartTime);
+ summaryMetrics.recordBatchSendMsgCost(batchEndTime -
batchStartTime);
batchMessageLogger.debug(
"batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
bizNo, batchEndTime - batchStartTime, topic);
@@ -277,7 +232,7 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
public void onException(OnExceptionContext context) {
long batchEndTime = System.currentTimeMillis();
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordBatchSendMsgCost(batchEndTime
- batchStartTime);
+ summaryMetrics.recordBatchSendMsgCost(batchEndTime -
batchStartTime);
batchMessageLogger.error(
"batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
bizNo, batchEndTime - batchStartTime, topic,
context.getException());
@@ -285,38 +240,20 @@ public class BatchSendMessageV2Processor implements
HttpRequestProcessor {
});
} catch (Exception e) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_SEND_BATCHLOG_MSG_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_SEND_BATCHLOG_MSG_ERR.getErrMsg()
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
EventMeshRetCode.EVENTMESH_SEND_BATCHLOG_MSG_ERR,
+
EventMeshRetCode.EVENTMESH_SEND_BATCHLOG_MSG_ERR.getErrMsg()
+
- EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(responseEventMeshCommand);
+ EventMeshUtil.stackTrace(e, 2),
SendMessageBatchV2ResponseBody.class);
long batchEndTime = System.currentTimeMillis();
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordBatchSendMsgCost(batchEndTime
- batchStartTime);
+ summaryMetrics.recordBatchSendMsgCost(batchEndTime -
batchStartTime);
batchMessageLogger.error(
"batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
bizNo, batchEndTime - batchStartTime, topic, e);
}
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
-
SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(),
- EventMeshRetCode.SUCCESS.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
-
- }
-
- private String getExtension(CloudEvent event, String protocolKey) {
- Object extension = event.getExtension(protocolKey);
- return Objects.isNull(extension) ? "" : extension.toString();
- }
-
- @Override
- public boolean rejectRequest() {
- return false;
+ completeResponse(request, asyncContext,
sendMessageBatchV2ResponseHeader,
+ EventMeshRetCode.SUCCESS, null,
SendMessageBatchV2ResponseBody.class);
}
}
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 893b4192d..e72a9bb79 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
@@ -21,14 +21,15 @@ package
org.apache.eventmesh.runtime.core.protocol.http.processor;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
import
org.apache.eventmesh.common.protocol.http.body.client.HeartbeatRequestBody;
import
org.apache.eventmesh.common.protocol.http.body.client.HeartbeatResponseBody;
-import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import
org.apache.eventmesh.common.protocol.http.header.client.HeartbeatRequestHeader;
import
org.apache.eventmesh.common.protocol.http.header.client.HeartbeatResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.runtime.acl.Acl;
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.async.CompleteHandler;
@@ -67,44 +68,37 @@ public class HeartBeatProcessor implements
HttpRequestProcessor {
public void processRequest(final ChannelHandlerContext ctx, final
AsyncContext<HttpCommand> asyncContext) throws Exception {
HttpCommand responseEventMeshCommand;
final String localAddress = IPUtils.getLocalAddress();
-
+ HttpCommand request = asyncContext.getRequest();
if (log.isInfoEnabled()) {
log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
-
RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
+ RequestCode.get(Integer.valueOf(request.getRequestCode())),
EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
localAddress);
}
- final HeartbeatRequestHeader heartbeatRequestHeader =
(HeartbeatRequestHeader) asyncContext.getRequest().getHeader();
- final HeartbeatRequestBody heartbeatRequestBody =
(HeartbeatRequestBody) asyncContext.getRequest().getBody();
+ final HeartbeatRequestHeader heartbeatRequestHeader =
(HeartbeatRequestHeader) request.getHeader();
+ final HeartbeatRequestBody heartbeatRequestBody =
(HeartbeatRequestBody) request.getBody();
+ EventMeshHTTPConfiguration httpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
final HeartbeatResponseHeader heartbeatResponseHeader =
-
HeartbeatResponseHeader.buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster(),
- localAddress,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+
HeartbeatResponseHeader.buildHeader(Integer.valueOf(request.getRequestCode()),
+ httpConfiguration.getEventMeshCluster(),
+ localAddress, httpConfiguration.getEventMeshEnv(),
+ httpConfiguration.getEventMeshIDC());
//validate header
- if (StringUtils.isBlank(heartbeatRequestHeader.getIdc())
- || StringUtils.isBlank(heartbeatRequestHeader.getPid())
- || !StringUtils.isNumeric(heartbeatRequestHeader.getPid())
- || StringUtils.isBlank(heartbeatRequestHeader.getSys())) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- heartbeatResponseHeader,
-
HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+
+ if (StringUtils.isAnyBlank(
+ heartbeatRequestHeader.getIdc(),
heartbeatRequestHeader.getPid(), heartbeatRequestHeader.getSys())
+ || !StringUtils.isNumeric(heartbeatRequestHeader.getPid())) {
+ completeResponse(request, asyncContext, heartbeatResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
HeartbeatResponseBody.class);
return;
}
//validate body
- if (StringUtils.isBlank(heartbeatRequestBody.getClientType())
- || StringUtils.isBlank(heartbeatRequestBody.getConsumerGroup())
+ if (StringUtils.isAnyBlank(heartbeatRequestBody.getClientType(),
heartbeatRequestBody.getConsumerGroup())
||
CollectionUtils.isEmpty(heartbeatRequestBody.getHeartbeatEntities())) {
-
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- heartbeatResponseHeader,
-
HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, heartbeatResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
HeartbeatResponseBody.class);
return;
}
final ConcurrentHashMap<String, List<Client>> tmpMap = new
ConcurrentHashMap<>();
@@ -122,7 +116,7 @@ public class HeartBeatProcessor implements
HttpRequestProcessor {
client.setUrl(heartbeatEntity.url);
client.setLastUpTime(new Date());
- if (StringUtils.isBlank(client.getTopic()) ||
StringUtils.isBlank(client.getUrl())) {
+ if (StringUtils.isAnyBlank(client.getTopic(), client.getUrl())) {
continue;
}
@@ -137,11 +131,8 @@ public class HeartBeatProcessor implements
HttpRequestProcessor {
client.getTopic(),
Integer.parseInt(heartbeatRequestHeader.getCode()));
} catch (Exception e) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- heartbeatResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(), e.getMessage()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
heartbeatResponseHeader,
+ EventMeshRetCode.EVENTMESH_ACL_ERR,
e.getMessage(), HeartbeatResponseBody.class);
if (log.isWarnEnabled()) {
log.warn("CLIENT HAS NO PERMISSION,HeartBeatProcessor
subscribe failed", e);
}
@@ -156,23 +147,24 @@ public class HeartBeatProcessor implements
HttpRequestProcessor {
}
- synchronized
(eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping()) {
+ ConcurrentHashMap<String, List<Client>> clientInfoMap =
+
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping();
+ synchronized (clientInfoMap) {
for (final Map.Entry<String, List<Client>> groupTopicClientMapping
: tmpMap.entrySet()) {
- final List<Client> localClientList =
-
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().get(groupTopicClientMapping.getKey());
+ final List<Client> localClientList =
clientInfoMap.get(groupTopicClientMapping.getKey());
if (CollectionUtils.isEmpty(localClientList)) {
-
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping()
- .put(groupTopicClientMapping.getKey(),
groupTopicClientMapping.getValue());
+ clientInfoMap.put(groupTopicClientMapping.getKey(),
groupTopicClientMapping.getValue());
} else {
final List<Client> tmpClientList =
groupTopicClientMapping.getValue();
supplyClientInfoList(tmpClientList, localClientList);
-
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().put(groupTopicClientMapping.getKey(),
localClientList);
+ clientInfoMap.put(groupTopicClientMapping.getKey(),
localClientList);
}
}
}
final long startTime = System.currentTimeMillis();
+ HttpSummaryMetrics summaryMetrics =
eventMeshHTTPServer.getMetrics().getSummaryMetrics();
try {
final CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
@@ -180,26 +172,25 @@ public class HeartBeatProcessor implements
HttpRequestProcessor {
log.debug("{}", httpCommand);
}
eventMeshHTTPServer.sendResponse(ctx,
httpCommand.httpResponse());
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
- System.currentTimeMillis() -
asyncContext.getRequest().getReqTime());
+ summaryMetrics.recordHTTPReqResTimeCost(
+ System.currentTimeMillis() - request.getReqTime());
} catch (Exception ex) {
//ignore
}
};
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
+ responseEventMeshCommand =
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
- final HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
- heartbeatResponseHeader,
-
HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(err);
+ completeResponse(request, asyncContext, heartbeatResponseHeader,
+ EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR,
+ EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2),
+ HeartbeatResponseBody.class);
final long elapsedTime = System.currentTimeMillis() - startTime;
if (log.isErrorEnabled()) {
log.error("message|eventMesh2mq|REQ|ASYNC|heartBeatMessageCost={}ms",
elapsedTime, e);
}
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(elapsedTime);
+ summaryMetrics.recordSendMsgFailed();
+ summaryMetrics.recordSendMsgCost(elapsedTime);
}
}
@@ -222,10 +213,4 @@ public class HeartBeatProcessor implements
HttpRequestProcessor {
}
}
}
-
- @Override
- public boolean rejectRequest() {
- return false;
- }
-
}
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 7b508d69a..d5a408394 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
@@ -25,16 +25,17 @@ import
org.apache.eventmesh.common.protocol.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
import
org.apache.eventmesh.common.protocol.http.body.message.ReplyMessageResponseBody;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
-import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
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.RequestCode;
import
org.apache.eventmesh.common.protocol.http.header.message.ReplyMessageRequestHeader;
import
org.apache.eventmesh.common.protocol.http.header.message.ReplyMessageResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
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.async.CompleteHandler;
@@ -44,11 +45,9 @@ import
org.apache.eventmesh.runtime.core.protocol.http.producer.SendMessageConte
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
+import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
-import java.nio.charset.StandardCharsets;
-import java.util.Objects;
-import java.util.Optional;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
@@ -60,13 +59,13 @@ import io.netty.channel.ChannelHandlerContext;
public class ReplyMessageProcessor implements HttpRequestProcessor {
- public Logger messageLogger =
LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
+ public final Logger messageLogger =
LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
- public Logger cmdLogger = LoggerFactory.getLogger(EventMeshConstants.CMD);
+ public final Logger cmdLogger =
LoggerFactory.getLogger(EventMeshConstants.CMD);
- public Logger httpLogger =
LoggerFactory.getLogger(EventMeshConstants.PROTOCOL_HTTP);
+ public final Logger httpLogger =
LoggerFactory.getLogger(EventMeshConstants.PROTOCOL_HTTP);
- private EventMeshHTTPServer eventMeshHTTPServer;
+ private final EventMeshHTTPServer eventMeshHTTPServer;
public ReplyMessageProcessor(EventMeshHTTPServer eventMeshHTTPServer) {
this.eventMeshHTTPServer = eventMeshHTTPServer;
@@ -76,124 +75,93 @@ public class ReplyMessageProcessor implements
HttpRequestProcessor {
public void processRequest(ChannelHandlerContext ctx,
AsyncContext<HttpCommand> asyncContext) throws Exception {
HttpCommand responseEventMeshCommand;
String localAddress = IPUtils.getLocalAddress();
- cmdLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
+ HttpCommand request = asyncContext.getRequest();
+ cmdLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(Integer.valueOf(request.getRequestCode())),
EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
localAddress);
- ReplyMessageRequestHeader replyMessageRequestHeader =
(ReplyMessageRequestHeader) asyncContext.getRequest().getHeader();
+ ReplyMessageRequestHeader replyMessageRequestHeader =
(ReplyMessageRequestHeader) request.getHeader();
String protocolType = replyMessageRequestHeader.getProtocolType();
ProtocolAdaptor<ProtocolTransportObject> httpCommandProtocolAdaptor =
ProtocolPluginFactory.getProtocolAdaptor(protocolType);
- CloudEvent event =
httpCommandProtocolAdaptor.toCloudEvent(asyncContext.getRequest());
-
+ CloudEvent event = httpCommandProtocolAdaptor.toCloudEvent(request);
+ EventMeshHTTPConfiguration httpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
ReplyMessageResponseHeader replyMessageResponseHeader =
-
ReplyMessageResponseHeader.buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster(),
- localAddress,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+
ReplyMessageResponseHeader.buildHeader(Integer.valueOf(request.getRequestCode()),
+ httpConfiguration.getEventMeshCluster(),
+ localAddress, httpConfiguration.getEventMeshEnv(),
+ httpConfiguration.getEventMeshIDC());
//validate event
- if (event == null
- || StringUtils.isBlank(event.getId())
- || event.getSource() == null
- || event.getSpecVersion() == null
- || StringUtils.isBlank(event.getType())
- || StringUtils.isBlank(event.getSubject())) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ if (!ObjectUtils.allNotNull(event, event.getSource(),
event.getSpecVersion())
+ || StringUtils.isAnyBlank(event.getId(), event.getType(),
event.getSubject())) {
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
ReplyMessageResponseBody.class);
return;
}
- String idc =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC)).toString();
- String pid =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID)).toString();
- String sys =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS)).toString();
+ String idc = getExtension(event, ProtocolKey.ClientInstanceKey.IDC);
+ String pid = getExtension(event, ProtocolKey.ClientInstanceKey.PID);
+ String sys = getExtension(event, ProtocolKey.ClientInstanceKey.SYS);
//validate HEADER
- if (StringUtils.isBlank(idc)
- || StringUtils.isBlank(pid)
- || !StringUtils.isNumeric(pid)
- || StringUtils.isBlank(sys)) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ if (StringUtils.isAnyBlank(idc, pid, sys)
+ || !StringUtils.isNumeric(pid)) {
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
ReplyMessageResponseBody.class);
return;
}
- String bizNo =
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.BIZSEQNO)).toString();
- String uniqueId =
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.UNIQUEID)).toString();
- String producerGroup =
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.PRODUCERGROUP)).toString();
+ String bizNo = getExtension(event, SendMessageRequestBody.BIZSEQNO);
+ String uniqueId = getExtension(event, SendMessageRequestBody.UNIQUEID);
+ String producerGroup = getExtension(event,
SendMessageRequestBody.PRODUCERGROUP);
//validate body
- if (StringUtils.isBlank(bizNo)
- || StringUtils.isBlank(uniqueId)
- || StringUtils.isBlank(producerGroup)
+ if (StringUtils.isAnyBlank(bizNo, uniqueId, producerGroup)
|| event.getData() == null) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
ReplyMessageResponseBody.class);
return;
}
// control flow rate limit
+ HttpSummaryMetrics summaryMetrics =
eventMeshHTTPServer.getMetrics().getSummaryMetrics();
if (!eventMeshHTTPServer.getMsgRateLimiter()
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPDiscard();
- asyncContext.onComplete(responseEventMeshCommand);
+ summaryMetrics.recordHTTPDiscard();
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR,
null, ReplyMessageResponseBody.class);
return;
}
- String content = event.getData() == null ? "" : new
String(event.getData().toBytes(), StandardCharsets.UTF_8);
- if (content.length() >
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()) {
+ String content = event.getData() == null ? "" : new
String(event.getData().toBytes(), Constants.DEFAULT_CHARSET);
+ if (content.length() > httpConfiguration.getEventMeshEventSize()) {
httpLogger.error("Event size exceeds the limit: {}",
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize());
-
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- "Event size exceeds the limit: " +
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()));
- asyncContext.onComplete(responseEventMeshCommand);
+ httpConfiguration.getEventMeshEventSize());
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
+ "Event size exceeds the limit: " +
httpConfiguration.getEventMeshEventSize(),
+ ReplyMessageResponseBody.class);
return;
}
EventMeshProducer eventMeshProducer =
eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);
if (!eventMeshProducer.isStarted()) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR,
null, ReplyMessageResponseBody.class);
return;
}
long startTime = System.currentTimeMillis();
-
String replyTopic = EventMeshConstants.RR_REPLY_TOPIC;
-
String origTopic = event.getSubject();
-
- final String replyMQCluster =
Optional.ofNullable(event.getExtension(EventMeshConstants.PROPERTY_MESSAGE_CLUSTER))
- .map(Objects::toString)
- .orElse("");
- if (!org.apache.commons.lang3.StringUtils.isEmpty(replyMQCluster)) {
+ final String replyMQCluster = getExtension(event,
EventMeshConstants.PROPERTY_MESSAGE_CLUSTER);
+ if (!StringUtils.isEmpty(replyMQCluster)) {
replyTopic = replyMQCluster + "-" + replyTopic;
} else {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR, null,
ReplyMessageResponseBody.class);
return;
}
@@ -213,25 +181,24 @@ public class ReplyMessageProcessor implements
HttpRequestProcessor {
} catch (Exception e) {
messageLogger.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo,
replyTopic, e);
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR,
+ EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2),
+ ReplyMessageResponseBody.class
+ );
return;
}
final SendMessageContext sendMessageContext = new
SendMessageContext(bizNo, event, eventMeshProducer, eventMeshHTTPServer);
- eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordReplyMsg();
-
+ summaryMetrics.recordReplyMsg();
CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
if (httpLogger.isDebugEnabled()) {
httpLogger.debug("{}", httpCommand);
}
eventMeshHTTPServer.sendResponse(ctx,
httpCommand.httpResponse());
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
- System.currentTimeMillis() -
asyncContext.getRequest().getReqTime());
+ summaryMetrics.recordHTTPReqResTimeCost(
+ System.currentTimeMillis() - request.getReqTime());
} catch (Exception ex) {
//ignore
}
@@ -245,12 +212,12 @@ public class ReplyMessageProcessor implements
HttpRequestProcessor {
eventMeshProducer.reply(sendMessageContext, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
- HttpCommand succ =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(),
EventMeshRetCode.SUCCESS.getErrMsg()));
+ HttpCommand succ = request.createHttpCommandResponse(
+ replyMessageResponseHeader,
+
ReplyMessageResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(),
EventMeshRetCode.SUCCESS.getErrMsg()));
asyncContext.onComplete(succ, handler);
long endTime = System.currentTimeMillis();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordReplyMsgCost(endTime
- startTime);
+ summaryMetrics.recordReplyMsgCost(endTime - startTime);
messageLogger.info("message|eventMesh2mq|RSP|SYNC|reply2MQCost={}|topic={}|origTopic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, replyMQCluster + "-" +
EventMeshConstants.RR_REPLY_TOPIC,
origTopic, bizNo, uniqueId);
@@ -258,36 +225,30 @@ public class ReplyMessageProcessor implements
HttpRequestProcessor {
@Override
public void onException(OnExceptionContext context) {
- HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getRetCode(),
+ HttpCommand err = request.createHttpCommandResponse(
+ replyMessageResponseHeader,
+
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getRetCode(),
EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getErrMsg()
+
EventMeshUtil.stackTrace(context.getException(), 2)));
asyncContext.onComplete(err, handler);
long endTime = System.currentTimeMillis();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordReplyMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordReplyMsgCost(endTime
- startTime);
+ summaryMetrics.recordReplyMsgFailed();
+ summaryMetrics.recordReplyMsgCost(endTime - startTime);
messageLogger.error("message|eventMesh2mq|RSP|SYNC|reply2MQCost={}|topic={}|origTopic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, replyMQCluster + "-" +
EventMeshConstants.RR_REPLY_TOPIC,
origTopic, bizNo, uniqueId, context.getException());
}
});
} catch (Exception ex) {
- HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getErrMsg() +
EventMeshUtil.stackTrace(ex, 2)));
- asyncContext.onComplete(err);
+ completeResponse(request, asyncContext, replyMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR,
+ EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getErrMsg() +
EventMeshUtil.stackTrace(ex, 2),
+ ReplyMessageResponseBody.class);
long endTime = System.currentTimeMillis();
messageLogger.error("message|eventMesh2mq|RSP|SYNC|reply2MQCost={}|topic={}|origTopic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, replyTopic, origTopic, bizNo, uniqueId,
ex);
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordReplyMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordReplyMsgCost(endTime
- startTime);
+ summaryMetrics.recordReplyMsgFailed();
+ summaryMetrics.recordReplyMsgCost(endTime - startTime);
}
}
-
- @Override
- public boolean rejectRequest() {
- return false;
- }
}
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 e42070061..f90eacee2 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
@@ -20,6 +20,7 @@ package
org.apache.eventmesh.runtime.core.protocol.http.processor;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
@@ -50,7 +51,6 @@ import
org.apache.eventmesh.trace.api.common.EventMeshTraceConstants;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
-import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
@@ -62,7 +62,6 @@ import io.cloudevents.core.builder.CloudEventBuilder;
import io.netty.channel.ChannelHandlerContext;
import io.opentelemetry.api.trace.Span;
-
public class SendAsyncMessageProcessor implements HttpRequestProcessor {
private static final Logger MESSAGE_LOGGER =
LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
@@ -91,7 +90,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
CMD_LOGGER.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(
Integer.valueOf(request.getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP,
+ EventMeshConstants.PROTOCOL_HTTP,
remoteAddr, localAddress);
SendMessageRequestHeader sendMessageRequestHeader =
(SendMessageRequestHeader) request.getHeader();
@@ -110,69 +109,55 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
CloudEvent event = httpCommandProtocolAdaptor.toCloudEvent(request);
Span span =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
true);
+ EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
true);
//validate event
if (!ObjectUtils.allNotNull(event, event.getSource(),
event.getSpecVersion())
|| StringUtils.isAnyBlank(event.getId(), event.getType(),
event.getSubject())) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SendMessageResponseBody.class);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR);
return;
}
- String idc =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC)).toString();
- String pid =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID)).toString();
- String sys =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS)).toString();
+ String idc = getExtension(event, ProtocolKey.ClientInstanceKey.IDC);
+ String pid = getExtension(event, ProtocolKey.ClientInstanceKey.PID);
+ String sys = getExtension(event, ProtocolKey.ClientInstanceKey.SYS);
//validate event-extension
if (StringUtils.isAnyBlank(idc, pid, sys)
|| !StringUtils.isNumeric(pid)) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SendMessageResponseBody.class);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR);
return;
}
- String bizNo =
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.BIZSEQNO)).toString();
- String uniqueId =
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.UNIQUEID)).toString();
- String producerGroup =
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.PRODUCERGROUP)).toString();
+ String bizNo = getExtension(event, SendMessageRequestBody.BIZSEQNO);
+ String uniqueId = getExtension(event, SendMessageRequestBody.UNIQUEID);
+ String producerGroup = getExtension(event,
SendMessageRequestBody.PRODUCERGROUP);
String topic = event.getSubject();
//validate body
if (StringUtils.isAnyBlank(bizNo, uniqueId, producerGroup, topic)
|| event.getData() == null) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
SendMessageResponseBody.class);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR);
return;
}
//do acl check
if (eventMeshHttpConfiguration.isEventMeshServerSecurityEnable()) {
- String user =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.USERNAME)).toString();
- String pass =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PASSWD)).toString();
- String subsystem =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS)).toString();
+ String user = getExtension(event,
ProtocolKey.ClientInstanceKey.USERNAME);
+ String pass = getExtension(event,
ProtocolKey.ClientInstanceKey.PASSWD);
+ String subsystem = getExtension(event,
ProtocolKey.ClientInstanceKey.SYS);
int requestCode = Integer.parseInt(request.getRequestCode());
try {
this.acl.doAclCheckInHttpSend(remoteAddr, user, pass,
subsystem, topic, requestCode);
} catch (Exception e) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(),
e.getMessage()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_ACL_ERR, e.getMessage(),
SendMessageResponseBody.class);
ACL_LOGGER.warn("CLIENT HAS NO
PERMISSION,SendAsyncMessageProcessor send failed", e);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_ACL_ERR);
@@ -184,12 +169,9 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
// control flow rate limit
if (!eventMeshHTTPServer.getMsgRateLimiter()
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR,
null, SendMessageResponseBody.class);
summaryMetrics.recordHTTPDiscard();
- asyncContext.onComplete(responseEventMeshCommand);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR);
return;
@@ -198,34 +180,27 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
EventMeshProducer eventMeshProducer =
eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);
if (!eventMeshProducer.isStarted()) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR,
null, SendMessageResponseBody.class);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR);
return;
}
String ttl = String.valueOf(EventMeshConstants.DEFAULT_MSG_TTL_MILLS);
- String ttlExt =
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.TTL)).toString();
+ String ttlExt = getExtension(event, SendMessageRequestBody.TTL);
if (StringUtils.isBlank(ttlExt) && !StringUtils.isNumeric(ttlExt)) {
event =
CloudEventBuilder.from(event).withExtension(SendMessageRequestBody.TTL,
ttl).build();
}
- String content = event.getData() == null ? "" : new
String(Objects.requireNonNull(event.getData()).toBytes(),
StandardCharsets.UTF_8);
+ String content = event.getData() == null ? "" : new
String(Objects.requireNonNull(event.getData()).toBytes(),
Constants.DEFAULT_CHARSET);
if (content.length() >
eventMeshHttpConfiguration.getEventMeshEventSize()) {
HTTP_LOGGER.error("Event size exceeds the limit: {}",
eventMeshHttpConfiguration.getEventMeshEventSize());
-
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_SIZE_ERR.getRetCode(),
- "Event size exceeds the limit: " +
eventMeshHttpConfiguration.getEventMeshEventSize()));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_SIZE_ERR,
+ "Event size exceeds the limit: " +
eventMeshHttpConfiguration.getEventMeshEventSize(),
+ SendMessageResponseBody.class);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_SIZE_ERR);
return;
}
@@ -234,8 +209,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
event = CloudEventBuilder.from(event)
.withExtension(EventMeshConstants.MSG_TYPE,
EventMeshConstants.PERSISTENT)
.withExtension(EventMeshConstants.REQ_C2EVENTMESH_TIMESTAMP,
request.reqTime)
- .withExtension(EventMeshConstants.REQ_SEND_EVENTMESH_IP,
- eventMeshHttpConfiguration.getEventMeshServerIp())
+ .withExtension(EventMeshConstants.REQ_SEND_EVENTMESH_IP,
eventMeshHttpConfiguration.getEventMeshServerIp())
.build();
if (MESSAGE_LOGGER.isDebugEnabled()) {
@@ -243,12 +217,10 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
}
} catch (Exception e) {
MESSAGE_LOGGER.error("msg2MQMsg err, bizSeqNo={}, topic={}",
bizNo, topic, e);
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR,
+ EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2),
+ SendMessageResponseBody.class);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR);
return;
}
@@ -326,13 +298,8 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
} catch (Exception ex) {
- HttpCommand err = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(ex, 2)));
- asyncContext.onComplete(err);
-
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR, null,
SendMessageResponseBody.class);
spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR);
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
@@ -350,10 +317,4 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
retCode.getErrMsg(), null);
}
-
- @Override
- public boolean rejectRequest() {
- return false;
- }
-
}
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 e7541ea5a..12ba0c32a 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
@@ -17,7 +17,9 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
+
import org.apache.eventmesh.api.RequestReplyCallback;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
@@ -47,7 +49,6 @@ import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
-import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
@@ -103,71 +104,46 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
//validate event
if (!ObjectUtils.allNotNull(event, event.getSource(),
event.getSpecVersion())
|| StringUtils.isAnyBlank(event.getId(), event.getType(),
event.getSubject())) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SendMessageResponseBody.class);
return;
}
- final String idc =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC))
- .toString();
- final String pid =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID))
- .toString();
- final String sys =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS))
- .toString();
+ final String idc = getExtension(event,
ProtocolKey.ClientInstanceKey.IDC);
+ final String pid = getExtension(event,
ProtocolKey.ClientInstanceKey.PID);
+ final String sys = getExtension(event,
ProtocolKey.ClientInstanceKey.SYS);
//validate event-extension
if (StringUtils.isAnyBlank(idc, pid, sys) ||
!StringUtils.isNumeric(pid)) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SendMessageResponseBody.class);
return;
}
- final String bizNo =
-
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.BIZSEQNO)).toString();
- final String uniqueId =
-
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.UNIQUEID)).toString();
- final String producerGroup =
-
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.PRODUCERGROUP))
- .toString();
+ final String bizNo = getExtension(event,
SendMessageRequestBody.BIZSEQNO);
+ final String uniqueId = getExtension(event,
SendMessageRequestBody.UNIQUEID);
+ final String producerGroup = getExtension(event,
SendMessageRequestBody.PRODUCERGROUP);
final String topic = event.getSubject();
- final String ttl =
-
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.TTL)).toString();
+ final String ttl = getExtension(event, SendMessageRequestBody.TTL);
//validate body
if (StringUtils.isAnyBlank(bizNo, uniqueId, producerGroup, topic, ttl)
|| event.getData() == null) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
SendMessageResponseBody.class);
return;
}
//do acl check
if (eventMeshHttpConfiguration.isEventMeshServerSecurityEnable()) {
- final String user =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.USERNAME)).toString();
- final String pass =
Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PASSWD)).toString();
+ final String user = getExtension(event,
ProtocolKey.ClientInstanceKey.USERNAME);
+ final String pass = getExtension(event,
ProtocolKey.ClientInstanceKey.PASSWD);
final int requestCode = Integer.parseInt(request.getRequestCode());
try {
this.acl.doAclCheckInHttpSend(remoteAddr, user, pass, sys,
topic, requestCode);
} catch (Exception e) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(),
- e.getMessage()));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext,
sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_ACL_ERR, e.getMessage(),
SendMessageResponseBody.class);
if (log.isWarnEnabled()) {
log.warn("CLIENT HAS NO
PERMISSION,SendSyncMessageProcessor send failed", e);
}
@@ -178,30 +154,23 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
final HttpSummaryMetrics summaryMetrics =
eventMeshHTTPServer.getMetrics().getSummaryMetrics();
// control flow rate limit
if (!eventMeshHTTPServer.getMsgRateLimiter()
-
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
- TimeUnit.MILLISECONDS)) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
+
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR,
null, SendMessageResponseBody.class);
summaryMetrics.recordHTTPDiscard();
- asyncContext.onComplete(responseEventMeshCommand);
return;
}
- final String content = new
String(Objects.requireNonNull(event.getData()).toBytes(),
StandardCharsets.UTF_8);
+ final String content = new
String(Objects.requireNonNull(event.getData()).toBytes(),
Constants.DEFAULT_CHARSET);
int eventMeshEventSize =
eventMeshHttpConfiguration.getEventMeshEventSize();
if (content.length() > eventMeshEventSize) {
if (log.isErrorEnabled()) {
log.error("Event size exceeds the limit: {}",
eventMeshEventSize);
}
-
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- "Event size exceeds the limit: " + eventMeshEventSize));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
+ "Event size exceeds the limit: " + eventMeshEventSize,
+ SendMessageResponseBody.class);
return;
}
@@ -209,12 +178,8 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);
if (!eventMeshProducer.isStarted()) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR,
null, SendMessageResponseBody.class);
return;
}
@@ -234,13 +199,8 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
if (log.isErrorEnabled()) {
log.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo,
topic, e);
}
- responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR, null,
SendMessageResponseBody.class);
return;
}
@@ -283,7 +243,7 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
.build();
final String rtnMsg = new
String(Objects.requireNonNull(newEvent.getData()).toBytes(),
- StandardCharsets.UTF_8);
+ Constants.DEFAULT_CHARSET);
final HttpCommand succ =
request.createHttpCommandResponse(
sendMessageResponseHeader,
@@ -329,14 +289,10 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
}
}, Integer.parseInt(ttl));
} catch (Exception ex) {
- final HttpCommand err = request.createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(ex, 2)));
- asyncContext.onComplete(err);
-
+ completeResponse(request, asyncContext, sendMessageResponseHeader,
+ EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR,
+ EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getErrMsg() +
EventMeshUtil.stackTrace(ex, 2),
+ SendMessageResponseBody.class);
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10_000));
final long endTime = System.currentTimeMillis();
summaryMetrics.recordSendMsgFailed();
@@ -351,9 +307,4 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
return;
}
-
- @Override
- public boolean rejectRequest() {
- return false;
- }
}
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 b79a6d2f8..d93ba6134 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
@@ -21,7 +21,6 @@ import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
import
org.apache.eventmesh.common.protocol.http.body.client.SubscribeRequestBody;
import
org.apache.eventmesh.common.protocol.http.body.client.SubscribeResponseBody;
-import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import
org.apache.eventmesh.common.protocol.http.header.client.SubscribeRequestHeader;
@@ -74,8 +73,7 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
final String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
if (log.isInfoEnabled()) {
log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
- RequestCode.get(requestCode),
- EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
+ RequestCode.get(requestCode),
EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
}
final SubscribeRequestHeader subscribeRequestHeader =
(SubscribeRequestHeader) request.getHeader();
@@ -93,24 +91,16 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
if (StringUtils.isAnyBlank(subscribeRequestHeader.getIdc(),
subscribeRequestHeader.getPid(),
subscribeRequestHeader.getSys())
|| !StringUtils.isNumeric(subscribeRequestHeader.getPid())) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- subscribeResponseHeader,
- SubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, subscribeResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR, null,
SubscribeResponseBody.class);
return;
}
//validate body
if (StringUtils.isAnyBlank(subscribeRequestBody.getUrl(),
subscribeRequestBody.getConsumerGroup())
|| CollectionUtils.isEmpty(subscribeRequestBody.getTopics())) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- subscribeResponseHeader,
- SubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, subscribeResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
SubscribeResponseBody.class);
return;
}
final List<SubscriptionItem> subTopicList =
subscribeRequestBody.getTopics();
@@ -125,11 +115,8 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
subscribeRequestHeader.getSys(), item.getTopic(),
requestCode);
} catch (Exception e) {
-
- responseEventMeshCommand =
request.createHttpCommandResponse(subscribeResponseHeader,
-
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(),
e.getMessage()));
- asyncContext.onComplete(responseEventMeshCommand);
-
+ completeResponse(request, asyncContext,
subscribeResponseHeader,
+ EventMeshRetCode.EVENTMESH_ACL_ERR,
e.getMessage(), SubscribeResponseBody.class);
if (log.isWarnEnabled()) {
log.warn("CLIENT HAS NO PERMISSION,SubscribeProcessor
subscribe failed", e);
}
@@ -146,24 +133,20 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
if (!IPUtils.isValidDomainOrIp(url,
eventMeshHttpConfiguration.getEventMeshIpv4BlackList(),
eventMeshHttpConfiguration.getEventMeshIpv6BlackList())) {
log.error("subscriber url {} is not valid", url);
- responseEventMeshCommand = request.createHttpCommandResponse(
- subscribeResponseHeader,
- SubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " invalid URL: " +
url));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext,
subscribeResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
+
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " invalid URL: " +
url,
+ SubscribeResponseBody.class);
return;
}
} catch (Exception e) {
if (log.isErrorEnabled()) {
log.error("subscriber url:{} is invalid.", url, e);
}
- responseEventMeshCommand = request.createHttpCommandResponse(
- subscribeResponseHeader,
- SubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " invalid URL: " +
url));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, subscribeResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() +
" invalid URL: " + url,
+ SubscribeResponseBody.class);
return;
}
@@ -171,12 +154,10 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
if
(!WebhookUtil.obtainDeliveryAgreement(eventMeshHTTPServer.getHttpClientPool().getClient(),
url, eventMeshHttpConfiguration.getEventMeshWebhookOrigin())) {
log.error("subscriber url {} is not allowed by the target system",
url);
- responseEventMeshCommand = request.createHttpCommandResponse(
- subscribeResponseHeader,
- SubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " unauthorized
webhook URL: " + url));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, subscribeResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() +
" unauthorized webhook URL: " + url,
+ SubscribeResponseBody.class);
return;
}
@@ -209,14 +190,10 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
responseEventMeshCommand =
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
- final HttpCommand err = request.createHttpCommandResponse(
- subscribeResponseHeader,
- SubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(err);
-
+ completeResponse(request, asyncContext,
subscribeResponseHeader,
+ EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR,
+ EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2),
+ SubscribeResponseBody.class);
final long endTime = System.currentTimeMillis();
if (log.isErrorEnabled()) {
log.error(
@@ -231,11 +208,6 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
}
}
- @Override
- public boolean rejectRequest() {
- return false;
- }
-
private ClientInfo getClientInfo(final SubscribeRequestHeader
subscribeRequestHeader) {
ClientInfo clientInfo = new ClientInfo();
clientInfo.setEnv(subscribeRequestHeader.getEnv());
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 d5c72ee51..ecc5a0980 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
@@ -76,7 +76,7 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
if (log.isInfoEnabled()) {
log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(Integer.valueOf(request.getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP,
+ EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
localAddress);
}
@@ -94,24 +94,16 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
if (StringUtils.isAnyBlank(unSubscribeRequestHeader.getIdc(),
unSubscribeRequestHeader.getPid(),
unSubscribeRequestHeader.getSys())
|| !StringUtils.isNumeric(unSubscribeRequestHeader.getPid())) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- unSubscribeResponseHeader,
- UnSubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ 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())) {
- responseEventMeshCommand = request.createHttpCommandResponse(
- unSubscribeResponseHeader,
- UnSubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
- asyncContext.onComplete(responseEventMeshCommand);
+ completeResponse(request, asyncContext, unSubscribeResponseHeader,
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, null,
UnSubscribeResponseBody.class);
return;
}
@@ -198,18 +190,14 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
localConsumerGroupMap.get(consumerGroup));
- responseEventMeshCommand =
-
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
+ responseEventMeshCommand =
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
- final HttpCommand err = request.createHttpCommandResponse(
- unSubscribeResponseHeader,
- UnSubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(err);
+ 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,
@@ -233,13 +221,10 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
localConsumerGroupMap.keySet()
.removeIf(s -> StringUtils.equals(consumerGroup, s));
} catch (Exception e) {
- final HttpCommand err = request.createHttpCommandResponse(
- unSubscribeResponseHeader,
- UnSubscribeResponseBody
-
.buildBody(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode(),
-
EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(e, 2)));
- asyncContext.onComplete(err);
+ 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,
@@ -252,11 +237,6 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
}
}
- @Override
- public boolean rejectRequest() {
- return false;
- }
-
private void registerClient(final UnSubscribeRequestHeader
unSubscribeRequestHeader,
final String consumerGroup,
final List<String> topicList,
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 f9ffec041..772503928 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
@@ -18,8 +18,17 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor.inf;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
+import org.apache.eventmesh.common.protocol.http.body.Body;
+import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
+import org.apache.eventmesh.common.protocol.http.header.Header;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
+import org.apache.commons.lang3.StringUtils;
+
+import java.lang.reflect.Method;
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
import io.netty.channel.ChannelHandlerContext;
/**
@@ -30,6 +39,27 @@ public interface HttpRequestProcessor {
void processRequest(final ChannelHandlerContext ctx, final
AsyncContext<HttpCommand> asyncContext)
throws Exception;
- boolean rejectRequest();
+ default boolean rejectRequest() {
+ return false;
+ }
+
+ default <T extends Header, E extends Body> void
completeResponse(HttpCommand req, AsyncContext asyncContext,
+ T
respHeader, EventMeshRetCode emCode,
+ String
msg, Class<E> clazz) {
+ try {
+ Method method = clazz.getMethod("buildBody", Integer.class,
String.class);
+ Object o = method.invoke(null, emCode.getRetCode(),
+ StringUtils.isNotBlank(msg) ? msg : emCode.getErrMsg());
+ HttpCommand response = req.createHttpCommandResponse(respHeader,
(E) o);
+ asyncContext.onComplete(response);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ default String getExtension(CloudEvent event, String protocolKey) {
+ Object extension = event.getExtension(protocolKey);
+ return Objects.isNull(extension) ? "" : extension.toString();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
