feature : multi-version protocol control

core/src/main/java/io/seata/core/protocol/ProtocolConstants.java

@@ -31,10 +31,17 @@ public interface ProtocolConstants {
      */
     byte[] MAGIC_CODE_BYTES = {(byte) 0xda, (byte) 0xda};
 
+    /**
+     * Old protocol version
+     */
+    byte VERSION_0 = 0;
+
     /**
      * Protocol version
      */
-    byte VERSION = 1;
+    byte VERSION_1 = 1;
+
+    byte VERSION_CURRENT = VERSION_1;
 
     /**
      * Max frame length

core/src/main/java/io/seata/core/protocol/RpcMessage.java

@@ -35,6 +35,8 @@ public class RpcMessage implements Serializable {
     private Map<String, String> headMap = new HashMap<>();
     private Object body;
 
+    private String version;
+
     /**
      * Gets id.
      *
@@ -169,6 +171,14 @@ public void setMessageType(byte messageType) {
         this.messageType = messageType;
     }
 
+    public String getVersion() {
+        return version;
+    }
+
+    public void setVersion(String version) {
+        this.version = version;
+    }
+
     @Override
     public String toString() {
         return StringUtils.toString(this);

core/src/main/java/io/seata/core/protocol/Version.java

@@ -37,7 +37,7 @@ public class Version {
      * The constant CURRENT.
      */
     private static final String CURRENT = VersionInfo.VERSION;
-    private static final String VERSION_0_7_1 = "0.7.1";
+    public static final String VERSION_0_7_1 = "0.7.1";
     private static final String VERSION_1_5_0 = "1.5.0";
     private static final int MAX_VERSION_DOT = 3;
 
@@ -89,7 +89,7 @@ public static void checkVersion(String version) throws IncompatibleVersionExcept
         long current = convertVersion(CURRENT);
         long clientVersion = convertVersion(version);
         long divideVersion = convertVersion(VERSION_0_7_1);
-        if ((current > divideVersion && clientVersion < divideVersion) || (current < divideVersion && clientVersion > divideVersion)) {
+        if (current < divideVersion && clientVersion > divideVersion) {
             throw new IncompatibleVersionException("incompatible client version:" + version);
         }
     }
@@ -150,6 +150,16 @@ public static long convertVersionNotThrowException(String version) {
         return -1;
     }
 
+    public static byte calcProtocolVersion(String sdkVersion) throws IncompatibleVersionException {
+        long version = convertVersion(sdkVersion);
+        long v0 = convertVersion(VERSION_0_7_1);
+        if (version <= v0) {
+            return ProtocolConstants.VERSION_0;
+        } else {
+            return ProtocolConstants.VERSION_1;
+        }
+    }
+
     private static long calculatePartValue(String partNumeric, int size, int index) {
         return Long.parseLong(partNumeric) * Double.valueOf(Math.pow(100, size - index)).longValue();
     }

core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemoting.java

@@ -42,6 +42,7 @@
 import io.seata.core.protocol.MessageTypeAware;
 import io.seata.core.protocol.ProtocolConstants;
 import io.seata.core.protocol.RpcMessage;
+import io.seata.core.protocol.Version;
 import io.seata.core.rpc.Disposable;
 import io.seata.core.rpc.hook.RpcHook;
 import io.seata.core.rpc.processor.Pair;
@@ -63,7 +64,7 @@ public abstract class AbstractNettyRemoting implements Disposable {
      * The Timer executor.
      */
     protected final ScheduledExecutorService timerExecutor = new ScheduledThreadPoolExecutor(1,
-        new NamedThreadFactory("timeoutChecker", 1, true));
+            new NamedThreadFactory("timeoutChecker", 1, true));
     /**
      * The Message executor.
      */
@@ -113,7 +114,7 @@ public void run() {
                         futures.remove(entry.getKey());
                         RpcMessage rpcMessage = future.getRequestMessage();
                         future.setResultMessage(new TimeoutException(String
-                            .format("msgId: %s ,msgType: %s ,msg: %s ,request timeout", rpcMessage.getId(), String.valueOf(rpcMessage.getMessageType()), rpcMessage.getBody().toString())));
+                                .format("msgId: %s ,msgType: %s ,msg: %s ,request timeout", rpcMessage.getId(), String.valueOf(rpcMessage.getMessageType()), rpcMessage.getBody().toString())));
                         if (LOGGER.isDebugEnabled()) {
                             LOGGER.debug("timeout clear future: {}", entry.getValue().getRequestMessage().getBody());
                         }
@@ -200,7 +201,7 @@ protected Object sendSync(Channel channel, RpcMessage rpcMessage, long timeoutMi
             return result;
         } catch (Exception exx) {
             LOGGER.error("wait response error:{},ip:{},request:{}", exx.getMessage(), channel.remoteAddress(),
-                rpcMessage.getBody());
+                    rpcMessage.getBody());
             if (exx instanceof TimeoutException) {
                 throw (TimeoutException) exx;
             } else {
@@ -219,7 +220,7 @@ protected void sendAsync(Channel channel, RpcMessage rpcMessage) {
         channelWritableCheck(channel, rpcMessage.getBody());
         if (LOGGER.isDebugEnabled()) {
             LOGGER.debug("write message:" + rpcMessage.getBody() + ", channel:" + channel + ",active?"
-                + channel.isActive() + ",writable?" + channel.isWritable() + ",isopen?" + channel.isOpen());
+                    + channel.isActive() + ",writable?" + channel.isWritable() + ",isopen?" + channel.isOpen());
         }
 
         doBeforeRpcHooks(ChannelUtil.getAddressFromChannel(channel), rpcMessage);
@@ -232,22 +233,32 @@ protected void sendAsync(Channel channel, RpcMessage rpcMessage) {
     }
 
     protected RpcMessage buildRequestMessage(Object msg, byte messageType) {
+        return buildRequestMessage(msg, messageType, Version.getCurrent());
+    }
+
+    protected RpcMessage buildRequestMessage(Object msg, byte messageType, String version) {
         RpcMessage rpcMessage = new RpcMessage();
         rpcMessage.setId(getNextMessageId());
         rpcMessage.setMessageType(messageType);
         rpcMessage.setCodec(ProtocolConstants.CONFIGURED_CODEC);
         rpcMessage.setCompressor(ProtocolConstants.CONFIGURED_COMPRESSOR);
         rpcMessage.setBody(msg);
+        rpcMessage.setVersion(version);
         return rpcMessage;
     }
 
     protected RpcMessage buildResponseMessage(RpcMessage rpcMessage, Object msg, byte messageType) {
+        return buildResponseMessage(rpcMessage, msg, messageType, Version.getCurrent());
+    }
+
+    protected RpcMessage buildResponseMessage(RpcMessage rpcMessage, Object msg, byte messageType, String version) {
         RpcMessage rpcMsg = new RpcMessage();
         rpcMsg.setMessageType(messageType);
         rpcMsg.setCodec(rpcMessage.getCodec()); // same with request
         rpcMsg.setCompressor(rpcMessage.getCompressor());
         rpcMsg.setBody(msg);
         rpcMsg.setId(rpcMessage.getId());
+        rpcMsg.setVersion(version);
         return rpcMsg;
     }
 

core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingClient.java

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

core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingServer.java

@@ -69,16 +69,20 @@ 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(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
+        RpcContext rpcContext = ChannelManager.getContextFromIdentified(channel);
+        RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC, rpcContext.getVersion());
         return super.sendSync(channel, rpcMessage, NettyServerConfig.getRpcRequestTimeout());
     }
 
+
+
     @Override
     public Object sendSyncRequest(Channel channel, Object msg) throws TimeoutException {
         if (channel == null) {
             throw new RuntimeException("client is not connected");
         }
-        RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
+        RpcContext rpcContext = ChannelManager.getContextFromIdentified(channel);
+        RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC, rpcContext.getVersion());
         return super.sendSync(channel, rpcMessage, NettyServerConfig.getRpcRequestTimeout());
     }
 
@@ -87,7 +91,8 @@ public void sendAsyncRequest(Channel channel, Object msg) {
         if (channel == null) {
             throw new RuntimeException("client is not connected");
         }
-        RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_ONEWAY);
+        RpcContext rpcContext = ChannelManager.getContextFromIdentified(channel);
+        RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_ONEWAY, rpcContext.getVersion());
         super.sendAsync(channel, rpcMessage);
     }
 
@@ -98,9 +103,10 @@ public void sendAsyncResponse(RpcMessage rpcMessage, Channel channel, Object msg
             clientChannel = ChannelManager.getSameClientChannel(channel);
         }
         if (clientChannel != null) {
+            RpcContext rpcContext = ChannelManager.getContextFromIdentified(channel);
             RpcMessage rpcMsg = buildResponseMessage(rpcMessage, msg, msg instanceof HeartbeatMessage
                 ? ProtocolConstants.MSGTYPE_HEARTBEAT_RESPONSE
-                : ProtocolConstants.MSGTYPE_RESPONSE);
+                : ProtocolConstants.MSGTYPE_RESPONSE, rpcContext.getVersion());
             super.sendAsync(clientChannel, rpcMsg);
         } else {
             throw new RuntimeException("channel is error.");
@@ -163,10 +169,13 @@ class ServerHandler extends ChannelDuplexHandler {
          */
         @Override
         public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
-            if (!(msg instanceof RpcMessage)) {
-                return;
+            RpcMessage rpcMessage = null;
+            if (msg instanceof ProtocolRpcMessage) {
+                rpcMessage = ((ProtocolRpcMessage) msg).protocolMsg2RpcMsg();
+                processMessage(ctx, rpcMessage);
+            } else {
+                LOGGER.error("rpcMessage type error");
             }
-            processMessage(ctx, (RpcMessage) msg);
         }
 
         @Override

core/src/main/java/io/seata/core/rpc/netty/CompatibleProtocolDecoder.java

@@ -0,0 +1,156 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed 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 io.seata.core.rpc.netty;
+
+import com.google.common.collect.ImmutableMap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import io.seata.core.exception.DecodeException;
+import io.seata.core.protocol.ProtocolConstants;
+import io.seata.core.rpc.netty.v0.ProtocolDecoderV0;
+import io.seata.core.rpc.netty.v1.ProtocolDecoderV1;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+/**
+ * <pre>
+ * (> 0.7.0)
+ * 0     1     2     3     4     5     6     7     8     9    10     11    12    13    14    15    16
+ * +-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+ * |   magic   |Proto|     Full length       |    Head   | Msg |Seria|Compr|     RequestId         |
+ * |   code    |colVer|    (head+body)      |   Length  |Type |lizer|ess  |                       |
+ * +-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+
+ *
+ * (<= 0.7.0)
+ * 0     1     2     3     4           6           8          10           12          14
+ * +-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+ * |   0xdada  |   flag    | typecode/ |                 requestid                     |
+ * |           |           | bodylength|                                               |
+ * +-----------+-----------+-----------+-----------+-----------+-----------+-----------+
+ *
+ * </pre>
+ * <p>
+ * <li>Full Length: include all data </li>
+ * <li>Head Length: include head data from magic code to head map. </li>
+ * <li>Body Length: Full Length - Head Length</li>
+ * </p>
+ * https://github.com/seata/seata/issues/893
+ *
+ * @author Bughue
+ * @since 2.0.0
+ */
+public class CompatibleProtocolDecoder extends LengthFieldBasedFrameDecoder {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(CompatibleProtocolDecoder.class);
+    private static Map<Byte, ProtocolDecoder> protocolDecoderMap;
+
+    public CompatibleProtocolDecoder() {
+        // default is 8M
+        this(ProtocolConstants.MAX_FRAME_LENGTH);
+    }
+
+    public CompatibleProtocolDecoder(int maxFrameLength) {
+        /*
+        int maxFrameLength,      
+        int lengthFieldOffset,  magic code is 2B, and version is 1B, and then FullLength. so value is 3
+        int lengthFieldLength,  FullLength is int(4B). so values is 4
+        int lengthAdjustment,   FullLength include all data and read 7 bytes before, so the left length is (FullLength-7). so values is -7
+        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();
+    }
+
+    @Override
+    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
+        ByteBuf frame;
+        Object decoded;
+        byte version;
+        try {
+            if (isV0(in)) {
+                decoded = in;
+                version = ProtocolConstants.VERSION_0;
+            } else {
+                decoded = super.decode(ctx, in);
+                version = decideVersion(decoded);
+            }
+
+            if (decoded instanceof ByteBuf) {
+                frame = (ByteBuf) decoded;
+                try {
+                    ProtocolDecoder decoder = protocolDecoderMap.get(version);
+                    if (decoder == null) {
+                        throw new IllegalArgumentException("Unknown version: " + version);
+                    }
+                    return decoder.decodeFrame(frame);
+                } finally {
+                    if (!isV0(version)) {
+                        frame.release();
+                    }
+                }
+            }
+        } catch (Exception exx) {
+            LOGGER.error("Decode frame error, cause: {}", exx.getMessage());
+            throw new DecodeException(exx);
+        }
+        return decoded;
+    }
+
+    protected byte decideVersion(Object in) {
+        if (in instanceof ByteBuf) {
+            ByteBuf frame = (ByteBuf) in;
+            frame.markReaderIndex();
+            byte b0 = frame.readByte();
+            byte b1 = frame.readByte();
+            if (ProtocolConstants.MAGIC_CODE_BYTES[0] != b0
+                    || ProtocolConstants.MAGIC_CODE_BYTES[1] != b1) {
+                throw new IllegalArgumentException("Unknown magic code: " + b0 + ", " + b1);
+            }
+
+            byte version = frame.readByte();
+            frame.resetReaderIndex();
+            return version;
+        }
+        return -1;
+    }
+
+
+    protected boolean isV0(ByteBuf in) {
+        boolean isV0 = false;
+        in.markReaderIndex();
+        byte b0 = in.readByte();
+        byte b1 = in.readByte();
+        byte version = in.readByte();
+        if (ProtocolConstants.MAGIC_CODE_BYTES[0] == b0
+                && ProtocolConstants.MAGIC_CODE_BYTES[1] == b1
+                && isV0(version)) {
+            isV0 = true;
+        }
+
+        in.resetReaderIndex();
+        return isV0;
+    }
+
+    protected boolean isV0(byte version) {
+        return version == ProtocolConstants.VERSION_0;
+    }
+}