feat: add generic invoke

dubbo-serialization-extensions/dubbo-serialization-protobuf/pom.xml

@@ -48,6 +48,10 @@ limitations under the License.
             <artifactId>protobuf-java-util</artifactId>
             <version>3.11.0</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

dubbo-serialization-extensions/dubbo-serialization-protobuf/src/main/java/org/apache/dubbo/common/serialize/protobuf/support/ProtobufObjectInput.java

@@ -0,0 +1,134 @@
+/*
+ *
+ *   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.dubbo.common.serialize.protobuf.support;
+
+import org.apache.dubbo.common.serialize.ObjectInput;
+import org.apache.dubbo.common.serialize.hessian2.Hessian2ObjectInput;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Type;
+
+public class ProtobufObjectInput implements ObjectInput {
+    private final Hessian2ObjectInput delegate;
+
+    ProtobufObjectInput(InputStream is) {
+        this.delegate = new Hessian2ObjectInput(is);
+    }
+
+    @Override
+    public Object readObject() {
+        String className = null;
+        try {
+            className = readUTF();
+            if (ProtobufUtils.canSerializeWithProtobuf(ProtobufUtils.loadClass(className))) {
+                byte[] data = readBytes();
+                return deserializeWithProtobuf(data, ProtobufUtils.loadClass(className));
+            } else {
+                return delegate.readObject();
+            }
+        } catch (Throwable ex) {
+            String message =
+                    className == null ? "Unable to deserializeWithProtobuf object" : "Unable to deserializeWithProtobuf object for " + className;
+            throw new ProtobufSerializationException(message, ex);
+        }
+    }
+
+    private <T> T deserializeWithProtobuf(byte[] data, Class<T> clazz) throws IOException, NoSuchMethodException,
+            IllegalAccessException, InvocationTargetException {
+        try (ByteArrayInputStream newIs = new ByteArrayInputStream(data)) {
+            if (ProtobufUtils.canSerializeWithProtobuf(clazz)) {
+                return ProtobufUtils.deserializeWithProtobuf(newIs, clazz);
+            } else {
+                return (T) delegate.readObject();
+            }
+        }
+    }
+
+    @Override
+    public <T> T readObject(Class<T> cls) {
+        if (cls == null || cls == Object.class) {
+            return (T) readObject();
+        }
+
+        try {
+            String className = readUTF();
+            if (ProtobufUtils.canSerializeWithProtobuf(ProtobufUtils.loadClass(className))) {
+                byte[] data = readBytes();
+                return deserializeWithProtobuf(data, cls);
+            }
+            return delegate.readObject(cls);
+        } catch (Throwable ex) {
+            throw new ProtobufSerializationException("Unable to deserializeWithProtobuf object for " + cls.getName(), ex);
+        }
+    }
+
+    @Override
+    public <T> T readObject(Class<T> cls, Type type) {
+        return readObject(cls);
+    }
+
+    @Override
+    public boolean readBool() throws IOException {
+        return delegate.readBool();
+    }
+
+    @Override
+    public byte readByte() throws IOException {
+        return delegate.readByte();
+    }
+
+    @Override
+    public short readShort() throws IOException {
+        return delegate.readShort();
+    }
+
+    @Override
+    public int readInt() throws IOException {
+        return delegate.readInt();
+    }
+
+    @Override
+    public long readLong() throws IOException {
+        return delegate.readLong();
+    }
+
+    @Override
+    public float readFloat() throws IOException {
+        return delegate.readFloat();
+    }
+
+    @Override
+    public double readDouble() throws IOException {
+        return delegate.readDouble();
+    }
+
+    @Override
+    public byte[] readBytes() throws IOException {
+        return delegate.readBytes();
+    }
+
+    @Override
+    public String readUTF() throws IOException {
+        return delegate.readUTF();
+    }
+}

dubbo-serialization-extensions/dubbo-serialization-protobuf/src/main/java/org/apache/dubbo/common/serialize/protobuf/support/ProtobufObjectOutput.java

@@ -0,0 +1,119 @@
+/*
+ *
+ *   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.dubbo.common.serialize.protobuf.support;
+
+import org.apache.dubbo.common.serialize.ObjectOutput;
+import org.apache.dubbo.common.serialize.hessian2.Hessian2ObjectOutput;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+
+public class ProtobufObjectOutput implements ObjectOutput {
+    private final Hessian2ObjectOutput delegate;
+
+    public ProtobufObjectOutput(OutputStream outputStream) {
+        this.delegate = new Hessian2ObjectOutput(outputStream);
+    }
+
+    /**
+     * when object insatanceOf protobuf object serialize with ProtobufBuilder
+     *
+     * @param obj object
+     */
+    @Override
+    public void writeObject(Object obj) {
+        try {
+            Class clazz = obj.getClass();
+            writeUTF(clazz.getName());
+            if (ProtobufUtils.canSerializeWithProtobuf(clazz)) {
+                writeBytes(serializeToBytes(obj));
+            } else {
+                delegate.writeObject(obj);
+            }
+        } catch (Throwable ex) {
+            String message =
+                    obj == null ? "Unable to serialize object" : "Unable to serialize object for " + obj.getClass().getName();
+            throw new ProtobufSerializationException(message, ex);
+        }
+    }
+
+    private byte[] serializeToBytes(Object obj) throws IOException {
+        try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
+            ProtobufUtils.serializeWithProtobuf(obj, os);
+            return os.toByteArray();
+        }
+    }
+
+    @Override
+    public void writeBool(boolean v) throws IOException {
+        delegate.writeBool(v);
+    }
+
+    @Override
+    public void writeByte(byte v) throws IOException {
+        delegate.writeByte(v);
+    }
+
+    @Override
+    public void writeShort(short v) throws IOException {
+        delegate.writeShort(v);
+    }
+
+    @Override
+    public void writeInt(int v) throws IOException {
+        delegate.writeInt(v);
+    }
+
+    @Override
+    public void writeLong(long v) throws IOException {
+        delegate.writeLong(v);
+    }
+
+    @Override
+    public void writeFloat(float v) throws IOException {
+        delegate.writeFloat(v);
+    }
+
+    @Override
+    public void writeDouble(double v) throws IOException {
+        delegate.writeDouble(v);
+    }
+
+    @Override
+    public void writeBytes(byte[] b) throws IOException {
+        delegate.writeBytes(b);
+    }
+
+    @Override
+    public void writeBytes(byte[] b, int off, int len) throws IOException {
+        delegate.writeBytes(b, off, len);
+    }
+
+    @Override
+    public void writeUTF(String v) throws IOException {
+        delegate.writeUTF(v);
+    }
+
+    @Override
+    public void flushBuffer() throws IOException {
+        delegate.flushBuffer();
+    }
+}

dubbo-serialization-extensions/dubbo-serialization-protobuf/src/main/java/org/apache/dubbo/common/serialize/protobuf/support/ProtobufSerialization.java

@@ -0,0 +1,60 @@
+/*
+ *
+ *   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.dubbo.common.serialize.protobuf.support;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.serialize.ObjectInput;
+import org.apache.dubbo.common.serialize.ObjectOutput;
+import org.apache.dubbo.common.serialize.Serialization;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * @author lin
+ * Date: 2019-10-01
+ */
+public class ProtobufSerialization implements Serialization {
+    //Serialization ID must less than SERIALIZATION_MASK and cannot be duplicated with other serializations
+    private static final int CONTENT_TYPE_ID = 30;
+
+    /**
+     * The content type id should not be changed in any circumstance!
+     */
+    @Override
+    public byte getContentTypeId() {
+        return CONTENT_TYPE_ID;
+    }
+
+    @Override
+    public String getContentType() {
+        return "x-application/protobuf";
+    }
+
+    @Override
+    public ObjectOutput serialize(URL url, OutputStream output) {
+        return new ProtobufObjectOutput(output);
+    }
+
+    @Override
+    public ObjectInput deserialize(URL url, InputStream input) {
+        return new ProtobufObjectInput(input);
+    }
+}

dubbo-serialization-extensions/dubbo-serialization-protobuf/src/main/java/org/apache/dubbo/common/serialize/protobuf/support/ProtobufSerializationException.java

@@ -0,0 +1,7 @@
+package org.apache.dubbo.common.serialize.protobuf.support;
+
+public class ProtobufSerializationException extends RuntimeException {
+    public ProtobufSerializationException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}