This is an automated email from the ASF dual-hosted git repository.
jianbin pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/2.x by this push:
new bf6601f5fb feature: support http2 (#7451)
bf6601f5fb is described below
commit bf6601f5fb53d7b37d2fa7598a1bc0c3bae0b221
Author: funkye <[email protected]>
AuthorDate: Wed Jul 2 15:47:35 2025 +0800
feature: support http2 (#7451)
---
changes/en-us/2.x.md | 1 +
changes/zh-cn/2.x.md | 2 +-
.../apache/seata/common/rpc/http/HttpContext.java | 50 ++++--
.../core/protocol/detector/Http2Detector.java | 30 +++-
.../seata/core/protocol/detector/HttpDetector.java | 69 ++++++-
.../rpc/netty/http/BaseHttpChannelHandler.java | 47 +++++
.../core/rpc/netty/http/Http2HttpHandler.java | 199 +++++++++++++++++++++
.../core/rpc/netty/http/HttpDispatchHandler.java | 33 +---
.../core/rpc/netty/http/SimpleHttp2Request.java | 50 ++++++
.../core/protocol/detector/Http2DetectorTest.java | 65 +++++++
.../core/protocol/detector/HttpDetectorTest.java | 93 +++++++++-
.../core/rpc/netty/http/Http2HttpHandlerTest.java | 179 ++++++++++++++++++
.../cluster/manager/ClusterWatcherManager.java | 36 ++--
13 files changed, 785 insertions(+), 69 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 6b0e7e7f57..5a4418005c 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -21,6 +21,7 @@ Add changes here for all PR submitted to the 2.x branch.
### feature:
- [[#7261](https://github.com/apache/incubator-seata/pull/7261)] enforce
account initialization and disable default credentials
+- [[#7451](https://github.com/apache/incubator-seata/pull/7451)] seata-server
supports the HTTP/2 protocol
### bugfix:
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index 145279ed99..0381b40c3d 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -21,7 +21,7 @@
### feature:
- [[#7261](https://github.com/apache/incubator-seata/pull/7261)]
强制进行账户初始化并禁用默认凭据
-
+- [[#7451](https://github.com/apache/incubator-seata/pull/7451)]
seata-server支持HTTP/2协议
### bugfix:
diff --git
a/common/src/main/java/org/apache/seata/common/rpc/http/HttpContext.java
b/common/src/main/java/org/apache/seata/common/rpc/http/HttpContext.java
index d3d60fc639..7b984d68ee 100644
--- a/common/src/main/java/org/apache/seata/common/rpc/http/HttpContext.java
+++ b/common/src/main/java/org/apache/seata/common/rpc/http/HttpContext.java
@@ -17,37 +17,45 @@
package org.apache.seata.common.rpc.http;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.http.HttpRequest;
-public class HttpContext {
+public class HttpContext<T> {
- HttpRequest request;
+ public static final String HTTP_1_1 = "HTTP/1.1";
+ public static final String HTTP_2_0 = "HTTP/2.0";
- ChannelHandlerContext context;
+ private T request;
- boolean keepAlive;
+ private ChannelHandlerContext context;
- boolean async = false;
+ private boolean keepAlive;
- public HttpContext(HttpRequest request, ChannelHandlerContext context,
boolean keepAlive) {
+ private boolean async = false;
+
+ private String httpVersion;
+
+ public HttpContext(T request, ChannelHandlerContext context, boolean
keepAlive, String httpVersion) {
this.request = request;
this.context = context;
this.keepAlive = keepAlive;
+ this.httpVersion = httpVersion;
}
- public boolean isAsync() {
- return async;
+ public HttpContext(T request, ChannelHandlerContext context, boolean
keepAlive) {
+ this.request = request;
+ this.context = context;
+ this.keepAlive = keepAlive;
+ this.httpVersion = HTTP_1_1;
}
- public void setAsync(boolean async) {
- this.async = async;
+ public boolean isHttp2() {
+ return HTTP_2_0.equals(httpVersion);
}
- public HttpRequest getRequest() {
+ public T getRequest() {
return request;
}
- public void setRequest(HttpRequest request) {
+ public void setRequest(T request) {
this.request = request;
}
@@ -66,4 +74,20 @@ public class HttpContext {
public void setKeepAlive(boolean keepAlive) {
this.keepAlive = keepAlive;
}
+
+ public boolean isAsync() {
+ return async;
+ }
+
+ public void setAsync(boolean async) {
+ this.async = async;
+ }
+
+ public String getHttpVersion() {
+ return httpVersion;
+ }
+
+ public void setHttpVersion(String httpVersion) {
+ this.httpVersion = httpVersion;
+ }
}
diff --git
a/core/src/main/java/org/apache/seata/core/protocol/detector/Http2Detector.java
b/core/src/main/java/org/apache/seata/core/protocol/detector/Http2Detector.java
index ce5d42aa7a..9099c82977 100644
---
a/core/src/main/java/org/apache/seata/core/protocol/detector/Http2Detector.java
+++
b/core/src/main/java/org/apache/seata/core/protocol/detector/Http2Detector.java
@@ -18,18 +18,23 @@ package org.apache.seata.core.protocol.detector;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
+import io.netty.handler.codec.http2.Http2HeadersFrame;
import io.netty.handler.codec.http2.Http2MultiplexHandler;
import io.netty.handler.codec.http2.Http2StreamChannel;
import io.netty.util.CharsetUtil;
import org.apache.seata.core.rpc.netty.grpc.GrpcDecoder;
import org.apache.seata.core.rpc.netty.grpc.GrpcEncoder;
+import org.apache.seata.core.rpc.netty.http.Http2HttpHandler;
public class Http2Detector implements ProtocolDetector {
private static final byte[] HTTP2_PREFIX_BYTES = "PRI *
HTTP/2.0\r\n\r\nSM\r\n\r\n".getBytes(CharsetUtil.UTF_8);
- private ChannelHandler[] serverHandlers;
+ private final ChannelHandler[] serverHandlers;
public Http2Detector(ChannelHandler[] serverHandlers) {
this.serverHandlers = serverHandlers;
@@ -55,12 +60,29 @@ public class Http2Detector implements ProtocolDetector {
new Http2MultiplexHandler(new
ChannelInitializer<Http2StreamChannel>() {
@Override
protected void initChannel(Http2StreamChannel ch) {
- final ChannelPipeline p = ch.pipeline();
+ ch.pipeline().addLast(new Http2SelectorHandler());
+ }
+ })
+ };
+ }
+
+ private class Http2SelectorHandler extends ChannelInboundHandlerAdapter {
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
+ if (msg instanceof Http2HeadersFrame) {
+ Http2HeadersFrame headersFrame = (Http2HeadersFrame) msg;
+ CharSequence contentType =
headersFrame.headers().get(HttpHeaderNames.CONTENT_TYPE);
+ final ChannelPipeline p = ctx.pipeline();
+ if (contentType != null &&
contentType.toString().endsWith("grpc")) {
p.addLast(new GrpcDecoder());
p.addLast(new GrpcEncoder());
p.addLast(serverHandlers);
+ } else {
+ p.addLast(new Http2HttpHandler());
}
- })
- };
+ p.remove(this);
+ }
+ ctx.fireChannelRead(msg);
+ }
}
}
diff --git
a/core/src/main/java/org/apache/seata/core/protocol/detector/HttpDetector.java
b/core/src/main/java/org/apache/seata/core/protocol/detector/HttpDetector.java
index 2b6cf03256..0bdadb8f4e 100644
---
a/core/src/main/java/org/apache/seata/core/protocol/detector/HttpDetector.java
+++
b/core/src/main/java/org/apache/seata/core/protocol/detector/HttpDetector.java
@@ -18,11 +18,26 @@ package org.apache.seata.core.protocol.detector;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.HttpServerUpgradeHandler;
+import io.netty.handler.codec.http2.Http2CodecUtil;
+import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
+import io.netty.handler.codec.http2.Http2MultiplexHandler;
+import io.netty.handler.codec.http2.Http2ServerUpgradeCodec;
+import io.netty.handler.codec.http2.Http2StreamChannel;
+import io.netty.util.AsciiString;
+import org.apache.seata.core.rpc.netty.http.Http2HttpHandler;
import org.apache.seata.core.rpc.netty.http.HttpDispatchHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class HttpDetector implements ProtocolDetector {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(HttpDetector.class);
private static final String[] HTTP_METHODS = {"GET", "POST", "PUT",
"DELETE", "HEAD", "OPTIONS", "PATCH"};
@Override
@@ -52,7 +67,59 @@ public class HttpDetector implements ProtocolDetector {
@Override
public ChannelHandler[] getHandlers() {
- return new ChannelHandler[] {new HttpServerCodec(), new
HttpObjectAggregator(1048576), new HttpDispatchHandler()
+ HttpServerCodec sourceCodec = new HttpServerCodec();
+ HttpServerUpgradeHandler upgradeHandler =
getHttpServerUpgradeHandler(sourceCodec);
+
+ ChannelInboundHandlerAdapter upgradeCleanupHandler = new
ChannelInboundHandlerAdapter() {
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object
evt) throws Exception {
+ if (evt instanceof HttpServerUpgradeHandler.UpgradeEvent) {
+ ChannelPipeline p = ctx.pipeline();
+ p.remove(HttpObjectAggregator.class);
+ p.remove(HttpDispatchHandler.class);
+ }
+ super.userEventTriggered(ctx, evt);
+ }
+ };
+
+ ChannelInboundHandlerAdapter finalExceptionHandler = new
ChannelInboundHandlerAdapter() {
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable
cause) {
+ if (cause instanceof java.io.IOException) {
+ LOGGER.trace("Connection closed by client: {}",
cause.getMessage());
+ } else {
+ LOGGER.error("Exception caught in HTTP pipeline: ", cause);
+ }
+ ctx.close();
+ }
+ };
+
+ return new ChannelHandler[] {
+ sourceCodec,
+ upgradeHandler,
+ upgradeCleanupHandler,
+ new HttpObjectAggregator(1048576),
+ new HttpDispatchHandler(),
+ finalExceptionHandler
+ };
+ }
+
+ private static HttpServerUpgradeHandler
getHttpServerUpgradeHandler(HttpServerCodec sourceCodec) {
+ HttpServerUpgradeHandler.UpgradeCodecFactory upgradeCodecFactory =
protocol -> {
+ if
(AsciiString.contentEquals(Http2CodecUtil.HTTP_UPGRADE_PROTOCOL_NAME,
protocol)) {
+ return new Http2ServerUpgradeCodec(
+ Http2FrameCodecBuilder.forServer().build(),
+ new Http2MultiplexHandler(new
ChannelInitializer<Http2StreamChannel>() {
+ @Override
+ protected void initChannel(Http2StreamChannel ch) {
+ ch.pipeline().addLast(new Http2HttpHandler());
+ }
+ }));
+ } else {
+ return null;
+ }
};
+
+ return new HttpServerUpgradeHandler(sourceCodec, upgradeCodecFactory,
1048576);
}
}
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/http/BaseHttpChannelHandler.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/http/BaseHttpChannelHandler.java
new file mode 100644
index 0000000000..995b4efb5f
--- /dev/null
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/http/BaseHttpChannelHandler.java
@@ -0,0 +1,47 @@
+/*
+ * 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.seata.core.rpc.netty.http;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.seata.common.thread.NamedThreadFactory;
+import org.apache.seata.core.rpc.netty.NettyServerConfig;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public abstract class BaseHttpChannelHandler<T> extends
SimpleChannelInboundHandler<T> {
+
+ protected static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ /**
+ * HTTP request processing thread pool, independent of Netty IO threads,
to avoid blocking network processing.
+ */
+ protected static final ExecutorService HTTP_HANDLER_THREADS = new
ThreadPoolExecutor(
+ NettyServerConfig.getMinHttpPoolSize(),
+ NettyServerConfig.getMaxHttpPoolSize(),
+ NettyServerConfig.getHttpKeepAliveTime(),
+ TimeUnit.SECONDS,
+ new
LinkedBlockingQueue<>(NettyServerConfig.getMaxHttpTaskQueueSize()),
+ new NamedThreadFactory("HTTPHandlerThread",
NettyServerConfig.getMaxHttpPoolSize()),
+ new ThreadPoolExecutor.AbortPolicy());
+
+ static {
+ Runtime.getRuntime().addShutdownHook(new
Thread(HTTP_HANDLER_THREADS::shutdown));
+ }
+}
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/http/Http2HttpHandler.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/http/Http2HttpHandler.java
new file mode 100644
index 0000000000..5dc3f813a9
--- /dev/null
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/http/Http2HttpHandler.java
@@ -0,0 +1,199 @@
+/*
+ * 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.seata.core.rpc.netty.http;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
+import io.netty.handler.codec.http2.Http2DataFrame;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.handler.codec.http2.Http2HeadersFrame;
+import io.netty.handler.codec.http2.Http2StreamFrame;
+import org.apache.seata.common.rpc.http.HttpContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * The http2 http handler.
+ */
+public class Http2HttpHandler extends BaseHttpChannelHandler<Http2StreamFrame>
{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(Http2HttpHandler.class);
+ private Http2Headers http2Headers;
+ private ByteBuf bodyBuffer;
+ private boolean headersEndStream = false;
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, Http2StreamFrame
msg) throws Exception {
+ if (bodyBuffer == null) {
+ bodyBuffer = ctx.alloc().buffer();
+ }
+ try {
+ if (msg instanceof Http2HeadersFrame) {
+ Http2HeadersFrame headersFrame = (Http2HeadersFrame) msg;
+ this.http2Headers = headersFrame.headers();
+ headersEndStream = headersFrame.isEndStream();
+ if (headersEndStream) {
+ handleRequest(ctx);
+ }
+ } else if (msg instanceof Http2DataFrame) {
+ Http2DataFrame dataFrame = (Http2DataFrame) msg;
+ bodyBuffer.writeBytes(dataFrame.content());
+ if (dataFrame.isEndStream()) {
+ handleRequest(ctx);
+ }
+ }
+ } catch (Exception e) {
+ if (bodyBuffer != null) {
+ bodyBuffer.release();
+ bodyBuffer = null;
+ }
+ throw e;
+ }
+ }
+
+ private void handleRequest(ChannelHandlerContext ctx) {
+ try {
+ if (http2Headers == null || http2Headers.method() == null ||
http2Headers.path() == null) {
+ sendErrorResponse(ctx, HttpResponseStatus.BAD_REQUEST);
+ return;
+ }
+ HttpMethod method =
HttpMethod.valueOf(http2Headers.method().toString());
+ String path = http2Headers.path().toString();
+ String body = bodyBuffer != null ?
bodyBuffer.toString(StandardCharsets.UTF_8) : "";
+ SimpleHttp2Request request = new SimpleHttp2Request(method, path,
http2Headers, body);
+
+ // reuse HttpDispatchHandler logic
+ boolean keepAlive = true; // In HTTP/2, connections are persistent
by default
+ QueryStringDecoder queryStringDecoder = new
QueryStringDecoder(request.getPath());
+ String requestPath = queryStringDecoder.path();
+ HttpInvocation httpInvocation =
ControllerManager.getHttpInvocation(requestPath);
+ if (httpInvocation == null) {
+ sendErrorResponse(ctx, HttpResponseStatus.NOT_FOUND);
+ return;
+ }
+ HttpContext<SimpleHttp2Request> httpContext =
+ new HttpContext<>(request, ctx, keepAlive,
HttpContext.HTTP_2_0);
+ ObjectNode requestDataNode = OBJECT_MAPPER.createObjectNode();
+ requestDataNode.set("param",
ParameterParser.convertParamMap(queryStringDecoder.parameters()));
+ if (request.getMethod() == HttpMethod.POST
+ && request.getBody() != null
+ && !request.getBody().isEmpty()) {
+ // assume body is json
+ try {
+ ObjectNode bodyDataNode = (ObjectNode)
OBJECT_MAPPER.readTree(request.getBody());
+ requestDataNode.set("body", bodyDataNode);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to parse http2 body as json: {}",
e.getMessage());
+ }
+ }
+ Object httpController = httpInvocation.getController();
+ Method handleMethod = httpInvocation.getMethod();
+ Object[] args = ParameterParser.getArgValues(
+ httpInvocation.getParamMetaData(), handleMethod,
requestDataNode, httpContext);
+ handle(httpController, handleMethod, args, ctx, httpContext);
+ } catch (Exception e) {
+ LOGGER.error("Exception occurred while processing HTTP2 request:
{}", e.getMessage(), e);
+ sendErrorResponse(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR);
+ } finally {
+ if (bodyBuffer != null) {
+ bodyBuffer.release();
+ bodyBuffer = null;
+ }
+ http2Headers = null;
+ headersEndStream = false;
+ }
+ }
+
+ private void handle(
+ Object httpController,
+ Method handleMethod,
+ Object[] args,
+ ChannelHandlerContext ctx,
+ HttpContext<SimpleHttp2Request> httpContext) {
+ HTTP_HANDLER_THREADS.execute(() -> {
+ Object result;
+ try {
+ result = handleMethod.invoke(httpController, args);
+ if (!httpContext.isAsync()) {
+ sendResponse(ctx, result);
+ }
+ } catch (IllegalAccessException e) {
+ LOGGER.error("Illegal argument exception: {}", e.getMessage(),
e);
+ sendErrorResponse(ctx, HttpResponseStatus.BAD_REQUEST);
+ } catch (Exception e) {
+ LOGGER.error("Exception occurred while processing HTTP2
request: {}", e.getMessage(), e);
+ sendErrorResponse(ctx,
HttpResponseStatus.INTERNAL_SERVER_ERROR);
+ }
+ });
+ }
+
+ private void sendResponse(ChannelHandlerContext ctx, Object result) throws
Exception {
+ byte[] body = result != null ? OBJECT_MAPPER.writeValueAsBytes(result)
: new byte[0];
+ Http2Headers headers = new
DefaultHttp2Headers().status(HttpResponseStatus.OK.codeAsText());
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "application/json;
charset=UTF-8");
+ headers.set(HttpHeaderNames.CONTENT_LENGTH,
String.valueOf(body.length));
+
+ ctx.write(new DefaultHttp2HeadersFrame(headers));
+ if (body.length > 0) {
+ ByteBuf content = Unpooled.wrappedBuffer(body);
+ ctx.write(new DefaultHttp2DataFrame(content, true));
+ } else {
+ ctx.write(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true));
+ }
+ ctx.flush();
+ }
+
+ private void sendErrorResponse(ChannelHandlerContext ctx,
HttpResponseStatus status) {
+ Http2Headers headers = new
DefaultHttp2Headers().status(status.codeAsText());
+ ctx.writeAndFlush(new DefaultHttp2HeadersFrame(headers, true));
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ // This is a common exception when the client (like curl) closes the
connection after receiving the response.
+ // We can safely ignore it by simply closing the context.
+ if (cause instanceof java.io.IOException) {
+ LOGGER.trace("Client connection closed: {}", cause.getMessage());
+ } else {
+ LOGGER.error("Exception caught in Http2HttpHandler: ", cause);
+ }
+ ctx.close();
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ try {
+ if (bodyBuffer != null) {
+ bodyBuffer.release();
+ bodyBuffer = null;
+ }
+ } finally {
+ super.channelInactive(ctx);
+ }
+ }
+}
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpDispatchHandler.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpDispatchHandler.java
index 23d28bc066..0baa5c47fe 100644
---
a/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpDispatchHandler.java
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpDispatchHandler.java
@@ -17,12 +17,10 @@
package org.apache.seata.core.rpc.netty.http;
import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
@@ -36,41 +34,18 @@ import io.netty.handler.codec.http.multipart.Attribute;
import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;
import io.netty.handler.codec.http.multipart.InterfaceHttpData;
import org.apache.seata.common.rpc.http.HttpContext;
-import org.apache.seata.common.thread.NamedThreadFactory;
-import org.apache.seata.core.rpc.netty.NettyServerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Method;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
/**
* A Netty HTTP request handler that dispatches incoming requests to
corresponding controller methods
*/
-public class HttpDispatchHandler extends
SimpleChannelInboundHandler<HttpRequest> {
+public class HttpDispatchHandler extends BaseHttpChannelHandler<HttpRequest> {
private static final Logger LOGGER =
LoggerFactory.getLogger(HttpDispatchHandler.class);
- private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
- /**
- * HTTP request processing thread pool, independent of Netty IO threads,
to avoid blocking network processing.
- */
- private static final ExecutorService HTTP_HANDLER_THREADS = new
ThreadPoolExecutor(
- NettyServerConfig.getMinHttpPoolSize(),
- NettyServerConfig.getMaxHttpPoolSize(),
- NettyServerConfig.getHttpKeepAliveTime(),
- TimeUnit.SECONDS,
- new
LinkedBlockingQueue<>(NettyServerConfig.getMaxHttpTaskQueueSize()),
- new NamedThreadFactory("HTTPHandlerThread",
NettyServerConfig.getMaxHttpPoolSize()),
- new ThreadPoolExecutor.AbortPolicy());
-
- static {
- Runtime.getRuntime().addShutdownHook(new
Thread(HTTP_HANDLER_THREADS::shutdown));
- }
@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpRequest
httpRequest) {
@@ -82,13 +57,13 @@ public class HttpDispatchHandler extends
SimpleChannelInboundHandler<HttpRequest
HttpInvocation httpInvocation =
ControllerManager.getHttpInvocation(path);
if (httpInvocation == null) {
- sendErrorResponse(ctx, HttpResponseStatus.NOT_FOUND,
keepAlive);
+ sendErrorResponse(ctx, HttpResponseStatus.NOT_FOUND, false);
return;
}
- HttpContext httpContext = new HttpContext(httpRequest, ctx,
keepAlive);
+ HttpContext<HttpRequest> httpContext = new
HttpContext<>(httpRequest, ctx, keepAlive, HttpContext.HTTP_1_1);
ObjectNode requestDataNode = OBJECT_MAPPER.createObjectNode();
- requestDataNode.putIfAbsent("param",
ParameterParser.convertParamMap(queryStringDecoder.parameters()));
+ requestDataNode.set("param",
ParameterParser.convertParamMap(queryStringDecoder.parameters()));
if (httpRequest.method() == HttpMethod.POST) {
HttpPostRequestDecoder httpPostRequestDecoder = null;
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/http/SimpleHttp2Request.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/http/SimpleHttp2Request.java
new file mode 100644
index 0000000000..fe31fb6b7d
--- /dev/null
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/http/SimpleHttp2Request.java
@@ -0,0 +1,50 @@
+/*
+ * 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.seata.core.rpc.netty.http;
+
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http2.Http2Headers;
+
+public class SimpleHttp2Request {
+ private final HttpMethod method;
+ private final String path;
+ private final Http2Headers headers;
+ private final String body;
+
+ public SimpleHttp2Request(HttpMethod method, String path, Http2Headers
headers, String body) {
+ this.method = method;
+ this.path = path;
+ this.headers = headers;
+ this.body = body;
+ }
+
+ public HttpMethod getMethod() {
+ return method;
+ }
+
+ public String getPath() {
+ return path;
+ }
+
+ public Http2Headers getHeaders() {
+ return headers;
+ }
+
+ public String getBody() {
+ return body;
+ }
+}
diff --git
a/core/src/test/java/org/apache/seata/core/protocol/detector/Http2DetectorTest.java
b/core/src/test/java/org/apache/seata/core/protocol/detector/Http2DetectorTest.java
new file mode 100644
index 0000000000..0f9bb1a151
--- /dev/null
+++
b/core/src/test/java/org/apache/seata/core/protocol/detector/Http2DetectorTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.seata.core.protocol.detector;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandler;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class Http2DetectorTest {
+ @Test
+ void testDetectWithHttp2Prefix() {
+ byte[] http2Prefix = "PRI *
HTTP/2.0\r\n\r\nSM\r\n\r\n".getBytes(StandardCharsets.UTF_8);
+ ByteBuf buf = Unpooled.copiedBuffer(http2Prefix);
+ Http2Detector detector = new Http2Detector(new ChannelHandler[] {});
+ assertTrue(detector.detect(buf));
+ buf.release();
+ }
+
+ @Test
+ void testDetectWithNonHttp2() {
+ ByteBuf buf = Unpooled.copiedBuffer("NOTHTTP2",
StandardCharsets.UTF_8);
+ Http2Detector detector = new Http2Detector(new ChannelHandler[] {});
+ assertFalse(detector.detect(buf));
+ buf.release();
+ }
+
+ @Test
+ void testDetectWithShortBuffer() {
+ ByteBuf buf = Unpooled.copiedBuffer("PRI * HTTP/2.0",
StandardCharsets.UTF_8);
+ Http2Detector detector = new Http2Detector(new ChannelHandler[] {});
+ assertFalse(detector.detect(buf));
+ buf.release();
+ }
+
+ @Test
+ void testGetHandlersNotNull() {
+ ChannelHandler mockHandler = mock(ChannelHandler.class);
+ Http2Detector detector = new Http2Detector(new ChannelHandler[]
{mockHandler});
+ ChannelHandler[] handlers = detector.getHandlers();
+ assertNotNull(handlers);
+ assertEquals(2, handlers.length);
+ assertNotNull(handlers[0]);
+ assertNotNull(handlers[1]);
+ }
+}
diff --git
a/core/src/test/java/org/apache/seata/core/protocol/detector/HttpDetectorTest.java
b/core/src/test/java/org/apache/seata/core/protocol/detector/HttpDetectorTest.java
index f4bc1350ad..b75f45b7dc 100644
---
a/core/src/test/java/org/apache/seata/core/protocol/detector/HttpDetectorTest.java
+++
b/core/src/test/java/org/apache/seata/core/protocol/detector/HttpDetectorTest.java
@@ -19,14 +19,19 @@ package org.apache.seata.core.protocol.detector;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.HttpServerUpgradeHandler;
import org.apache.seata.core.rpc.netty.http.HttpDispatchHandler;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
class HttpDetectorTest {
private final HttpDetector httpDetector = new HttpDetector();
@@ -55,13 +60,93 @@ class HttpDetectorTest {
@Test
void testGetHandlers() {
ChannelHandler[] handlers = httpDetector.getHandlers();
- assertEquals(3, handlers.length);
+ assertEquals(6, handlers.length);
assertInstanceOf(HttpServerCodec.class, handlers[0]);
- assertInstanceOf(HttpObjectAggregator.class, handlers[1]);
- assertInstanceOf(HttpDispatchHandler.class, handlers[2]);
+ assertInstanceOf(HttpServerUpgradeHandler.class, handlers[1]);
+ assertInstanceOf(ChannelHandler.class, handlers[2]); //
upgradeCleanupHandler
+ assertInstanceOf(HttpObjectAggregator.class, handlers[3]);
+ assertInstanceOf(HttpDispatchHandler.class, handlers[4]);
+ assertInstanceOf(ChannelHandler.class, handlers[5]); //
finalExceptionHandler
// Verify aggregator size
- HttpObjectAggregator aggregator = (HttpObjectAggregator) handlers[1];
+ HttpObjectAggregator aggregator = (HttpObjectAggregator) handlers[3];
assertEquals(1048576, aggregator.maxContentLength());
}
+
+ @Test
+ void testDetectWithShortBuffer() {
+ ByteBuf buf = Unpooled.copiedBuffer("GET", StandardCharsets.UTF_8);
+ assertFalse(httpDetector.detect(buf));
+ buf.release();
+ }
+
+ @Test
+ void testStartsWithNegativeBranch() {
+ ByteBuf buf = Unpooled.copiedBuffer("GE", StandardCharsets.UTF_8);
+ try {
+ java.lang.reflect.Method m =
+ HttpDetector.class.getDeclaredMethod("startsWith",
ByteBuf.class, String.class);
+ m.setAccessible(true);
+ boolean result = (boolean) m.invoke(httpDetector, buf, "GET");
+ assertFalse(result);
+ } catch (Exception e) {
+ fail(e);
+ }
+ buf.release();
+ }
+
+ @Test
+ void testUpgradeHandlerNonHttp2() {
+ try {
+ java.lang.reflect.Method m = HttpDetector.class.getDeclaredMethod(
+ "getHttpServerUpgradeHandler",
io.netty.handler.codec.http.HttpServerCodec.class);
+ m.setAccessible(true);
+ io.netty.handler.codec.http.HttpServerCodec codec = new
io.netty.handler.codec.http.HttpServerCodec();
+ Object handler = m.invoke(null, codec);
+ assertNotNull(handler);
+ assertTrue(handler instanceof
io.netty.handler.codec.http.HttpServerUpgradeHandler);
+ } catch (Exception e) {
+ fail(e);
+ }
+ }
+
+ @Test
+ void testUpgradeCleanupHandlerEvent() throws Exception {
+ ChannelHandler[] handlers = httpDetector.getHandlers();
+ ChannelInboundHandlerAdapter upgradeCleanupHandler =
(ChannelInboundHandlerAdapter) handlers[2];
+ HttpServerUpgradeHandler.UpgradeEvent mockEvent =
mock(HttpServerUpgradeHandler.UpgradeEvent.class);
+ ChannelHandlerContext mockCtx = mock(ChannelHandlerContext.class);
+ ChannelPipeline mockPipeline = mock(ChannelPipeline.class);
+ when(mockCtx.pipeline()).thenReturn(mockPipeline);
+ upgradeCleanupHandler.userEventTriggered(mockCtx, mockEvent);
+ verify(mockPipeline).remove(HttpObjectAggregator.class);
+ verify(mockPipeline).remove(HttpDispatchHandler.class);
+ }
+
+ @Test
+ void testUpgradeCleanupHandlerNonUpgradeEvent() throws Exception {
+ ChannelHandler[] handlers = httpDetector.getHandlers();
+ ChannelHandlerContext mockCtx = mock(ChannelHandlerContext.class);
+ ChannelInboundHandlerAdapter upgradeCleanupHandler =
(ChannelInboundHandlerAdapter) handlers[2];
+ upgradeCleanupHandler.userEventTriggered(mockCtx, "not-upgrade-event");
+ }
+
+ @Test
+ void testFinalExceptionHandler() throws Exception {
+ ChannelHandler[] handlers = httpDetector.getHandlers();
+ ChannelHandler finalExceptionHandler = handlers[5];
+ ChannelHandlerContext mockCtx = mock(ChannelHandlerContext.class);
+ finalExceptionHandler.exceptionCaught(mockCtx, new
java.io.IOException("test"));
+ finalExceptionHandler.exceptionCaught(mockCtx, new
RuntimeException("test"));
+ verify(mockCtx, times(2)).close();
+ }
+
+ @Test
+ void testFinalExceptionHandlerNonIOException() throws Exception {
+ ChannelHandler[] handlers = httpDetector.getHandlers();
+ ChannelInboundHandlerAdapter finalExceptionHandler =
(ChannelInboundHandlerAdapter) handlers[5];
+ ChannelHandlerContext mockCtx = mock(ChannelHandlerContext.class);
+ finalExceptionHandler.exceptionCaught(mockCtx, new
IllegalArgumentException("test"));
+ verify(mockCtx).close();
+ }
}
diff --git
a/core/src/test/java/org/apache/seata/core/rpc/netty/http/Http2HttpHandlerTest.java
b/core/src/test/java/org/apache/seata/core/rpc/netty/http/Http2HttpHandlerTest.java
new file mode 100644
index 0000000000..c4ca1e0dda
--- /dev/null
+++
b/core/src/test/java/org/apache/seata/core/rpc/netty/http/Http2HttpHandlerTest.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.seata.core.rpc.netty.http;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.handler.codec.http2.Http2HeadersFrame;
+import io.netty.handler.codec.http2.Http2StreamFrame;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class Http2HttpHandlerTest {
+ private Http2HttpHandler handler;
+ private EmbeddedChannel channel;
+ private TestController testController = new TestController();
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ static class TestController {
+ public String handleRequest(String param) {
+ return "Processed: " + param;
+ }
+ }
+
+ @BeforeEach
+ void setUp() throws Exception {
+ handler = new Http2HttpHandler();
+ channel = new EmbeddedChannel(handler);
+ Method method = TestController.class.getMethod("handleRequest",
String.class);
+ ParamMetaData paramMetaData = new ParamMetaData();
+
paramMetaData.setParamConvertType(ParamMetaData.ParamConvertType.REQUEST_PARAM);
+ paramMetaData.setParamName("param");
+ ParamMetaData[] paramMetaDatas = new ParamMetaData[] {paramMetaData};
+ HttpInvocation invocation = new HttpInvocation();
+ invocation.setController(testController);
+ invocation.setMethod(method);
+ invocation.setPath("/test");
+ invocation.setParamMetaData(paramMetaDatas);
+ ControllerManager.addHttpInvocation(invocation);
+ }
+
+ private Http2StreamFrame waitForHttp2Response(long timeoutMs) {
+ long startTime = System.currentTimeMillis();
+ Http2StreamFrame response = null;
+ while (response == null && (System.currentTimeMillis() - startTime) <
timeoutMs) {
+ response = channel.readOutbound();
+ if (response == null) {
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for
response", e);
+ }
+ }
+ }
+ return response;
+ }
+
+ @Test
+ void testHttp2GetRequestWithParameters() throws Exception {
+ Http2Headers headers = new DefaultHttp2Headers();
+ headers.method("GET");
+ headers.path("/test?param=testValue");
+ Http2HeadersFrame headersFrame = new DefaultHttp2HeadersFrame(headers,
true);
+ channel.writeInbound(headersFrame);
+
+ Http2StreamFrame responseHeadersFrame = waitForHttp2Response(5000);
+ assertNotNull(responseHeadersFrame);
+ assertTrue(responseHeadersFrame instanceof DefaultHttp2HeadersFrame);
+ DefaultHttp2HeadersFrame respHeaders = (DefaultHttp2HeadersFrame)
responseHeadersFrame;
+ assertEquals("200", respHeaders.headers().status().toString());
+
+ Http2StreamFrame responseDataFrame = waitForHttp2Response(5000);
+ assertNotNull(responseDataFrame);
+ assertTrue(responseDataFrame instanceof DefaultHttp2DataFrame);
+ DefaultHttp2DataFrame respData = (DefaultHttp2DataFrame)
responseDataFrame;
+ String content = respData.content().toString(StandardCharsets.UTF_8);
+ assertTrue(content.contains("Processed: testValue"));
+ }
+
+ @Test
+ void testHttp2RequestToNonexistentPath() {
+ Http2Headers headers = new DefaultHttp2Headers();
+ headers.method("GET");
+ headers.path("/notfound");
+ Http2HeadersFrame headersFrame = new DefaultHttp2HeadersFrame(headers,
true);
+ channel.writeInbound(headersFrame);
+
+ Http2StreamFrame responseHeadersFrame = channel.readOutbound();
+ assertTrue(responseHeadersFrame instanceof DefaultHttp2HeadersFrame);
+ DefaultHttp2HeadersFrame respHeaders = (DefaultHttp2HeadersFrame)
responseHeadersFrame;
+ assertEquals("404", respHeaders.headers().status().toString());
+ }
+
+ @Test
+ void testHttp2PostRequestWithJsonBody() throws Exception {
+ String json = OBJECT_MAPPER.writeValueAsString(new HashMap<String,
Object>() {
+ {
+ put("foo", "bar");
+ }
+ });
+ Http2Headers headers = new DefaultHttp2Headers();
+ headers.method("POST");
+ headers.path("/test?param=jsonValue");
+ Http2HeadersFrame headersFrame = new DefaultHttp2HeadersFrame(headers,
false);
+ channel.writeInbound(headersFrame);
+ DefaultHttp2DataFrame dataFrame =
+ new DefaultHttp2DataFrame(Unpooled.copiedBuffer(json,
StandardCharsets.UTF_8), true);
+ channel.writeInbound(dataFrame);
+
+ Http2StreamFrame frame1 = null, frame2 = null;
+ long deadline = System.currentTimeMillis() + 5000; // 最多等5秒
+ while ((frame1 == null || frame2 == null) &&
System.currentTimeMillis() < deadline) {
+ if (frame1 == null) frame1 = channel.readOutbound();
+ if (frame2 == null) frame2 = channel.readOutbound();
+ if (frame1 == null || frame2 == null) Thread.sleep(500);
+ }
+ assertNotNull(frame1);
+ assertNotNull(frame2);
+ DefaultHttp2HeadersFrame respHeaders;
+ DefaultHttp2DataFrame respData;
+ if (frame1 instanceof DefaultHttp2HeadersFrame) {
+ respHeaders = (DefaultHttp2HeadersFrame) frame1;
+ respData = (DefaultHttp2DataFrame) frame2;
+ } else {
+ respHeaders = (DefaultHttp2HeadersFrame) frame2;
+ respData = (DefaultHttp2DataFrame) frame1;
+ }
+ assertEquals("200", respHeaders.headers().status().toString());
+ String content = respData.content().toString(StandardCharsets.UTF_8);
+ assertTrue(content.contains("Processed: jsonValue"));
+ }
+
+ @Test
+ void testHttp2BadRequest() {
+ Http2Headers headers = new DefaultHttp2Headers();
+ Http2HeadersFrame headersFrame = new DefaultHttp2HeadersFrame(headers,
true);
+ channel.writeInbound(headersFrame);
+ Http2StreamFrame responseHeadersFrame = channel.readOutbound();
+ assertTrue(responseHeadersFrame instanceof DefaultHttp2HeadersFrame);
+ DefaultHttp2HeadersFrame respHeaders = (DefaultHttp2HeadersFrame)
responseHeadersFrame;
+ assertEquals("400", respHeaders.headers().status().toString());
+ }
+
+ @org.junit.jupiter.api.AfterEach
+ void tearDown() throws Exception {
+ // Clean up ControllerManager
+ Field field =
ControllerManager.class.getDeclaredField("HTTP_CONTROLLER_MAP");
+ field.setAccessible(true);
+ Map<String, HttpInvocation> map = (Map<String, HttpInvocation>)
field.get(null);
+ map.clear();
+ }
+}
diff --git
a/server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java
b/server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java
index c3ac1bd5ea..81b75ec716 100644
---
a/server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java
+++
b/server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java
@@ -49,7 +49,7 @@ public class ClusterWatcherManager implements
ClusterChangeListener {
private final Logger logger = LoggerFactory.getLogger(getClass());
- private static final Map<String, Queue<Watcher<?>>> WATCHERS = new
ConcurrentHashMap<>();
+ private static final Map<String, Queue<Watcher<HttpContext>>> WATCHERS =
new ConcurrentHashMap<>();
private static final Map<String, Long> GROUP_UPDATE_TIME = new
ConcurrentHashMap<>();
@@ -92,13 +92,13 @@ public class ClusterWatcherManager implements
ClusterChangeListener {
}
}
- private void notifyWatcher(Watcher<?> watcher) {
+ private void notifyWatcher(Watcher<HttpContext> watcher) {
watcher.setDone(true);
sendWatcherResponse(watcher, HttpResponseStatus.OK);
}
- private void sendWatcherResponse(Watcher<?> watcher, HttpResponseStatus
nettyStatus) {
- Object context = watcher.getAsyncContext();
+ private void sendWatcherResponse(Watcher<HttpContext> watcher,
HttpResponseStatus nettyStatus) {
+ HttpContext context = watcher.getAsyncContext();
if (!(context instanceof HttpContext)) {
logger.warn(
"Unsupported context type for watcher on group {}: {}",
@@ -106,25 +106,27 @@ public class ClusterWatcherManager implements
ClusterChangeListener {
context != null ? context.getClass().getName() : "null");
return;
}
- HttpContext httpContext = (HttpContext) context;
- ChannelHandlerContext ctx = httpContext.getContext();
- if (ctx.channel().isActive()) {
- HttpResponse response =
- new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
nettyStatus, Unpooled.EMPTY_BUFFER);
- response.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0);
+ ChannelHandlerContext ctx = context.getContext();
+ if (!context.isHttp2()) {
+ if (ctx.channel().isActive()) {
+ HttpResponse response =
+ new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
nettyStatus, Unpooled.EMPTY_BUFFER);
+ response.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0);
- if (!httpContext.isKeepAlive()) {
-
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ if (!context.isKeepAlive()) {
+
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ } else {
+ ctx.writeAndFlush(response);
+ }
} else {
- ctx.writeAndFlush(response);
+ logger.warn(
+ "Netty channel is not active for watcher on group {},
cannot send response.",
+ watcher.getGroup());
}
- } else {
- logger.warn(
- "Netty channel is not active for watcher on group {},
cannot send response.", watcher.getGroup());
}
}
- public void registryWatcher(Watcher<?> watcher) {
+ public void registryWatcher(Watcher<HttpContext> watcher) {
String group = watcher.getGroup();
Long term = GROUP_UPDATE_TIME.get(group);
if (term == null || watcher.getTerm() >= term) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
