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 e287e76fd [ISSUE #4222]support metadata management in http protocol
(#4231)
e287e76fd is described below
commit e287e76fd12624e7c4053a08499dd3371dab2562
Author: lrhkobe <[email protected]>
AuthorDate: Mon Jul 17 16:17:35 2023 +0800
[ISSUE #4222]support metadata management in http protocol (#4231)
* support metadata management in http protocol
* fix checkstyle:file format problem
* fix checkstyle problem
---
.../protocol/http/common/EventMeshRetCode.java | 3 +-
.../common/protocol/http/common/RequestURI.java | 10 +-
.../runtime/boot/EventMeshHTTPServer.java | 12 ++
.../http/processor/CreateTopicProcessor.java | 156 ++++++++++++++++++
.../http/processor/DeleteTopicProcessor.java | 179 +++++++++++++++++++++
.../http/processor/QuerySubscriptionProcessor.java | 118 ++++++++++++++
6 files changed, 475 insertions(+), 3 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/EventMeshRetCode.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/EventMeshRetCode.java
index bc551a2b4..d5b46475c 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/EventMeshRetCode.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/EventMeshRetCode.java
@@ -41,7 +41,8 @@ public enum EventMeshRetCode {
EVENTMESH_UNSUBSCRIBE_ERR(20, "eventMesh unsubscribe err"),
EVENTMESH_HEARTBEAT_ERR(21, "eventMesh heartbeat err"),
EVENTMESH_ACL_ERR(22, "eventMesh acl err"),
- EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR(23, "eventMesh http msg send over
the limit, ");
+ EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR(23, "eventMesh http msg send over
the limit, "),
+ EVENTMESH_OPERATE_FAIL(100, "operate fail, ");
private final Integer retCode;
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/RequestURI.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/RequestURI.java
index da5ebf58b..6aa7ab47c 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/RequestURI.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/RequestURI.java
@@ -29,9 +29,15 @@ public enum RequestURI {
SUBSCRIBE_REMOTE("/eventmesh/subscribe/remote", "SUBSCRIBE REMOTE"),
- UNSUBSCRIBE_LOCAL("/eventmesh/unsubscribe/local", "SUBSCRIBE LOCAL"),
+ UNSUBSCRIBE_LOCAL("/eventmesh/unsubscribe/local", "UNSUBSCRIBE LOCAL"),
- UNSUBSCRIBE_REMOTE("/eventmesh/unsubscribe/remote", "SUBSCRIBE REMOTE");
+ UNSUBSCRIBE_REMOTE("/eventmesh/unsubscribe/remote", "UNSUBSCRIBE REMOTE"),
+
+ CRATE_TOPIC("/eventmesh/topic/create", "CRATE TOPIC"),
+
+ DELETE_TOPIC("/eventmesh/topic/delete", "DELETE TOPIC"),
+
+ SUBSCRIPTION_QUERY("/eventmesh/subscrition/query", "SUBSCRIPTION QUERY");
private final String requestURI;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
index 5036f6660..9d407b3cf 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
@@ -34,10 +34,13 @@ import
org.apache.eventmesh.runtime.core.protocol.http.consumer.ConsumerManager;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.AdminMetricsProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.BatchSendMessageProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.BatchSendMessageV2Processor;
+import
org.apache.eventmesh.runtime.core.protocol.http.processor.CreateTopicProcessor;
+import
org.apache.eventmesh.runtime.core.protocol.http.processor.DeleteTopicProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.HandlerService;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.HeartBeatProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.LocalSubscribeEventProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.LocalUnSubscribeEventProcessor;
+import
org.apache.eventmesh.runtime.core.protocol.http.processor.QuerySubscriptionProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.RemoteSubscribeEventProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.RemoteUnSubscribeEventProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.ReplyMessageProcessor;
@@ -363,6 +366,15 @@ public class EventMeshHTTPServer extends
AbstractHTTPServer {
final ReplyMessageProcessor replyMessageProcessor = new
ReplyMessageProcessor(this);
registerProcessor(RequestCode.REPLY_MESSAGE.getRequestCode(),
replyMessageProcessor, replyMsgExecutor);
+ CreateTopicProcessor createTopicProcessor = new
CreateTopicProcessor(this);
+ this.getHandlerService().register(createTopicProcessor,
clientManageExecutor);
+
+ DeleteTopicProcessor deleteTopicProcessor = new
DeleteTopicProcessor(this);
+ this.getHandlerService().register(deleteTopicProcessor,
clientManageExecutor);
+
+ QuerySubscriptionProcessor querySubscriptionProcessor = new
QuerySubscriptionProcessor(this);
+ this.getHandlerService().register(querySubscriptionProcessor,
clientManageExecutor);
+
}
private void initWebhook() throws Exception {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
new file mode 100644
index 000000000..35adc3f30
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.core.protocol.http.processor;
+
+import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
+import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.RequestURI;
+import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
+import org.apache.eventmesh.runtime.core.protocol.http.async.CompleteHandler;
+import
org.apache.eventmesh.runtime.core.protocol.http.consumer.HttpClientGroupMapping;
+import org.apache.eventmesh.runtime.util.EventMeshUtil;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+public class CreateTopicProcessor implements AsyncHttpProcessor {
+
+ public Logger httpLogger = LoggerFactory.getLogger("http");
+
+ private final transient EventMeshHTTPServer eventMeshHTTPServer;
+
+ public CreateTopicProcessor(EventMeshHTTPServer eventMeshHTTPServer) {
+ this.eventMeshHTTPServer = eventMeshHTTPServer;
+ }
+
+ @Override
+ public void handler(HandlerService.HandlerSpecific handlerSpecific,
HttpRequest httpRequest)
+ throws Exception {
+ final AsyncContext<HttpEventWrapper> asyncContext =
handlerSpecific.getAsyncContext();
+ final ChannelHandlerContext ctx = handlerSpecific.getCtx();
+ final HttpEventWrapper requestWrapper = asyncContext.getRequest();
+
+ HttpEventWrapper responseWrapper;
+
+ httpLogger.info("uri={}|{}|client2eventMesh|from={}|to={}",
requestWrapper.getRequestURI(),
+ EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtils.getLocalAddress()
+ );
+
+ // user request header
+ Map<String, Object> userRequestHeaderMap =
requestWrapper.getHeaderMap();
+ String requestIp =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
+ userRequestHeaderMap.put(ProtocolKey.ClientInstanceKey.IP, requestIp);
+
+
+ Map<String, Object> responseHeaderMap = new HashMap<>();
+ responseHeaderMap.put(ProtocolKey.REQUEST_URI,
requestWrapper.getRequestURI());
+ responseHeaderMap
+ .put(ProtocolKey.EventMeshInstanceKey.EVENTMESHCLUSTER,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHIP,
IPUtils.getLocalAddress());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHENV,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHIDC,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+
+ //validate body
+ byte[] requestBody = requestWrapper.getBody();
+
+ Map<String, Object> requestBodyMap =
JsonUtils.parseTypeReferenceObject(new String(requestBody),
+ new TypeReference<HashMap<String, Object>>() {});
+
+
+ if (requestBodyMap.get("topic") == null ||
StringUtils.isBlank(requestBodyMap.get("topic").toString())) {
+ Map<String, Object> responseBodyMap = new HashMap<>();
+ responseBodyMap.put("retCode",
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
+ responseBodyMap.put("retMsg",
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + "topic is null");
+ httpLogger.warn("create topic fail, topic is null");
+ responseWrapper =
requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
+
responseWrapper.setHttpResponseStatus(HttpResponseStatus.BAD_REQUEST);
+
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
responseHeaderMap,
+ responseBodyMap, null);
+ return;
+ }
+
+ String topic = requestBodyMap.get("topic").toString();
+
+ long startTime = System.currentTimeMillis();
+ try {
+ // pub topic in local cache
+ String[] topicArr = topic.split(";");
+ for (String item : topicArr) {
+ item = StringUtils.deleteWhitespace(item);
+ if
(!HttpClientGroupMapping.getInstance().getLocalTopicSet().contains(item)) {
+
HttpClientGroupMapping.getInstance().getLocalTopicSet().add(item);
+ httpLogger.info("create topic success, topic:{}", item);
+ }
+ }
+
+ final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper
-> {
+ try {
+ if (httpLogger.isDebugEnabled()) {
+ httpLogger.debug("{}", httpEventWrapper);
+ }
+ eventMeshHTTPServer.sendResponse(ctx,
httpEventWrapper.httpResponse());
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
+ System.currentTimeMillis() -
requestWrapper.getReqTime());
+ } catch (Exception ex) {
+ // ignore
+ httpLogger.warn("create topic, sendResponse fail,", ex);
+ }
+ };
+
+ responseWrapper =
requestWrapper.createHttpResponse(EventMeshRetCode.SUCCESS);
+ asyncContext.onComplete(responseWrapper, handler);
+
+ } catch (Exception e) {
+ Map<String, Object> responseBodyMap = new HashMap<>();
+ responseBodyMap.put("retCode",
EventMeshRetCode.EVENTMESH_RUNTIME_ERR.getRetCode());
+ responseBodyMap.put("retMsg",
EventMeshRetCode.EVENTMESH_RUNTIME_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2));
+ responseWrapper = asyncContext.getRequest().createHttpResponse(
+ responseHeaderMap, responseBodyMap);
+
responseWrapper.setHttpResponseStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
+
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_RUNTIME_ERR,
responseHeaderMap,
+ responseBodyMap, null);
+ long endTime = System.currentTimeMillis();
+ httpLogger.warn(
+ "create topic fail, eventMesh2client|cost={}ms|topic={}",
endTime - startTime, topic, e);
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ }
+ }
+
+ @Override
+ public String[] paths() {
+ return new String[]{RequestURI.CRATE_TOPIC.getRequestURI()};
+ }
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
new file mode 100644
index 000000000..5b6dbd13d
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.eventmesh.runtime.core.protocol.http.processor;
+
+import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
+import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.RequestURI;
+import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
+import org.apache.eventmesh.runtime.core.protocol.http.async.CompleteHandler;
+import
org.apache.eventmesh.runtime.core.protocol.http.consumer.HttpClientGroupMapping;
+import org.apache.eventmesh.runtime.util.EventMeshUtil;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+public class DeleteTopicProcessor implements AsyncHttpProcessor {
+
+ public Logger httpLogger = LoggerFactory.getLogger("http");
+
+ private final transient EventMeshHTTPServer eventMeshHTTPServer;
+
+ public DeleteTopicProcessor(EventMeshHTTPServer eventMeshHTTPServer) {
+ this.eventMeshHTTPServer = eventMeshHTTPServer;
+ }
+
+ @Override
+ public void handler(HandlerService.HandlerSpecific handlerSpecific,
HttpRequest httpRequest)
+ throws Exception {
+ final AsyncContext<HttpEventWrapper> asyncContext =
handlerSpecific.getAsyncContext();
+ final ChannelHandlerContext ctx = handlerSpecific.getCtx();
+ final HttpEventWrapper requestWrapper = asyncContext.getRequest();
+
+ HttpEventWrapper responseWrapper;
+
+ httpLogger.info("uri={}|{}|client2eventMesh|from={}|to={}",
requestWrapper.getRequestURI(),
+ EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtils.getLocalAddress()
+ );
+
+ // user request header
+ Map<String, Object> userRequestHeaderMap =
requestWrapper.getHeaderMap();
+ String requestIp =
RemotingHelper.parseChannelRemoteAddr(ctx.channel());
+ userRequestHeaderMap.put(ProtocolKey.ClientInstanceKey.IP, requestIp);
+
+
+ Map<String, Object> responseHeaderMap = new HashMap<>();
+ responseHeaderMap.put(ProtocolKey.REQUEST_URI,
requestWrapper.getRequestURI());
+ responseHeaderMap
+ .put(ProtocolKey.EventMeshInstanceKey.EVENTMESHCLUSTER,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHIP,
IPUtils.getLocalAddress());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHENV,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHIDC,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+
+ //validate body
+ byte[] requestBody = requestWrapper.getBody();
+
+ Map<String, Object> requestBodyMap =
JsonUtils.parseTypeReferenceObject(new String(requestBody),
+ new TypeReference<HashMap<String, Object>>() {});
+
+ if (requestBodyMap.get("topic") == null ||
StringUtils.isBlank(requestBodyMap.get("topic").toString())) {
+ Map<String, Object> responseBodyMap = new HashMap<>();
+ responseBodyMap.put("retCode",
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode());
+ responseBodyMap.put("retMsg",
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + "topic is null");
+ httpLogger.warn("delete topic fail, topic is null");
+ responseWrapper =
requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
+
responseWrapper.setHttpResponseStatus(HttpResponseStatus.BAD_REQUEST);
+
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
responseHeaderMap,
+ responseBodyMap, null);
+ return;
+ }
+
+ String topic = requestBodyMap.get("topic").toString();
+
+ long startTime = System.currentTimeMillis();
+ try {
+ // pub topic in local cache
+ String[] topicArr = topic.split(";");
+
+ //validate topic is exist in eventmesh
+ List<String> faildTopic = new ArrayList<>();
+ for (String deleteTopic : topicArr) {
+ if
(!HttpClientGroupMapping.getInstance().getLocalTopicSet().contains(deleteTopic))
{
+ faildTopic.add(deleteTopic);
+ }
+ }
+ if (faildTopic.size() > 0) {
+
+ Map<String, Object> responseBodyMap = new HashMap<>();
+ StringBuilder sb = new StringBuilder();
+ sb.append(faildTopic.toString()).append(" not exist in
eventmesh");
+ httpLogger.warn("delete topic fail, {}", sb.toString());
+ responseBodyMap.put("retCode",
EventMeshRetCode.EVENTMESH_OPERATE_FAIL.getRetCode());
+ responseBodyMap.put("retMsg",
EventMeshRetCode.EVENTMESH_OPERATE_FAIL.getErrMsg() + sb.toString());
+ responseWrapper =
requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
+
responseWrapper.setHttpResponseStatus(HttpResponseStatus.BAD_REQUEST);
+
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_OPERATE_FAIL,
responseHeaderMap,
+ responseBodyMap, null);
+ return;
+ }
+ for (String item : topicArr) {
+ boolean flag =
HttpClientGroupMapping.getInstance().getLocalTopicSet().remove(StringUtils.deleteWhitespace(item));
+ if (flag) {
+ httpLogger.info("remove topic success, topic:{}", item);
+ }
+ }
+
+ final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper
-> {
+ try {
+ if (httpLogger.isDebugEnabled()) {
+ httpLogger.debug("{}", httpEventWrapper);
+ }
+ eventMeshHTTPServer.sendResponse(ctx,
httpEventWrapper.httpResponse());
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
+ System.currentTimeMillis() -
requestWrapper.getReqTime());
+ } catch (Exception ex) {
+ // ignore
+ httpLogger.warn("delete topic, sendResponse fail,", ex);
+ }
+ };
+
+ responseWrapper =
requestWrapper.createHttpResponse(EventMeshRetCode.SUCCESS);
+ asyncContext.onComplete(responseWrapper, handler);
+
+ } catch (Exception e) {
+ Map<String, Object> responseBodyMap = new HashMap<>();
+ responseBodyMap.put("retCode",
EventMeshRetCode.EVENTMESH_RUNTIME_ERR.getRetCode());
+ responseBodyMap.put("retMsg",
EventMeshRetCode.EVENTMESH_RUNTIME_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2));
+ responseWrapper = asyncContext.getRequest().createHttpResponse(
+ responseHeaderMap, responseBodyMap);
+
responseWrapper.setHttpResponseStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
+
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_RUNTIME_ERR,
responseHeaderMap,
+ responseBodyMap, null);
+ long endTime = System.currentTimeMillis();
+ httpLogger.warn(
+ "delete topic fail, eventMesh2client|cost={}ms|topic={}",
endTime - startTime, topic, e);
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ }
+ }
+
+ @Override
+ public String[] paths() {
+ return new String[]{RequestURI.DELETE_TOPIC.getRequestURI()};
+ }
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
new file mode 100644
index 000000000..8d06c2a8d
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.core.protocol.http.processor;
+
+import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
+import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.RequestURI;
+import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
+import org.apache.eventmesh.runtime.core.protocol.http.async.CompleteHandler;
+import
org.apache.eventmesh.runtime.core.protocol.http.consumer.HttpClientGroupMapping;
+import org.apache.eventmesh.runtime.util.EventMeshUtil;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+public class QuerySubscriptionProcessor implements AsyncHttpProcessor {
+
+ public Logger httpLogger = LoggerFactory.getLogger("http");
+
+ private final transient EventMeshHTTPServer eventMeshHTTPServer;
+
+ public QuerySubscriptionProcessor(EventMeshHTTPServer eventMeshHTTPServer)
{
+ this.eventMeshHTTPServer = eventMeshHTTPServer;
+ }
+
+ @Override
+ public void handler(HandlerService.HandlerSpecific handlerSpecific,
HttpRequest httpRequest)
+ throws Exception {
+ final AsyncContext<HttpEventWrapper> asyncContext =
handlerSpecific.getAsyncContext();
+ final ChannelHandlerContext ctx = handlerSpecific.getCtx();
+ final HttpEventWrapper requestWrapper = asyncContext.getRequest();
+
+ HttpEventWrapper responseWrapper;
+
+ httpLogger.info("uri={}|{}|client2eventMesh|from={}|to={}",
requestWrapper.getRequestURI(),
+ EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtils.getLocalAddress()
+ );
+
+ Map<String, Object> responseHeaderMap = new HashMap<>();
+ responseHeaderMap.put(ProtocolKey.REQUEST_URI,
requestWrapper.getRequestURI());
+ responseHeaderMap
+ .put(ProtocolKey.EventMeshInstanceKey.EVENTMESHCLUSTER,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshCluster());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHIP,
IPUtils.getLocalAddress());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHENV,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEnv());
+ responseHeaderMap.put(ProtocolKey.EventMeshInstanceKey.EVENTMESHIDC,
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIDC());
+
+ long startTime = System.currentTimeMillis();
+ try {
+ // pub topic in local cache
+
+ final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper
-> {
+ try {
+ if (httpLogger.isDebugEnabled()) {
+ httpLogger.debug("{}", httpEventWrapper);
+ }
+ eventMeshHTTPServer.sendResponse(ctx,
httpEventWrapper.httpResponse());
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
+ System.currentTimeMillis() -
requestWrapper.getReqTime());
+ } catch (Exception ex) {
+ httpLogger.warn("query subscription, sendResponse fail",
ex);
+ }
+ };
+
+ Map<String, Object> responseBodyMap = new HashMap<>();
+
+ responseBodyMap.put("subsrciption",
HttpClientGroupMapping.getInstance().querySubscription());
+ responseBodyMap.put("localTopicSet",
HttpClientGroupMapping.getInstance().getLocalTopicSet());
+
+ responseWrapper =
requestWrapper.createHttpResponse(responseHeaderMap, responseBodyMap);
+ asyncContext.onComplete(responseWrapper, handler);
+ } catch (Exception e) {
+ Map<String, Object> responseBodyMap = new HashMap<>();
+ responseBodyMap.put("retCode",
EventMeshRetCode.EVENTMESH_RUNTIME_ERR.getRetCode());
+ responseBodyMap.put("retMsg",
EventMeshRetCode.EVENTMESH_RUNTIME_ERR.getErrMsg() +
EventMeshUtil.stackTrace(e, 2));
+ responseWrapper = asyncContext.getRequest().createHttpResponse(
+ responseHeaderMap, responseBodyMap);
+
responseWrapper.setHttpResponseStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
+
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_RUNTIME_ERR,
responseHeaderMap,
+ responseBodyMap, null);
+ long endTime = System.currentTimeMillis();
+ httpLogger.warn("query subscription
fail,eventMesh2client|cost={}ms", endTime - startTime, e);
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgFailed();
+
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordSendMsgCost(endTime
- startTime);
+ }
+ }
+
+ @Override
+ public String[] paths() {
+ return new String[]{RequestURI.SUBSCRIPTION_QUERY.getRequestURI()};
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
