From 2327da77b147324c711859394d6cb2d40796d19f Mon Sep 17 00:00:00 2001
From: CalvinKirs <kirs@apache.org>
Date: Tue, 12 Jan 2021 19:21:38 +0800
Subject: [PATCH] rpc

---
 dolphinscheduler-remote/pom.xml               | 13 +++++
 .../remote/NettyRemotingClient.java           |  2 -
 .../remote/NettyRemotingServer.java           |  1 -
 .../remote/decoder/NettyDecoder.java          | 44 ++++++++++++++++
 .../remote/decoder/NettyEncoder.java          | 32 ++++++++++++
 .../remote/handler/NettyClientHandler.java    |  2 -
 .../remote/handler/NettyServerHandler.java    |  2 -
 .../remote/serialize/ProtoStuffUtils.java     | 52 +++++++++++++++++++
 8 files changed, 141 insertions(+), 7 deletions(-)
 create mode 100644 dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyDecoder.java
 create mode 100644 dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyEncoder.java
 create mode 100644 dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/serialize/ProtoStuffUtils.java

diff --git a/dolphinscheduler-remote/pom.xml b/dolphinscheduler-remote/pom.xml
index e40b7e1274..27f0923017 100644
--- a/dolphinscheduler-remote/pom.xml
+++ b/dolphinscheduler-remote/pom.xml
@@ -51,6 +51,19 @@
             <groupId>net.bytebuddy</groupId>
             <artifactId>byte-buddy</artifactId>
         </dependency>
+        <!--protostuff-->
+        <!-- https://mvnrepository.com/artifact/io.protostuff/protostuff-core -->
+        <dependency>
+            <groupId>io.protostuff</groupId>
+            <artifactId>protostuff-core</artifactId>
+            <version>1.7.2</version>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/io.protostuff/protostuff-runtime -->
+        <dependency>
+            <groupId>io.protostuff</groupId>
+            <artifactId>protostuff-runtime</artifactId>
+            <version>1.7.2</version>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingClient.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingClient.java
index c1aea90393..0092b682e8 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingClient.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingClient.java
@@ -19,8 +19,6 @@ package org.apache.dolphinscheduler.remote;
 
 import org.apache.dolphinscheduler.remote.codec.NettyDecoder;
 import org.apache.dolphinscheduler.remote.codec.NettyEncoder;
-import org.apache.dolphinscheduler.remote.command.Command;
-import org.apache.dolphinscheduler.remote.command.CommandType;
 import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
 import org.apache.dolphinscheduler.remote.exceptions.RemotingException;
 import org.apache.dolphinscheduler.remote.exceptions.RemotingTimeoutException;
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingServer.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingServer.java
index 867cf4dc56..56b31cd0b5 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingServer.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingServer.java
@@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.remote;
 
 import org.apache.dolphinscheduler.remote.codec.NettyDecoder;
 import org.apache.dolphinscheduler.remote.codec.NettyEncoder;
-import org.apache.dolphinscheduler.remote.command.CommandType;
 import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
 import org.apache.dolphinscheduler.remote.handler.NettyServerHandler;
 import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyDecoder.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyDecoder.java
new file mode 100644
index 0000000000..f212b77389
--- /dev/null
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyDecoder.java
@@ -0,0 +1,44 @@
+package org.apache.dolphinscheduler.remote.decoder;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+
+import org.apache.dolphinscheduler.remote.serialize.ProtoStuffUtils;
+
+import java.util.List;
+
+/**
+ * @author jiangli
+ * @date 2021-01-12 18:53
+ */
+public class NettyDecoder extends ByteToMessageDecoder {
+
+    private Class<?> genericClass;
+
+    // 构造函数传入向反序列化的class
+    public NettyDecoder(Class<?> genericClass) {
+        this.genericClass = genericClass;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
+        if (byteBuf.readableBytes() < 4) {
+            return;
+        }
+        byteBuf.markReaderIndex();
+        int dataLength = byteBuf.readInt();
+        if (dataLength < 0) {
+            channelHandlerContext.close();
+        }
+        if (byteBuf.readableBytes() < dataLength) {
+            byteBuf.resetReaderIndex();
+        }
+        //将ByteBuf转换为byte[]
+        byte[] data = new byte[dataLength];
+        byteBuf.readBytes(data);
+        //将data转换成object
+        Object obj = ProtoStuffUtils.deserialize(data, genericClass);
+        list.add(obj);
+    }
+}
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyEncoder.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyEncoder.java
new file mode 100644
index 0000000000..3ba5781fb4
--- /dev/null
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyEncoder.java
@@ -0,0 +1,32 @@
+package org.apache.dolphinscheduler.remote.decoder;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+
+import org.apache.dolphinscheduler.remote.serialize.ProtoStuffUtils;
+
+/**
+ * @author jiangli
+ * @date 2021-01-12 18:52
+ */
+public class NettyEncoder extends MessageToByteEncoder {
+
+
+    private Class<?> genericClass;
+
+    // 构造函数传入向反序列化的class
+    public NettyEncoder(Class<?> genericClass) {
+        this.genericClass = genericClass;
+    }
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception {
+
+        if (genericClass.isInstance(o)) {
+            byte[] data = ProtoStuffUtils.serialize(o);
+            byteBuf.writeInt(data.length);
+            byteBuf.writeBytes(data);
+        }
+
+    }
+}
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyClientHandler.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyClientHandler.java
index a988acfe17..701c3b9a09 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyClientHandler.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyClientHandler.java
@@ -18,8 +18,6 @@
 package org.apache.dolphinscheduler.remote.handler;
 
 import org.apache.dolphinscheduler.remote.NettyRemotingClient;
-import org.apache.dolphinscheduler.remote.command.Command;
-import org.apache.dolphinscheduler.remote.command.CommandType;
 import org.apache.dolphinscheduler.remote.future.ResponseFuture;
 import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
 import org.apache.dolphinscheduler.remote.utils.ChannelUtils;
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyServerHandler.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyServerHandler.java
index 09e41e9b54..0e34f99996 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyServerHandler.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyServerHandler.java
@@ -18,8 +18,6 @@
 package org.apache.dolphinscheduler.remote.handler;
 
 import org.apache.dolphinscheduler.remote.NettyRemotingServer;
-import org.apache.dolphinscheduler.remote.command.Command;
-import org.apache.dolphinscheduler.remote.command.CommandType;
 import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
 import org.apache.dolphinscheduler.remote.utils.ChannelUtils;
 import org.apache.dolphinscheduler.remote.utils.Pair;
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/serialize/ProtoStuffUtils.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/serialize/ProtoStuffUtils.java
new file mode 100644
index 0000000000..4face03f90
--- /dev/null
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/serialize/ProtoStuffUtils.java
@@ -0,0 +1,52 @@
+package org.apache.dolphinscheduler.remote.serialize;
+
+import io.protostuff.LinkedBuffer;
+import io.protostuff.ProtostuffIOUtil;
+import io.protostuff.Schema;
+import io.protostuff.runtime.RuntimeSchema;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author jiangli
+ * @date 2021-01-12 18:56
+ */
+public class ProtoStuffUtils {
+
+    private static LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
+
+    private static Map<Class<?>, Schema<?>> schemaCache = new ConcurrentHashMap<>();
+
+    @SuppressWarnings("unchecked")
+    public static <T> byte[] serialize(T obj) {
+        Class<T> clazz = (Class<T>) obj.getClass();
+        Schema<T> schema = getSchema(clazz);
+        byte[] data;
+        try {
+            data = ProtostuffIOUtil.toByteArray(obj, schema, buffer);
+        } finally {
+            buffer.clear();
+        }
+        return data;
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> Schema<T> getSchema(Class<T> clazz) {
+        Schema<T> schema = (Schema<T>) schemaCache.get(clazz);
+        if (schema == null) {
+            schema = RuntimeSchema.getSchema(clazz);
+            if (schema == null) {
+                schemaCache.put(clazz, schema);
+            }
+        }
+        return schema;
+    }
+
+    public static <T> T deserialize(byte[] bytes, Class<T> clazz) {
+        Schema<T> schema = getSchema(clazz);
+        T obj = schema.newMessage();
+        ProtostuffIOUtil.mergeFrom(bytes, obj, schema);
+        return obj;
+    }
+}