optimize: select channel handles based on protocol versions (#6634)

Author: funky-eyes

changes/en-us/2.x.md

@@ -3,7 +3,7 @@ Add changes here for all PR submitted to the 2.x branch.
 <!-- Please add the `changes` to the following location(feature/bugfix/optimize/test) based on the type of PR -->
 
 ### feature:
-
+- [[#6226](https://github.com/apache/incubator-seata/pull/6226)] multi-version seata protocol support
 
 ### bugfix:
 - [[#6592](https://github.com/apache/incubator-seata/pull/6592)] fix @Async annotation not working in ClusterWatcherManager
@@ -16,8 +16,10 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#6499](https://github.com/apache/incubator-seata/pull/6499)] split the task thread pool for committing and rollbacking statuses
 - [[#6208](https://github.com/apache/incubator-seata/pull/6208)] optimize : load SeataSerializer by version
 - [[#6209](https://github.com/apache/incubator-seata/pull/6209)] Eliminate RpcMessage and Encoder/Decoder dependencies
+- [[#6634](https://github.com/apache/incubator-seata/pull/6634)] select channel handles based on protocol versions
 - [[#6523](https://github.com/apache/incubator-seata/pull/6523)] upgrade alibaba/druid version to 1.2.20
 
+
 ### refactor:
 - [[#6534](https://github.com/apache/incubator-seata/pull/6534)] optimize: send async response
 
@@ -36,6 +38,7 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [liuqiufeng](https://github.com/liuqiufeng)
 - [God-Gan](https://github.com/God-Gan)
 - [Bughue](https://github.com/Bughue)
+- [funky-eyes](https://github.com/funky-eyes)
 - [tanyaofei](https://github.com/tanyaofei)
 
 Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.

changes/zh-cn/2.x.md

@@ -3,6 +3,7 @@
 <!-- 请根据PR的类型添加 `变更记录` 到以下对应位置(feature/bugfix/optimize/test) 下 -->
 
 ### feature:
+- [[#6226](https://github.com/apache/incubator-seata/pull/6226)] 支持seata私有协议多版本兼容
 
 ### bugfix:
 - [[#6592](https://github.com/apache/incubator-seata/pull/6592)] fix @Async注解ClusterWatcherManager中不生效的问题
@@ -15,8 +16,10 @@
 - [[#6499](https://github.com/apache/incubator-seata/pull/6499)] 拆分 committing 和 rollbacking 状态的任务线程池
 - [[#6208](https://github.com/apache/incubator-seata/pull/6208)] 支持多版本的Seata序列化
 - [[#6209](https://github.com/apache/incubator-seata/pull/6209)] 解开 RpcMessage 和 Encoder/Decoder 的互相依赖
+- [[#6634](https://github.com/apache/incubator-seata/pull/6634)] 根据协议版本指定channel handle
 - [[#6523](https://github.com/apache/incubator-seata/pull/6523)] 升级 alibaba/druid 的版本到1.2.20
 
+
 ### refactor:
 - [[#6534](https://github.com/apache/incubator-seata/pull/6534)] 优化: 发送异步响应
 
@@ -34,6 +37,7 @@
 - [liuqiufeng](https://github.com/liuqiufeng)
 - [God-Gan](https://github.com/God-Gan)
 - [Bughue](https://github.com/Bughue)
+- [funky-eyes](https://github.com/funky-eyes)
 - [tanyaofei](https://github.com/tanyaofei)
 
 

core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java

@@ -409,10 +409,8 @@ class ClientHandler extends ChannelDuplexHandler {
 
         @Override
         public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
-            RpcMessage rpcMessage = null;
-            if (msg instanceof ProtocolRpcMessage) {
-                rpcMessage = ((ProtocolRpcMessage) msg).protocolMsg2RpcMsg();
-                processMessage(ctx, rpcMessage);
+            if (msg instanceof RpcMessage) {
+                processMessage(ctx, (RpcMessage)msg);
             } else {
                 LOGGER.error("rpcMessage type error");
             }

core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingServer.java

@@ -69,7 +69,7 @@ public Object sendSyncRequest(String resourceId, String clientId, Object msg, bo
         if (channel == null) {
             throw new RuntimeException("rm client is not connected. dbkey:" + resourceId + ",clientId:" + clientId);
         }
-        RpcMessage rpcMessage = buildRequestMessage(channel, msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
+        RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
         return super.sendSync(channel, rpcMessage, NettyServerConfig.getRpcRequestTimeout());
     }
 
@@ -78,7 +78,7 @@ public Object sendSyncRequest(Channel channel, Object msg) throws TimeoutExcepti
         if (channel == null) {
             throw new RuntimeException("client is not connected");
         }
-        RpcMessage rpcMessage = buildRequestMessage(channel, msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
+        RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
         return super.sendSync(channel, rpcMessage, NettyServerConfig.getRpcRequestTimeout());
     }
 
@@ -87,7 +87,7 @@ public void sendAsyncRequest(Channel channel, Object msg) {
         if (channel == null) {
             throw new RuntimeException("client is not connected");
         }
-        RpcMessage rpcMessage = buildRequestMessage(channel, msg, ProtocolConstants.MSGTYPE_RESQUEST_ONEWAY);
+        RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_ONEWAY);
         super.sendAsync(channel, rpcMessage);
     }
 
@@ -98,7 +98,7 @@ public void sendAsyncResponse(RpcMessage rpcMessage, Channel channel, Object msg
             clientChannel = ChannelManager.getSameClientChannel(channel);
         }
         if (clientChannel != null) {
-            RpcMessage rpcMsg = buildResponseMessage(channel, rpcMessage, msg, msg instanceof HeartbeatMessage
+            RpcMessage rpcMsg = buildResponseMessage(rpcMessage, msg, msg instanceof HeartbeatMessage
                 ? ProtocolConstants.MSGTYPE_HEARTBEAT_RESPONSE
                 : ProtocolConstants.MSGTYPE_RESPONSE);
             super.sendAsync(clientChannel, rpcMsg);
@@ -108,21 +108,6 @@ public void sendAsyncResponse(RpcMessage rpcMessage, Channel channel, Object msg
     }
 
 
-    private RpcMessage buildResponseMessage(Channel channel, RpcMessage fromRpcMessage, Object msg, byte messageType) {
-        RpcMessage rpcMessage = super.buildResponseMessage(fromRpcMessage, msg, messageType);
-        RpcContext rpcContext = ChannelManager.getContextFromIdentified(channel);
-        rpcMessage.setOtherSideVersion(rpcContext.getVersion());
-        return rpcMessage;
-    }
-
-    protected RpcMessage buildRequestMessage(Channel channel, Object msg, byte messageType) {
-        RpcMessage rpcMessage = super.buildRequestMessage(msg, messageType);
-        RpcContext rpcContext = ChannelManager.getContextFromIdentified(channel);
-        rpcMessage.setOtherSideVersion(rpcContext.getVersion());
-        return rpcMessage;
-    }
-
-
     @Override
     public void registerProcessor(int messageType, RemotingProcessor processor, ExecutorService executor) {
         Pair<RemotingProcessor, ExecutorService> pair = new Pair<>(processor, executor);
@@ -179,10 +164,8 @@ class ServerHandler extends ChannelDuplexHandler {
          */
         @Override
         public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
-            RpcMessage rpcMessage = null;
-            if (msg instanceof ProtocolRpcMessage) {
-                rpcMessage = ((ProtocolRpcMessage) msg).protocolMsg2RpcMsg();
-                processMessage(ctx, rpcMessage);
+            if (msg instanceof RpcMessage) {
+                processMessage(ctx, (RpcMessage)msg);
             } else {
                 LOGGER.error("rpcMessage type error");
             }

core/src/main/java/org/apache/seata/core/rpc/netty/CompatibleProtocolEncoder.java

@@ -18,12 +18,15 @@
 
 import com.google.common.collect.ImmutableMap;
 import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import org.apache.seata.core.exception.DecodeException;
 import org.apache.seata.core.protocol.ProtocolConstants;
 import org.apache.seata.core.rpc.netty.v0.ProtocolDecoderV0;
+import org.apache.seata.core.rpc.netty.v0.ProtocolEncoderV0;
 import org.apache.seata.core.rpc.netty.v1.ProtocolDecoderV1;
+import org.apache.seata.core.rpc.netty.v1.ProtocolEncoderV1;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -52,17 +55,26 @@
  * <li>Body Length: Full Length - Head Length</li>
  * </p>
  */
-public class CompatibleProtocolDecoder extends LengthFieldBasedFrameDecoder {
+public class MultiProtocolDecoder extends LengthFieldBasedFrameDecoder {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(CompatibleProtocolDecoder.class);
-    private static Map<Byte, ProtocolDecoder> protocolDecoderMap;
+    private static final Logger LOGGER = LoggerFactory.getLogger(MultiProtocolDecoder.class);
+    private final Map<Byte, ProtocolDecoder> protocolDecoderMap;
 
-    public CompatibleProtocolDecoder() {
+    private final Map<Byte, ProtocolEncoder> protocolEncoderMap;
+    
+    private final ChannelHandler[] channelHandlers;
+
+    public MultiProtocolDecoder(ChannelHandler... channelHandlers) {
         // default is 8M
-        this(ProtocolConstants.MAX_FRAME_LENGTH);
+        this(ProtocolConstants.MAX_FRAME_LENGTH, channelHandlers);
     }
 
-    public CompatibleProtocolDecoder(int maxFrameLength) {
+    public MultiProtocolDecoder() {
+        // default is 8M
+        this(ProtocolConstants.MAX_FRAME_LENGTH, null);
+    }
+
+    public MultiProtocolDecoder(int maxFrameLength, ChannelHandler[] channelHandlers) {
         /*
         int maxFrameLength,      
         int lengthFieldOffset,  magic code is 2B, and version is 1B, and then FullLength. so value is 3
@@ -71,10 +83,13 @@ int lengthFieldLength,  FullLength is int(4B). so values is 4
         int initialBytesToStrip we will check magic code and version self, so do not strip any bytes. so values is 0
         */
         super(maxFrameLength, 3, 4, -7, 0);
-        protocolDecoderMap = ImmutableMap.<Byte, ProtocolDecoder>builder()
-                .put(ProtocolConstants.VERSION_0, new ProtocolDecoderV0())
-                .put(ProtocolConstants.VERSION_1, new ProtocolDecoderV1())
-                .build();
+        this.protocolDecoderMap =
+            ImmutableMap.<Byte, ProtocolDecoder>builder().put(ProtocolConstants.VERSION_0, new ProtocolDecoderV0())
+                .put(ProtocolConstants.VERSION_1, new ProtocolDecoderV1()).build();
+        this.protocolEncoderMap =
+            ImmutableMap.<Byte, ProtocolEncoder>builder().put(ProtocolConstants.VERSION_0, new ProtocolEncoderV0())
+                .put(ProtocolConstants.VERSION_1, new ProtocolEncoderV1()).build();
+        this.channelHandlers = channelHandlers;
     }
 
     @Override
@@ -93,16 +108,23 @@ protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception
 
             if (decoded instanceof ByteBuf) {
                 frame = (ByteBuf) decoded;
+                ProtocolDecoder decoder = protocolDecoderMap.get(version);
+                ProtocolEncoder encoder = protocolEncoderMap.get(version);
                 try {
-                    ProtocolDecoder decoder = protocolDecoderMap.get(version);
-                    if (decoder == null) {
+                    if (decoder == null || encoder == null) {
                         throw new UnsupportedOperationException("Unsupported version: " + version);
                     }
                     return decoder.decodeFrame(frame);
                 } finally {
                     if (version != ProtocolConstants.VERSION_0) {
                         frame.release();
                     }
+                    ctx.pipeline().addLast((ChannelHandler)decoder);
+                    ctx.pipeline().addLast((ChannelHandler)encoder);
+                    if (channelHandlers != null) {
+                        ctx.pipeline().addLast(channelHandlers);
+                    }
+                    ctx.pipeline().remove(this);
                 }
             }
         } catch (Exception exx) {

…rpc/netty/CompatibleProtocolDecoder.java …core/rpc/netty/MultiProtocolDecoder.javacore/src/main/java/org/apache/seata/core/rpc/netty/CompatibleProtocolDecoder.java renamed to core/src/main/java/org/apache/seata/core/rpc/netty/MultiProtocolDecoder.java core/src/main/java/org/apache/seata/core/rpc/netty/CompatibleProtocolDecoder.java renamed to core/src/main/java/org/apache/seata/core/rpc/netty/MultiProtocolDecoder.java

@@ -35,6 +35,8 @@
 import org.apache.seata.common.exception.FrameworkException;
 import org.apache.seata.common.thread.NamedThreadFactory;
 import org.apache.seata.core.rpc.RemotingBootstrap;
+import org.apache.seata.core.rpc.netty.v1.ProtocolDecoderV1;
+import org.apache.seata.core.rpc.netty.v1.ProtocolEncoderV1;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -128,12 +130,12 @@ public void start() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     ChannelPipeline pipeline = ch.pipeline();
-                    pipeline.addLast(
-                        new IdleStateHandler(nettyClientConfig.getChannelMaxReadIdleSeconds(),
+                    pipeline
+                        .addLast(new IdleStateHandler(nettyClientConfig.getChannelMaxReadIdleSeconds(),
                             nettyClientConfig.getChannelMaxWriteIdleSeconds(),
                             nettyClientConfig.getChannelMaxAllIdleSeconds()))
-                        .addLast(new CompatibleProtocolDecoder())
-                        .addLast(new CompatibleProtocolEncoder());
+                        .addLast(new ProtocolDecoderV1())
+                        .addLast(new ProtocolEncoderV1());
                     if (channelHandlers != null) {
                         addChannelPipelineLast(ch, channelHandlers);
                     }

core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientBootstrap.java

@@ -158,13 +158,10 @@ public void start() {
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) {
-                    ch.pipeline().addLast(new IdleStateHandler(nettyServerConfig.getChannelMaxReadIdleSeconds(), 0, 0))
-                        .addLast(new CompatibleProtocolDecoder())
-                        .addLast(new CompatibleProtocolEncoder());
-                    if (channelHandlers != null) {
-                        addChannelPipelineLast(ch, channelHandlers);
-                    }
-
+                    MultiProtocolDecoder multiProtocolDecoder = new MultiProtocolDecoder(channelHandlers);
+                    ch.pipeline()
+                        .addLast(new IdleStateHandler(nettyServerConfig.getChannelMaxReadIdleSeconds(), 0, 0))
+                        .addLast(multiProtocolDecoder);
                 }
             });
 

core/src/main/java/org/apache/seata/core/rpc/netty/NettyServerBootstrap.java

@@ -17,13 +17,14 @@
 package org.apache.seata.core.rpc.netty;
 
 import io.netty.buffer.ByteBuf;
+import org.apache.seata.core.protocol.RpcMessage;
 
 /**
  * the protocol decoder
  *
  **/
 public interface ProtocolDecoder {
 
-    ProtocolRpcMessage decodeFrame(ByteBuf in);
+    RpcMessage decodeFrame(ByteBuf in);
 
 }