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 c28c4e88d [ISSUE #3733] Invoke method(s) only
conditionally.[SendAsyncMessageProcessor]
c28c4e88d is described below
commit c28c4e88d8cf6c5bba85975086435bfd83c588c9
Author: pandaapo <[email protected]>
AuthorDate: Wed Apr 19 14:59:36 2023 +0800
[ISSUE #3733] Invoke method(s) only
conditionally.[SendAsyncMessageProcessor]
* Fix issue: reuse variable.
* Invoke method(s) only conditionally.
---
.../http/processor/SendAsyncMessageProcessor.java | 83 +++++++----------
.../http/processor/SendSyncMessageProcessor.java | 99 +++++++++-----------
.../http/processor/SubscribeProcessor.java | 59 ++++++------
.../http/processor/UnSubscribeProcessor.java | 102 ++++++++++-----------
4 files changed, 154 insertions(+), 189 deletions(-)
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 bbe4bdcaf..e42070061 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
@@ -30,6 +30,7 @@ import
org.apache.eventmesh.common.protocol.http.common.RequestCode;
import
org.apache.eventmesh.common.protocol.http.header.message.SendMessageRequestHeader;
import
org.apache.eventmesh.common.protocol.http.header.message.SendMessageResponseHeader;
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;
@@ -46,6 +47,7 @@ import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
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;
@@ -86,10 +88,11 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
HttpCommand responseEventMeshCommand;
String localAddress = IPUtils.getLocalAddress();
HttpCommand request = asyncContext.getRequest();
+ String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
CMD_LOGGER.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(
Integer.valueOf(request.getRequestCode())),
EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
localAddress);
+ remoteAddr, localAddress);
SendMessageRequestHeader sendMessageRequestHeader =
(SendMessageRequestHeader) request.getHeader();
@@ -110,9 +113,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
true);
//validate event
- if (event == null
- || event.getSource() == null
- || event.getSpecVersion() == null
+ if (!ObjectUtils.allNotNull(event, event.getSource(),
event.getSpecVersion())
|| StringUtils.isAnyBlank(event.getId(), event.getType(),
event.getSubject())) {
responseEventMeshCommand = request.createHttpCommandResponse(
sendMessageResponseHeader,
@@ -120,10 +121,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
- EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg(),
null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR);
return;
}
@@ -140,10 +138,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
- EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg(),
null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR);
return;
}
@@ -161,16 +156,12 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
- EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg(),
null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR);
return;
}
//do acl check
if (eventMeshHttpConfiguration.isEventMeshServerSecurityEnable()) {
- String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
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();
@@ -184,14 +175,12 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
asyncContext.onComplete(responseEventMeshCommand);
ACL_LOGGER.warn("CLIENT HAS NO
PERMISSION,SendAsyncMessageProcessor send failed", e);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
-
EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN, false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
- EventMeshRetCode.EVENTMESH_ACL_ERR.getErrMsg(), null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_ACL_ERR);
return;
}
}
+ final HttpSummaryMetrics summaryMetrics =
eventMeshHTTPServer.getMetrics().getSummaryMetrics();
// control flow rate limit
if (!eventMeshHTTPServer.getMsgRateLimiter()
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
@@ -199,13 +188,10 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
sendMessageResponseHeader,
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPDiscard();
+ summaryMetrics.recordHTTPDiscard();
asyncContext.onComplete(responseEventMeshCommand);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
-
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg(), null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR);
return;
}
@@ -218,17 +204,14 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
-
EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getErrMsg(), null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR);
return;
}
String ttl = String.valueOf(EventMeshConstants.DEFAULT_MSG_TTL_MILLS);
- if
(StringUtils.isBlank(Objects.requireNonNull(event.getExtension(SendMessageRequestBody.TTL)).toString())
- &&
!StringUtils.isNumeric(Objects.requireNonNull(event.getExtension(SendMessageRequestBody.TTL)).toString()))
{
+ String ttlExt =
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.TTL)).toString();
+ if (StringUtils.isBlank(ttlExt) && !StringUtils.isNumeric(ttlExt)) {
event =
CloudEventBuilder.from(event).withExtension(SendMessageRequestBody.TTL,
ttl).build();
}
@@ -243,10 +226,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
"Event size exceeds the limit: " +
eventMeshHttpConfiguration.getEventMeshEventSize()));
asyncContext.onComplete(responseEventMeshCommand);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
- EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_SIZE_ERR.getErrMsg(),
null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_SIZE_ERR);
return;
}
@@ -269,16 +249,13 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2)));
asyncContext.onComplete(responseEventMeshCommand);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
- EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg(), null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR);
return;
}
final SendMessageContext sendMessageContext = new
SendMessageContext(bizNo, event, eventMeshProducer,
eventMeshHTTPServer);
- eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsg();
+ summaryMetrics.recordSendMsg();
long startTime = System.currentTimeMillis();
@@ -289,7 +266,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
}
eventMeshHTTPServer.sendResponse(ctx,
httpCommand.httpResponse());
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
+ summaryMetrics.recordHTTPReqResTimeCost(
System.currentTimeMillis() - request.getReqTime());
} catch (Exception ex) {
//ignore
@@ -315,7 +292,7 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
EventMeshRetCode.SUCCESS.getErrMsg() +
sendResult.toString()));
asyncContext.onComplete(succ, handler);
long endTime = System.currentTimeMillis();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ summaryMetrics.recordSendMsgCost(endTime - startTime);
MESSAGE_LOGGER.info("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, bizNo, uniqueId);
@@ -333,8 +310,8 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
long endTime = System.currentTimeMillis();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ summaryMetrics.recordSendMsgFailed();
+ summaryMetrics.recordSendMsgCost(endTime - startTime);
MESSAGE_LOGGER.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, bizNo, uniqueId,
context.getException());
@@ -356,20 +333,24 @@ public class SendAsyncMessageProcessor implements
HttpRequestProcessor {
+ EventMeshUtil.stackTrace(ex, 2)));
asyncContext.onComplete(err);
- Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
- EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
- TraceUtils.finishSpanWithException(excepSpan,
EventMeshUtil.getCloudEventExtensionMap(protocolVersion, event),
- EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getErrMsg(),
null);
+ spanWithException(event, protocolVersion,
EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR);
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
long endTime = System.currentTimeMillis();
MESSAGE_LOGGER.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, bizNo, uniqueId, ex);
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ summaryMetrics.recordSendMsgFailed();
+ summaryMetrics.recordSendMsgCost(endTime - startTime);
}
}
+ private void spanWithException(CloudEvent event, String protocolVersion,
EventMeshRetCode retCode) {
+ Span excepSpan =
TraceUtils.prepareServerSpan(EventMeshUtil.getCloudEventExtensionMap(protocolVersion,
event),
+ EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
false);
+ 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 4e0430a73..f93fb7855 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,6 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
-
import org.apache.eventmesh.api.RequestReplyCallback;
import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
@@ -30,10 +29,12 @@ import
org.apache.eventmesh.common.protocol.http.header.message.SendMessageReque
import
org.apache.eventmesh.common.protocol.http.header.message.SendMessageResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+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.async.CompleteHandler;
@@ -43,6 +44,7 @@ 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;
@@ -72,17 +74,16 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
throws Exception {
HttpCommand responseEventMeshCommand;
-
+ HttpCommand request = asyncContext.getRequest();
final String localAddress = IPUtils.getLocalAddress();
-
+ final String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
if (log.isInfoEnabled()) {
log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
-
RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
localAddress);
+ RequestCode.get(Integer.valueOf(request.getRequestCode())),
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
}
- SendMessageRequestHeader sendMessageRequestHeader =
(SendMessageRequestHeader) asyncContext.getRequest().getHeader();
+ SendMessageRequestHeader sendMessageRequestHeader =
(SendMessageRequestHeader) request.getHeader();
String protocolType = sendMessageRequestHeader.getProtocolType();
@@ -90,22 +91,19 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
final CloudEvent event =
httpCommandProtocolAdaptor.toCloudEvent(asyncContext.getRequest());
+ EventMeshHTTPConfiguration eventMeshHttpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
final SendMessageResponseHeader sendMessageResponseHeader =
SendMessageResponseHeader
-
.buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster(),
- localAddress,
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+ .buildHeader(Integer.valueOf(request.getRequestCode()),
+ eventMeshHttpConfiguration.getEventMeshCluster(),
+ localAddress,
+ eventMeshHttpConfiguration.getEventMeshEnv(),
+ eventMeshHttpConfiguration.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(
+ 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(),
@@ -122,11 +120,8 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
.toString();
//validate event-extension
- if (StringUtils.isBlank(idc)
- || StringUtils.isBlank(pid)
- || !StringUtils.isNumeric(pid)
- || StringUtils.isBlank(sys)) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
+ if (StringUtils.isAnyBlank(idc, pid, sys) ||
!StringUtils.isNumeric(pid)) {
+ responseEventMeshCommand = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
@@ -147,13 +142,8 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
Objects.requireNonNull(event.getExtension(SendMessageRequestBody.TTL)).toString();
//validate body
- if (StringUtils.isBlank(bizNo)
- || StringUtils.isBlank(uniqueId)
- || StringUtils.isBlank(producerGroup)
- || StringUtils.isBlank(topic)
- || event.getData() == null
- || StringUtils.isBlank(ttl)) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
+ if (StringUtils.isAnyBlank(bizNo, uniqueId, producerGroup, topic, ttl)
|| event.getData() == null) {
+ responseEventMeshCommand = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
@@ -163,17 +153,15 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
}
//do acl check
- if
(eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable())
{
- final String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
+ 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 int requestCode =
Integer.parseInt(asyncContext.getRequest().getRequestCode());
+ final int requestCode = Integer.parseInt(request.getRequestCode());
try {
this.acl.doAclCheckInHttpSend(remoteAddr, user, pass, sys,
topic, requestCode);
} catch (Exception e) {
-
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
+ responseEventMeshCommand = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(),
@@ -187,31 +175,32 @@ 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 =
asyncContext.getRequest().createHttpCommandResponse(
+ responseEventMeshCommand = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPDiscard();
+ summaryMetrics.recordHTTPDiscard();
asyncContext.onComplete(responseEventMeshCommand);
return;
}
final String content = new
String(Objects.requireNonNull(event.getData()).toBytes(),
StandardCharsets.UTF_8);
- if (content.length() >
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()) {
+ int eventMeshEventSize =
eventMeshHttpConfiguration.getEventMeshEventSize();
+ if (content.length() > eventMeshEventSize) {
if (log.isErrorEnabled()) {
- log.error("Event size exceeds the limit: {}",
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize());
+ log.error("Event size exceeds the limit: {}",
eventMeshEventSize);
}
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
+ responseEventMeshCommand = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
- "Event size exceeds the limit: " +
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()));
+ "Event size exceeds the limit: " + eventMeshEventSize));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -220,7 +209,7 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);
if (!eventMeshProducer.isStarted()) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
+ responseEventMeshCommand = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_GROUP_PRODUCER_STOPED_ERR.getRetCode(),
@@ -245,7 +234,7 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
if (log.isErrorEnabled()) {
log.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo,
topic, e);
}
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
+ responseEventMeshCommand = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(),
@@ -256,9 +245,8 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
}
final SendMessageContext sendMessageContext =
- new SendMessageContext(bizNo, newEevent, eventMeshProducer,
- eventMeshHTTPServer);
- eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsg();
+ new SendMessageContext(bizNo, newEevent, eventMeshProducer,
eventMeshHTTPServer);
+ summaryMetrics.recordSendMsg();
final long startTime = System.currentTimeMillis();
@@ -270,8 +258,7 @@ public class SendSyncMessageProcessor 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) {
log.error("onResponse error", ex);
// ignore
@@ -300,7 +287,7 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
final String rtnMsg = new
String(Objects.requireNonNull(newEvent.getData()).toBytes(),
StandardCharsets.UTF_8);
- final HttpCommand succ =
asyncContext.getRequest().createHttpCommandResponse(
+ final HttpCommand succ =
request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(),
JsonUtils.toJSONString(SendMessageResponseBody.ReplyMessage.builder()
@@ -310,7 +297,7 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
.build())));
asyncContext.onComplete(succ, handler);
} catch (Exception ex) {
- final HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
+ final HttpCommand err =
request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody.buildBody(
EventMeshRetCode.EVENTMESH_WAITING_RR_MSG_ERR.getRetCode(),
@@ -326,7 +313,7 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
@Override
public void onException(final Throwable e) {
- final HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
+ final HttpCommand err = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_WAITING_RR_MSG_ERR.getRetCode(),
@@ -344,7 +331,7 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
}
}, Integer.parseInt(ttl));
} catch (Exception ex) {
- final HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
+ final HttpCommand err = request.createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getRetCode(),
@@ -354,8 +341,8 @@ public class SendSyncMessageProcessor implements
HttpRequestProcessor {
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10_000));
final long endTime = System.currentTimeMillis();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ summaryMetrics.recordSendMsgFailed();
+ summaryMetrics.recordSendMsgCost(endTime - startTime);
if (log.isErrorEnabled()) {
log.error(
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 352ce0dc8..b79a6d2f8 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
@@ -28,8 +28,10 @@ import
org.apache.eventmesh.common.protocol.http.header.client.SubscribeRequestH
import
org.apache.eventmesh.common.protocol.http.header.client.SubscribeResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+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.consumer.ClientInfo;
import org.apache.eventmesh.runtime.core.consumer.SubscriptionManager;
@@ -67,31 +69,30 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
throws Exception {
HttpCommand responseEventMeshCommand;
final HttpCommand request = asyncContext.getRequest();
- final Integer requestCode =
Integer.valueOf(asyncContext.getRequest().getRequestCode());
+ final Integer requestCode = Integer.valueOf(request.getRequestCode());
final String localAddress = IPUtils.getLocalAddress();
-
+ final String remoteAddr =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
if (log.isInfoEnabled()) {
log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
RequestCode.get(requestCode),
- EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
localAddress);
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
}
final SubscribeRequestHeader subscribeRequestHeader =
(SubscribeRequestHeader) request.getHeader();
final SubscribeRequestBody subscribeRequestBody =
(SubscribeRequestBody) request.getBody();
+ EventMeshHTTPConfiguration eventMeshHttpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
final SubscribeResponseHeader subscribeResponseHeader =
SubscribeResponseHeader
.buildHeader(requestCode,
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster(),
- localAddress,
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+ eventMeshHttpConfiguration.getEventMeshCluster(),
+ localAddress,
+ eventMeshHttpConfiguration.getEventMeshEnv(),
+ eventMeshHttpConfiguration.getEventMeshIDC());
//validate header
- if (StringUtils.isBlank(subscribeRequestHeader.getIdc())
- || StringUtils.isBlank(subscribeRequestHeader.getPid())
- || !StringUtils.isNumeric(subscribeRequestHeader.getPid())
- || StringUtils.isBlank(subscribeRequestHeader.getSys())) {
+ if (StringUtils.isAnyBlank(subscribeRequestHeader.getIdc(),
+ subscribeRequestHeader.getPid(),
subscribeRequestHeader.getSys())
+ || !StringUtils.isNumeric(subscribeRequestHeader.getPid())) {
responseEventMeshCommand = request.createHttpCommandResponse(
subscribeResponseHeader,
SubscribeResponseBody
@@ -102,10 +103,8 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
}
//validate body
- if (StringUtils.isBlank(subscribeRequestBody.getUrl())
- || CollectionUtils.isEmpty(subscribeRequestBody.getTopics())
- || StringUtils.isBlank(subscribeRequestBody.getConsumerGroup())) {
-
+ if (StringUtils.isAnyBlank(subscribeRequestBody.getUrl(),
subscribeRequestBody.getConsumerGroup())
+ || CollectionUtils.isEmpty(subscribeRequestBody.getTopics())) {
responseEventMeshCommand = request.createHttpCommandResponse(
subscribeResponseHeader,
SubscribeResponseBody
@@ -117,17 +116,17 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
final List<SubscriptionItem> subTopicList =
subscribeRequestBody.getTopics();
//do acl check
- if
(eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerSecurityEnable())
{
+ if (eventMeshHttpConfiguration.isEventMeshServerSecurityEnable()) {
for (final SubscriptionItem item : subTopicList) {
try {
-
this.acl.doAclCheckInHttpReceive(RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
+ this.acl.doAclCheckInHttpReceive(remoteAddr,
subscribeRequestHeader.getUsername(),
subscribeRequestHeader.getPasswd(),
subscribeRequestHeader.getSys(), item.getTopic(),
requestCode);
} catch (Exception e) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(subscribeResponseHeader,
+ responseEventMeshCommand =
request.createHttpCommandResponse(subscribeResponseHeader,
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(),
e.getMessage()));
asyncContext.onComplete(responseEventMeshCommand);
@@ -144,8 +143,8 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
// validate URL
try {
- if (!IPUtils.isValidDomainOrIp(url,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIpv4BlackList(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIpv6BlackList()))
{
+ if (!IPUtils.isValidDomainOrIp(url,
eventMeshHttpConfiguration.getEventMeshIpv4BlackList(),
+ eventMeshHttpConfiguration.getEventMeshIpv6BlackList())) {
log.error("subscriber url {} is not valid", url);
responseEventMeshCommand = request.createHttpCommandResponse(
subscribeResponseHeader,
@@ -170,7 +169,7 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
// obtain webhook delivery agreement for Abuse Protection
if
(!WebhookUtil.obtainDeliveryAgreement(eventMeshHTTPServer.getHttpClientPool().getClient(),
- url,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshWebhookOrigin()))
{
+ url, eventMeshHttpConfiguration.getEventMeshWebhookOrigin())) {
log.error("subscriber url {} is not allowed by the target system",
url);
responseEventMeshCommand = request.createHttpCommandResponse(
subscribeResponseHeader,
@@ -181,17 +180,18 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
return;
}
- synchronized
(eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping()) {
+ SubscriptionManager subscriptionManager =
eventMeshHTTPServer.getSubscriptionManager();
+ synchronized (subscriptionManager.getLocalClientInfoMapping()) {
ClientInfo clientInfo = getClientInfo(subscribeRequestHeader);
- SubscriptionManager subscriptionManager =
eventMeshHTTPServer.getSubscriptionManager();
subscriptionManager.registerClient(clientInfo, consumerGroup,
subTopicList, url);
subscriptionManager.updateSubscription(clientInfo, consumerGroup,
url, subTopicList);
final long startTime = System.currentTimeMillis();
+ HttpSummaryMetrics summaryMetrics =
eventMeshHTTPServer.getMetrics().getSummaryMetrics();
try {
// subscription relationship change notification
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
-
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().get(consumerGroup));
+
subscriptionManager.getLocalConsumerGroupMapping().get(consumerGroup));
final CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
@@ -200,8 +200,7 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
}
eventMeshHTTPServer.sendResponse(ctx,
httpCommand.httpResponse());
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
- System.currentTimeMillis() - request.getReqTime());
+
summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() -
request.getReqTime());
} catch (Exception ex) {
log.error("onResponse error", ex);
}
@@ -210,7 +209,7 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
responseEventMeshCommand =
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
- final HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
+ final HttpCommand err = request.createHttpCommandResponse(
subscribeResponseHeader,
SubscribeResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getRetCode(),
@@ -226,8 +225,8 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
JsonUtils.toJSONString(subscribeRequestBody.getTopics()),
subscribeRequestBody.getUrl(), e);
}
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ summaryMetrics.recordSendMsgFailed();
+ summaryMetrics.recordSendMsgCost(endTime - startTime);
}
}
}
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 83331458a..d5c72ee51 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
@@ -26,8 +26,11 @@ import
org.apache.eventmesh.common.protocol.http.header.client.UnSubscribeReques
import
org.apache.eventmesh.common.protocol.http.header.client.UnSubscribeResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
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.consumer.SubscriptionManager;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicConf;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
@@ -48,6 +51,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import io.netty.channel.ChannelHandlerContext;
@@ -67,31 +71,30 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
public void processRequest(final ChannelHandlerContext ctx, final
AsyncContext<HttpCommand> asyncContext)
throws Exception {
HttpCommand responseEventMeshCommand;
+ final HttpCommand request = asyncContext.getRequest();
final String localAddress = IPUtils.getLocalAddress();
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 UnSubscribeRequestHeader unSubscribeRequestHeader =
- (UnSubscribeRequestHeader) asyncContext.getRequest().getHeader();
- final UnSubscribeRequestBody unSubscribeRequestBody =
- (UnSubscribeRequestBody) asyncContext.getRequest().getBody();
+ final UnSubscribeRequestHeader unSubscribeRequestHeader =
(UnSubscribeRequestHeader) request.getHeader();
+ final UnSubscribeRequestBody unSubscribeRequestBody =
(UnSubscribeRequestBody) request.getBody();
+ EventMeshHTTPConfiguration eventMeshHttpConfiguration =
eventMeshHTTPServer.getEventMeshHttpConfiguration();
final UnSubscribeResponseHeader unSubscribeResponseHeader =
UnSubscribeResponseHeader
-
.buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster(),
- localAddress,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv(),
-
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+ .buildHeader(Integer.valueOf(request.getRequestCode()),
+ eventMeshHttpConfiguration.getEventMeshCluster(),
+ localAddress,
eventMeshHttpConfiguration.getEventMeshEnv(),
+ eventMeshHttpConfiguration.getEventMeshIDC());
//validate header
- if (StringUtils.isBlank(unSubscribeRequestHeader.getIdc())
- || StringUtils.isBlank(unSubscribeRequestHeader.getPid())
- || !StringUtils.isNumeric(unSubscribeRequestHeader.getPid())
- || StringUtils.isBlank(unSubscribeRequestHeader.getSys())) {
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
+ 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(),
@@ -101,11 +104,9 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
}
//validate body
- if (StringUtils.isBlank(unSubscribeRequestBody.getUrl())
- || CollectionUtils.isEmpty(unSubscribeRequestBody.getTopics())
- || StringUtils.isBlank(unSubscribeRequestBody.getConsumerGroup()))
{
-
- responseEventMeshCommand =
asyncContext.getRequest().createHttpCommandResponse(
+ if (StringUtils.isAnyBlank(unSubscribeRequestBody.getUrl(),
unSubscribeRequestBody.getConsumerGroup())
+ || CollectionUtils.isEmpty(unSubscribeRequestBody.getTopics())) {
+ responseEventMeshCommand = request.createHttpCommandResponse(
unSubscribeResponseHeader,
UnSubscribeResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
@@ -118,27 +119,29 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
final String consumerGroup = unSubscribeRequestBody.getConsumerGroup();
final String unSubscribeUrl = unSubscribeRequestBody.getUrl();
final List<String> unSubTopicList = unSubscribeRequestBody.getTopics();
-
+ HttpSummaryMetrics summaryMetrics =
eventMeshHTTPServer.getMetrics().getSummaryMetrics();
final CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
if (log.isDebugEnabled()) {
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
}
};
- synchronized
(eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping()) {
+ SubscriptionManager subscriptionManager =
eventMeshHTTPServer.getSubscriptionManager();
+ ConcurrentHashMap<String, ConsumerGroupConf> localConsumerGroupMap =
+ subscriptionManager.getLocalConsumerGroupMapping();
+ synchronized (subscriptionManager.getLocalClientInfoMapping()) {
boolean isChange = true;
registerClient(unSubscribeRequestHeader, consumerGroup,
unSubTopicList, unSubscribeUrl);
for (final String unSubTopic : unSubTopicList) {
- final List<Client> groupTopicClients =
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping()
+ final List<Client> groupTopicClients =
subscriptionManager.getLocalClientInfoMapping()
.get(consumerGroup + "@" + unSubTopic);
final Iterator<Client> clientIterator =
groupTopicClients.iterator();
@@ -164,30 +167,24 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
}
}
- synchronized
(eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping()) {
- final ConsumerGroupConf consumerGroupConf =
-
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().get(consumerGroup);
- final Map<String, ConsumerGroupTopicConf> map =
- consumerGroupConf.getConsumerGroupTopicConf();
+ synchronized (localConsumerGroupMap) {
+ final ConsumerGroupConf consumerGroupConf =
localConsumerGroupMap.get(consumerGroup);
+
+ final Map<String, ConsumerGroupTopicConf> map =
consumerGroupConf.getConsumerGroupTopicConf();
for (final Map.Entry<String, ConsumerGroupTopicConf>
topicConf : map.entrySet()) {
// only modify the topic to subscribe
if (StringUtils.equals(unSubTopic,
topicConf.getKey())) {
- final ConsumerGroupTopicConf latestTopicConf =
- new ConsumerGroupTopicConf();
+ final ConsumerGroupTopicConf latestTopicConf =
new ConsumerGroupTopicConf();
latestTopicConf.setConsumerGroup(consumerGroup);
latestTopicConf.setTopic(unSubTopic);
- latestTopicConf
-
.setSubscriptionItem(topicConf.getValue().getSubscriptionItem());
+
latestTopicConf.setSubscriptionItem(topicConf.getValue().getSubscriptionItem());
latestTopicConf.setUrls(clientUrls);
-
latestTopicConf.setIdcUrls(idcUrls);
-
map.put(unSubTopic, latestTopicConf);
}
}
-
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping()
- .put(consumerGroup, consumerGroupConf);
+ localConsumerGroupMap.put(consumerGroup,
consumerGroupConf);
}
} else {
isChange = false;
@@ -199,14 +196,14 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
if (isChange) {
try {
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
-
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().get(consumerGroup));
+ localConsumerGroupMap.get(consumerGroup));
responseEventMeshCommand =
-
asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
+
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
- final HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
+ final HttpCommand err = request.createHttpCommandResponse(
unSubscribeResponseHeader,
UnSubscribeResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode(),
@@ -218,8 +215,8 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
endTime - startTime,
JsonUtils.toJSONString(unSubscribeRequestBody.getTopics()),
unSubscribeRequestBody.getUrl(), e);
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ summaryMetrics.recordSendMsgFailed();
+ summaryMetrics.recordSendMsgCost(endTime - startTime);
}
} else {
//remove
@@ -227,16 +224,16 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
eventMeshHTTPServer.getConsumerManager()
.notifyConsumerManager(consumerGroup, null);
responseEventMeshCommand =
-
asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
+
request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
// clean ClientInfo
-
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().keySet()
+ subscriptionManager.getLocalClientInfoMapping().keySet()
.removeIf(s -> StringUtils.contains(s, consumerGroup));
// clean ConsumerGroupInfo
-
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().keySet()
+ localConsumerGroupMap.keySet()
.removeIf(s -> StringUtils.equals(consumerGroup, s));
} catch (Exception e) {
- final HttpCommand err =
asyncContext.getRequest().createHttpCommandResponse(
+ final HttpCommand err = request.createHttpCommandResponse(
unSubscribeResponseHeader,
UnSubscribeResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode(),
@@ -248,8 +245,8 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
+ "|topic={}|url={}", endTime - startTime,
JsonUtils.toJSONString(unSubscribeRequestBody.getTopics()),
unSubscribeRequestBody.getUrl(), e);
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
-
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ summaryMetrics.recordSendMsgFailed();
+ summaryMetrics.recordSendMsgCost(endTime - startTime);
}
}
}
@@ -277,9 +274,10 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
client.setLastUpTime(new Date());
final String groupTopicKey = client.getConsumerGroup() + "@" +
client.getTopic();
- if
(eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().containsKey(groupTopicKey))
{
- final List<Client> localClients =
-
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().get(groupTopicKey);
+ ConcurrentHashMap<String, List<Client>> localClientInfoMap =
eventMeshHTTPServer.getSubscriptionManager()
+ .getLocalClientInfoMapping();
+ if (localClientInfoMap.containsKey(groupTopicKey)) {
+ final List<Client> localClients =
localClientInfoMap.get(groupTopicKey);
boolean isContains = false;
for (final Client localClient : localClients) {
if (StringUtils.equals(localClient.getUrl(),
client.getUrl())) {
@@ -294,7 +292,7 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
} else {
final List<Client> clients = new ArrayList<>();
clients.add(client);
-
eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().put(groupTopicKey,
clients);
+ localClientInfoMap.put(groupTopicKey, clients);
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
