serialization

rpt-base/src/main/java/cn/promptness/rpt/base/coder/MessageDecoder.java

@@ -3,8 +3,8 @@
 import cn.promptness.rpt.base.protocol.Message;
 import cn.promptness.rpt.base.protocol.MessageType;
 import cn.promptness.rpt.base.protocol.Meta;
-import cn.promptness.rpt.base.serialize.Jackson;
-import cn.promptness.rpt.base.serialize.Serializer;
+import cn.promptness.rpt.base.serialize.SerializerDispatcher;
+import cn.promptness.rpt.base.serialize.api.SerializationType;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.channel.ChannelHandlerContext;
@@ -15,7 +15,7 @@
 
 public class MessageDecoder extends MessageToMessageDecoder<ByteBuf> {
 
-    private static final Serializer SERIALIZER = new Jackson();
+    private static final SerializerDispatcher SERIALIZER_DISPATCHER = new SerializerDispatcher();
 
     @Override
     protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
@@ -27,7 +27,7 @@ protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteB
         if (protobufLength > 0) {
             byte[] metaByte = new byte[protobufLength];
             byteBuf.readBytes(metaByte);
-            Meta meta = SERIALIZER.deserialize(metaByte);
+            Meta meta = SERIALIZER_DISPATCHER.deserialize(SerializationType.PROTOSTUFF, metaByte);
             proxyMessage.setMeta(meta);
         }
         if (byteBuf.isReadable()) {

rpt-base/src/main/java/cn/promptness/rpt/base/coder/MessageEncoder.java

@@ -3,8 +3,8 @@
 import cn.promptness.rpt.base.protocol.Message;
 import cn.promptness.rpt.base.protocol.MessageType;
 import cn.promptness.rpt.base.protocol.Meta;
-import cn.promptness.rpt.base.serialize.Jackson;
-import cn.promptness.rpt.base.serialize.Serializer;
+import cn.promptness.rpt.base.serialize.SerializerDispatcher;
+import cn.promptness.rpt.base.serialize.api.SerializationType;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToByteEncoder;
@@ -13,15 +13,15 @@
 
 public class MessageEncoder extends MessageToByteEncoder<Message> {
 
-    private static final Serializer SERIALIZER = new Jackson();
+    private static final SerializerDispatcher SERIALIZER_DISPATCHER = new SerializerDispatcher();
 
     @Override
     protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf out) throws Exception {
         MessageType type = message.getType();
         out.writeInt(type.getCode());
 
         Meta meta = message.getMeta();
-        byte[] protobuf = meta == null ? EmptyArrays.EMPTY_BYTES : SERIALIZER.serialize(meta);
+        byte[] protobuf = meta == null ? EmptyArrays.EMPTY_BYTES : SERIALIZER_DISPATCHER.serialize(SerializationType.PROTOSTUFF, meta);
         out.writeInt(protobuf.length);
         out.writeBytes(protobuf);
 

rpt-base/src/main/java/cn/promptness/rpt/base/executor/MessageDispatcher.java

@@ -4,21 +4,18 @@
 import cn.promptness.rpt.base.protocol.MessageType;
 import io.netty.channel.ChannelHandlerContext;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.ServiceLoader;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class MessageDispatcher {
 
-    private static final Map<MessageType, MessageExecutor> MESSAGE_EXECUTOR_MAP = new HashMap<>();
+    private static final Map<MessageType, MessageExecutor> MESSAGE_EXECUTOR_MAP = new ConcurrentHashMap<>();
 
     static {
         ServiceLoader<MessageExecutor> executors = ServiceLoader.load(MessageExecutor.class);
         for (MessageExecutor executor : executors) {
-            MessageExecutor messageExecutor = MESSAGE_EXECUTOR_MAP.put(executor.getMessageType(), executor);
-            if (messageExecutor != null) {
-                throw new RuntimeException(String.format("%s Message Executor Register Repeat", messageExecutor.getMessageType().name()));
-            }
+            MESSAGE_EXECUTOR_MAP.put(executor.getMessageType(), executor);
         }
     }
 

rpt-base/src/main/java/cn/promptness/rpt/base/serialize/SerializerDispatcher.java

@@ -0,0 +1,26 @@
+package cn.promptness.rpt.base.serialize;
+
+import cn.promptness.rpt.base.serialize.api.*;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+
+public class SerializerDispatcher {
+
+    public byte[] serialize(SerializationType serializationType, Object obj) throws Exception {
+        Serialization serialization = SerializeFactory.getSerialization(serializationType);
+        try (ByteArrayOutputStream byteBufOutputStream = new ByteArrayOutputStream(); ObjectOutputStream objectOutputStream = serialization.serialize(byteBufOutputStream)) {
+            objectOutputStream.writeObject(obj);
+            objectOutputStream.flush();
+            return byteBufOutputStream.toByteArray();
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> T deserialize(SerializationType serializationType, byte[] data) throws Exception {
+        Serialization serialization = SerializeFactory.getSerialization(serializationType);
+        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(data); ObjectInputStream objectInputStream = serialization.deserialize(byteArrayInputStream)) {
+            return (T) objectInputStream.readObject(Object.class);
+        }
+    }
+}

rpt-base/src/main/java/cn/promptness/rpt/base/serialize/api/ObjectInputStream.java

@@ -0,0 +1,31 @@
+package cn.promptness.rpt.base.serialize.api;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.util.Map;
+
+public interface ObjectInputStream extends Closeable {
+
+    int readInt() throws IOException;
+
+    byte readByte() throws IOException;
+
+    byte[] readBytes() throws IOException;
+
+    String readUTF() throws IOException;
+
+    <T> T readObject(Class<T> cls) throws IOException, ClassNotFoundException;
+
+    default <T> T readObject(Class<T> cls, Type genericType) throws IOException, ClassNotFoundException {
+        return readObject(cls);
+    }
+
+    default Throwable readThrowable() throws IOException, ClassNotFoundException {
+        return readObject(Throwable.class);
+    }
+
+    default Map readMap() throws IOException, ClassNotFoundException {
+        return readObject(Map.class);
+    }
+}

rpt-base/src/main/java/cn/promptness/rpt/base/serialize/api/ObjectOutputStream.java

@@ -0,0 +1,32 @@
+package cn.promptness.rpt.base.serialize.api;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Map;
+
+public interface ObjectOutputStream extends Closeable {
+
+    void writeInt(int v) throws IOException;
+
+    void writeByte(byte v) throws IOException;
+
+    void writeBytes(byte[] b) throws IOException;
+
+    void writeUTF(String v) throws IOException;
+
+    void writeObject(Object obj) throws IOException;
+
+    default void writeMap(Map<String, String> map) throws IOException {
+        writeObject(map);
+    }
+
+    default void writeThrowable(Object obj) throws IOException {
+        writeObject(obj);
+    }
+
+    default void writeEvent(Object data) throws IOException {
+        writeObject(data);
+    }
+
+    void flush() throws IOException;
+}

rpt-base/src/main/java/cn/promptness/rpt/base/serialize/api/Serialization.java

@@ -0,0 +1,15 @@
+package cn.promptness.rpt.base.serialize.api;
+
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+public interface Serialization {
+
+    SerializationType getType();
+
+    ObjectOutputStream serialize(OutputStream output) throws IOException;
+
+    ObjectInputStream deserialize(InputStream input) throws IOException;
+}

rpt-base/src/main/java/cn/promptness/rpt/base/serialize/api/SerializationType.java

@@ -0,0 +1,27 @@
+package cn.promptness.rpt.base.serialize.api;
+
+public enum SerializationType {
+
+    /**
+     *
+     */
+    PROTOSTUFF(1, "protostuff"),
+    JSON(2, "jackson");
+
+    SerializationType(int code, String desc) {
+        this.code = code;
+        this.desc = desc;
+    }
+
+    final int code;
+
+    final String desc;
+
+    public int getCode() {
+        return code;
+    }
+
+    public String getDesc() {
+        return desc;
+    }
+}

rpt-base/src/main/java/cn/promptness/rpt/base/serialize/api/SerializeFactory.java

@@ -0,0 +1,23 @@
+package cn.promptness.rpt.base.serialize.api;
+
+
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class SerializeFactory {
+
+    private static final Map<SerializationType, Serialization> SERIALIZATION_MAP = new ConcurrentHashMap<>();
+
+    static {
+        ServiceLoader<Serialization> serializations = ServiceLoader.load(Serialization.class);
+        for (Serialization serialization : serializations) {
+            SERIALIZATION_MAP.put(serialization.getType(), serialization);
+        }
+    }
+
+    public static Serialization getSerialization(SerializationType serializationType) {
+        return SERIALIZATION_MAP.get(serializationType);
+    }
+
+}

rpt-base/src/main/java/cn/promptness/rpt/base/serialize/json/JacksonUtil.java

@@ -0,0 +1,52 @@
+package cn.promptness.rpt.base.serialize.json;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.cfg.MapperConfig;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.fasterxml.jackson.databind.jsontype.PolymorphicTypeValidator;
+
+import static com.fasterxml.jackson.databind.ObjectMapper.DefaultTyping.NON_FINAL;
+
+public class JacksonUtil {
+
+    /**
+     * used for take the bean actual type with json string; for de-serializing the json string into real type bean.
+     */
+    private static final String TYPE_AS_JSON_PROPERTY = "typeAsJsonProperty";
+
+    private static final JsonMapper MAPPER = new JsonMapper();
+
+    static {
+        final boolean takeTypeAsProperty = Boolean.getBoolean(TYPE_AS_JSON_PROPERTY);
+        if (takeTypeAsProperty) {
+            MAPPER.activateDefaultTypingAsProperty(new PolymorphicTypeValidator() {
+                @Override
+                public Validity validateBaseType(final MapperConfig<?> config,
+                                                 final JavaType baseType) {
+                    return Validity.ALLOWED;
+                }
+
+                @Override
+                public Validity validateSubClassName(final MapperConfig<?> config,
+                                                     final JavaType baseType,
+                                                     final String subClassName) throws JsonMappingException {
+                    return Validity.ALLOWED;
+                }
+
+                @Override
+                public Validity validateSubType(final MapperConfig<?> config,
+                                                final JavaType baseType,
+                                                final JavaType subType) throws JsonMappingException {
+                    return Validity.ALLOWED;
+                }
+            }, NON_FINAL, null);
+        }
+        MAPPER.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+    }
+
+    public static JsonMapper getJsonMapper() {
+        return MAPPER;
+    }
+}