This is an automated email from the ASF dual-hosted git repository.
mxsm 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 5494919f4 [ISSUE #4335] Fix Webhook connection not closing after
processing (#4336)
5494919f4 is described below
commit 5494919f48ce2572640ade5d60a64c037822c526
Author: Pil0tXia <[email protected]>
AuthorDate: Mon Aug 14 21:05:54 2023 +0800
[ISSUE #4335] Fix Webhook connection not closing after processing (#4336)
* fix: Close connection asynchronously
* chore: add final to cacheWebHookConfig
* Revert "chore: add final to cacheWebHookConfig"
This reverts commit 2604d2b998f8a3b54a57876b023b9d39c0230bec.
* feat: add a short-lived interface instead of edit persistent connection
* add license
* refactor: add ConnectionType param for postHandler
* fix: some 'crate' typos and a grammar mistake
---
.../eventmesh/common/enums/ConnectionType.java | 20 +++--------
.../common/protocol/http/common/RequestURI.java | 2 +-
.../http/processor/AsyncHttpProcessor.java | 1 +
.../http/processor/CreateTopicProcessor.java | 2 +-
.../protocol/http/processor/HandlerService.java | 39 ++++++++++++++++++----
...cHttpProcessor.java => ShortHttpProcessor.java} | 15 ++-------
.../protocol/http/processor/WebHookProcessor.java | 2 +-
.../eventmesh/runtime/util/HttpResponseUtils.java | 4 +--
.../webhook/admin/FileWebHookConfigOperation.java | 2 +-
9 files changed, 45 insertions(+), 42 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/enums/ConnectionType.java
similarity index 60%
copy from
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
copy to
eventmesh-common/src/main/java/org/apache/eventmesh/common/enums/ConnectionType.java
index 501d5e3c5..ce7e062a5 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/enums/ConnectionType.java
@@ -15,21 +15,9 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.core.protocol.http.processor;
+package org.apache.eventmesh.common.enums;
-import
org.apache.eventmesh.runtime.core.protocol.http.processor.HandlerService.HandlerSpecific;
-
-import io.netty.handler.codec.http.HttpRequest;
-import io.netty.handler.codec.http.HttpResponse;
-
-/**
- * async http processor
- */
-public interface AsyncHttpProcessor extends HttpProcessor {
-
- default HttpResponse handler(HttpRequest httpRequest) {
- return null;
- }
-
- void handler(HandlerSpecific handlerSpecific, HttpRequest httpRequest)
throws Exception;
+public enum ConnectionType {
+ PERSISTENT,
+ SHORT_LIVED
}
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 6aa7ab47c..8fe2c0c74 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
@@ -33,7 +33,7 @@ public enum RequestURI {
UNSUBSCRIBE_REMOTE("/eventmesh/unsubscribe/remote", "UNSUBSCRIBE REMOTE"),
- CRATE_TOPIC("/eventmesh/topic/create", "CRATE TOPIC"),
+ CREATE_TOPIC("/eventmesh/topic/create", "CREATE TOPIC"),
DELETE_TOPIC("/eventmesh/topic/delete", "DELETE TOPIC"),
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
index 501d5e3c5..d641efb53 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
@@ -25,6 +25,7 @@ import io.netty.handler.codec.http.HttpResponse;
/**
* async http processor
*/
+
public interface AsyncHttpProcessor extends HttpProcessor {
default HttpResponse handler(HttpRequest httpRequest) {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
index c4baefd33..777ff09d2 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
@@ -153,6 +153,6 @@ public class CreateTopicProcessor implements
AsyncHttpProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.CRATE_TOPIC.getRequestURI()};
+ return new String[]{RequestURI.CREATE_TOPIC.getRequestURI()};
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
index f4c92bb64..6339f8cee 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.enums.ConnectionType;
import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.utils.JsonUtils;
@@ -148,10 +149,13 @@ public class HandlerService {
}
private void sendResponse(ChannelHandlerContext ctx, HttpRequest request,
HttpResponse response) {
- this.sendResponse(ctx, request, response, true);
+ this.sendPersistentResponse(ctx, request, response, true);
}
- private void sendResponse(ChannelHandlerContext ctx, HttpRequest
httpRequest, HttpResponse response, boolean isClose) {
+ /**
+ * persistent connection
+ */
+ private void sendPersistentResponse(ChannelHandlerContext ctx, HttpRequest
httpRequest, HttpResponse response, boolean isClose) {
ReferenceCountUtil.release(httpRequest);
ctx.writeAndFlush(response).addListener((ChannelFutureListener) f -> {
if (!f.isSuccess()) {
@@ -164,6 +168,19 @@ public class HandlerService {
});
}
+ /**
+ * short-lived connection
+ */
+ private void sendShortResponse(ChannelHandlerContext ctx, HttpRequest
httpRequest, HttpResponse response) {
+ ReferenceCountUtil.release(httpRequest);
+ ctx.writeAndFlush(response).addListener((ChannelFutureListener) f -> {
+ if (!f.isSuccess()) {
+ httpLogger.warn("send response to [{}] with short-lived
connection fail, will close this channel",
+ RemotingHelper.parseChannelRemoteAddr(f.channel()));
+ }
+ }).addListener(ChannelFutureListener.CLOSE);
+ }
+
private HttpEventWrapper parseHttpRequest(HttpRequest httpRequest) throws
IOException {
HttpEventWrapper httpEventWrapper = new HttpEventWrapper();
httpEventWrapper.setHttpMethod(httpRequest.method().name());
@@ -267,7 +284,11 @@ public class HandlerService {
}
response = processorWrapper.httpProcessor.handler(request);
- this.postHandler();
+ if (processorWrapper.httpProcessor instanceof
ShortHttpProcessor) {
+ this.postHandler(ConnectionType.SHORT_LIVED);
+ return;
+ }
+ this.postHandler(ConnectionType.PERSISTENT);
} catch (Throwable e) {
exception = e;
// todo: according exception to generate response
@@ -276,7 +297,7 @@ public class HandlerService {
}
}
- private void postHandler() {
+ private void postHandler(ConnectionType type) {
metrics.getSummaryMetrics().recordHTTPRequest();
if (httpLogger.isDebugEnabled()) {
httpLogger.debug("{}", request);
@@ -285,7 +306,11 @@ public class HandlerService {
this.response = HttpResponseUtils.createSuccess();
}
this.traceOperation.endTrace(ce);
- HandlerService.this.sendResponse(ctx, this.request, this.response);
+ if (type == ConnectionType.PERSISTENT) {
+ HandlerService.this.sendResponse(ctx, this.request,
this.response);
+ } else if (type == ConnectionType.SHORT_LIVED) {
+ sendShortResponse(ctx, this.request, this.response);
+ }
}
private void preHandler() {
@@ -314,7 +339,7 @@ public class HandlerService {
public void sendResponse(HttpResponse response) {
this.response = response;
- this.postHandler();
+ this.postHandler(ConnectionType.PERSISTENT);
}
public void sendResponse(Map<String, Object> responseHeaderMap,
Map<String, Object> responseBodyMap) {
@@ -322,7 +347,7 @@ public class HandlerService {
HttpEventWrapper responseWrapper =
asyncContext.getRequest().createHttpResponse(responseHeaderMap,
responseBodyMap);
asyncContext.onComplete(responseWrapper);
this.response = asyncContext.getResponse().httpResponse();
- this.postHandler();
+ this.postHandler(ConnectionType.PERSISTENT);
} catch (Exception e) {
this.exception = e;
// todo: according exception to generate response
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
similarity index 65%
copy from
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
copy to
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
index 501d5e3c5..428ba8653 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AsyncHttpProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ShortHttpProcessor.java
@@ -17,19 +17,8 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
-import
org.apache.eventmesh.runtime.core.protocol.http.processor.HandlerService.HandlerSpecific;
-
-import io.netty.handler.codec.http.HttpRequest;
-import io.netty.handler.codec.http.HttpResponse;
-
/**
- * async http processor
+ * short-lived HTTP processor
*/
-public interface AsyncHttpProcessor extends HttpProcessor {
-
- default HttpResponse handler(HttpRequest httpRequest) {
- return null;
- }
- void handler(HandlerSpecific handlerSpecific, HttpRequest httpRequest)
throws Exception;
-}
+public interface ShortHttpProcessor extends HttpProcessor {}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/WebHookProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/WebHookProcessor.java
index 885d5cbd9..256b149b3 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/WebHookProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/WebHookProcessor.java
@@ -34,7 +34,7 @@ import io.netty.handler.codec.http.HttpResponse;
import lombok.Setter;
@EventMeshTrace(isEnable = true)
-public class WebHookProcessor implements HttpProcessor {
+public class WebHookProcessor implements ShortHttpProcessor {
@Setter
private WebHookController webHookController;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/HttpResponseUtils.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/HttpResponseUtils.java
index 0b3812f8a..ccaa13b0e 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/HttpResponseUtils.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/HttpResponseUtils.java
@@ -45,7 +45,7 @@ public class HttpResponseUtils {
return new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.INTERNAL_SERVER_ERROR);
}
- private static ByteBuf crateByteBuf(ChannelHandlerContext ctx, String
body) {
+ private static ByteBuf createByteBuf(ChannelHandlerContext ctx, String
body) {
byte[] bytes = body.getBytes(Constants.DEFAULT_CHARSET);
ByteBuf byteBuf = ctx.alloc().buffer(bytes.length);
byteBuf.writeBytes(bytes);
@@ -64,7 +64,7 @@ public class HttpResponseUtils {
public static HttpResponse getHttpResponse(String body,
ChannelHandlerContext ctx, AsciiString headerValue) {
HttpHeaders responseHeaders = new DefaultHttpHeaders();
responseHeaders.add(HttpHeaderNames.CONTENT_TYPE, headerValue);
- return new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.OK, crateByteBuf(ctx, body),
+ return new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.OK, createByteBuf(ctx, body),
responseHeaders, responseHeaders);
}
diff --git
a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
index 7fe69a21c..fd8535a53 100644
---
a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
+++
b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
@@ -76,7 +76,7 @@ public class FileWebHookConfigOperation implements
WebHookConfigOperation {
final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
if (webhookConfigFile.exists()) {
if (log.isErrorEnabled()) {
- log.error("webhookConfig {} is existed",
webHookConfig.getCallbackPath());
+ log.error("webhookConfig {} exists",
webHookConfig.getCallbackPath());
}
return 0;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
