resolve

apache · Jan 16, 2024 · da04b07 · da04b07
1 parent d7dded7
commit da04b07
Show file tree

Hide file tree

Showing 26 changed files with 81 additions and 86 deletions.
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
@@ -397,7 +397,6 @@ private void injectHttpRequestHeader(final ChannelHandlerContext ctx, final Http
 
         private void processHttpCommandRequest(final ChannelHandlerContext ctx, final AsyncContext<HttpCommand> asyncContext) {
             final HttpCommand request = asyncContext.getRequest();
-
             final HttpRequestProcessor choosed = httpRequestProcessorTable.get(request.getRequestCode());
             Runnable runnable = () -> {
                 try {
@@ -426,6 +425,7 @@ private void processHttpCommandRequest(final ChannelHandlerContext ctx, final As
                     if (!asyncContext.isComplete()) {
                         return;
                     }
+
                     log.debug("{}", asyncContext.getResponse());
                     metrics.getSummaryMetrics()
                         .recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());

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
@@ -70,6 +70,7 @@
 
 import lombok.extern.slf4j.Slf4j;
 
+
 /**
  * Add multiple managers to the underlying server
  */
@@ -83,22 +84,16 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
 
     private final Acl acl;
     private final EventBus eventBus = new EventBus();
-
+    private final transient HTTPClientPool httpClientPool = new HTTPClientPool(10);
     private ConsumerManager consumerManager;
     private ProducerManager producerManager;
     private SubscriptionManager subscriptionManager;
-
     private FilterEngine filterEngine;
-
     private TransformerEngine transformerEngine;
-
     private HttpRetryer httpRetryer;
-
     private transient RateLimiter msgRateLimiter;
     private transient RateLimiter batchRateLimiter;
 
-    private final transient HTTPClientPool httpClientPool = new HTTPClientPool(10);
-
     public EventMeshHTTPServer(final EventMeshServer eventMeshServer, final EventMeshHTTPConfiguration eventMeshHttpConfiguration) {
 
         super(eventMeshHttpConfiguration.getHttpServerPort(),

diff --git a/...java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java b/...java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
@@ -17,15 +17,17 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.common.protocol.http.HttpCommand;
 import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
 import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
 
+import java.util.concurrent.Executor;
+
 import io.netty.channel.ChannelHandlerContext;
 
 import lombok.RequiredArgsConstructor;
 
+
 @RequiredArgsConstructor
 public class AdminMetricsProcessor extends AbstractHttpRequestProcessor {
 

diff --git a/...ava/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java b/...ava/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.common.protocol.http.HttpCommand;
 import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
 import org.apache.eventmesh.common.protocol.http.common.RequestCode;
@@ -27,6 +26,8 @@
 import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
 import org.apache.eventmesh.runtime.util.RemotingHelper;
 
+import java.util.concurrent.Executor;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,6 +58,8 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
 
     @Override
     public Executor executor() {
-        return (Runnable runnable)-> {runnable.run();};
+        return (Runnable runnable) -> {
+            runnable.run();
+        };
     }
 }
diff --git a/.../org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java b/.../org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.api.SendCallback;
 import org.apache.eventmesh.api.SendResult;
 import org.apache.eventmesh.api.exception.OnExceptionContext;
@@ -53,6 +52,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;

diff --git a/...rg/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java b/...rg/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.api.SendCallback;
 import org.apache.eventmesh.api.SendResult;
 import org.apache.eventmesh.api.exception.OnExceptionContext;
@@ -49,6 +48,7 @@
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.Objects;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;

diff --git a/.../java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java b/.../java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 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;
@@ -37,6 +36,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.Executor;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

diff --git a/.../java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java b/.../java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 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;
@@ -39,6 +38,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executor;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

diff --git a/...c/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java b/...c/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.common.enums.ConnectionType;
 import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
@@ -43,6 +42,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import org.slf4j.Logger;

diff --git a/...in/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java b/...in/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 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;
@@ -46,6 +45,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 
 import io.netty.channel.ChannelHandlerContext;
 

diff --git a/...rc/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HttpProcessor.java b/...rc/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HttpProcessor.java
@@ -17,9 +17,10 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
+import java.util.concurrent.Executor;
+
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
-import java.util.concurrent.Executor;
 
 /**
  * http processor
@@ -31,8 +32,9 @@ public interface HttpProcessor {
     HttpResponse handler(HttpRequest httpRequest);
 
     /**
-     *
      * @return {@link Executor}
      */
-    default Executor executor() {return null;}
+    default Executor executor() {
+        return null;
+    }
 }
diff --git a/...g/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java b/...g/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.common.protocol.SubscriptionItem;
 import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
@@ -41,6 +40,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.Executor;
 
 import io.netty.channel.Channel;
 import io.netty.handler.codec.http.HttpRequest;
@@ -186,7 +186,7 @@ public void handler(final HandlerService.HandlerSpecific handlerSpecific, final
 
     @Override
     public String[] paths() {
-        return new String[]{RequestURI.SUBSCRIBE_LOCAL.getRequestURI()};
+        return new String[] {RequestURI.SUBSCRIBE_LOCAL.getRequestURI()};
     }
 
     @Override

diff --git a/...apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java b/...apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
 import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
@@ -49,6 +48,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.Executor;
 
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.HttpRequest;
@@ -194,6 +194,7 @@ public void handler(final HandlerService.HandlerSpecific handlerSpecific, final
                 } catch (Exception e) {
                     log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
                         + "|topic={}|url={}", System.currentTimeMillis() - startTime, JsonUtils.toJSONString(unSubTopicList), unSubscribeUrl, e);
+
                     handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR, responseHeaderMap,
                         responseBodyMap, null);
                 }
@@ -213,8 +214,10 @@ public void handler(final HandlerService.HandlerSpecific handlerSpecific, final
                     eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().keySet()
                         .removeIf(s -> StringUtils.equals(consumerGroup, s));
                 } catch (Exception e) {
+
                     log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
                         + "|topic={}|url={}", System.currentTimeMillis() - startTime, JsonUtils.toJSONString(unSubTopicList), unSubscribeUrl, e);
+
                     handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR, responseHeaderMap,
                         responseBodyMap, null);
                 }
@@ -227,7 +230,7 @@ public void handler(final HandlerService.HandlerSpecific handlerSpecific, final
 
     @Override
     public String[] paths() {
-        return new String[]{RequestURI.UNSUBSCRIBE_LOCAL.getRequestURI()};
+        return new String[] {RequestURI.UNSUBSCRIBE_LOCAL.getRequestURI()};
     }
 
     @Override

diff --git a/...org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java b/...org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 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;
@@ -34,6 +33,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.Executor;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

diff --git a/.../apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java b/.../apache/eventmesh/runtime/core/protocol/http/processor/RemoteSubscribeEventProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.common.protocol.SubscriptionItem;
 import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
@@ -44,6 +43,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.Executor;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -218,7 +218,7 @@ public void handler(HandlerService.HandlerSpecific handlerSpecific, HttpRequest
 
     @Override
     public String[] paths() {
-        return new String[]{RequestURI.SUBSCRIBE_REMOTE.getRequestURI()};
+        return new String[] {RequestURI.SUBSCRIBE_REMOTE.getRequestURI()};
     }
 
     @Override

diff --git a/...pache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java b/...pache/eventmesh/runtime/core/protocol/http/processor/RemoteUnSubscribeEventProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.common.protocol.SubscriptionItem;
 import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
@@ -44,6 +43,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.Executor;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -185,7 +185,7 @@ public void handler(HandlerService.HandlerSpecific handlerSpecific, HttpRequest
 
     @Override
     public String[] paths() {
-        return new String[]{RequestURI.UNSUBSCRIBE_REMOTE.getRequestURI()};
+        return new String[] {RequestURI.UNSUBSCRIBE_REMOTE.getRequestURI()};
     }
 
     @Override

diff --git a/...java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java b/...java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.api.SendCallback;
 import org.apache.eventmesh.api.SendResult;
 import org.apache.eventmesh.api.exception.OnExceptionContext;
@@ -48,6 +47,7 @@
 import org.apache.commons.lang3.ObjectUtils;
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;

diff --git a/...va/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java b/...va/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java
@@ -17,7 +17,6 @@
 
 package org.apache.eventmesh.runtime.core.protocol.http.processor;
 
-import java.util.concurrent.Executor;
 import org.apache.eventmesh.api.SendCallback;
 import org.apache.eventmesh.api.SendResult;
 import org.apache.eventmesh.api.exception.AclException;
@@ -53,6 +52,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 import io.cloudevents.CloudEvent;
@@ -309,7 +309,7 @@ public void onException(final OnExceptionContext context) {
 
     @Override
     public String[] paths() {
-        return new String[]{RequestURI.PUBLISH.getRequestURI()};
+        return new String[] {RequestURI.PUBLISH.getRequestURI()};
     }
 
     @Override