[Dubbo-1983] Support Protobuf Serialization (#2618)

* finish support protobuf

* polish

* fix code review

* use the general test for serialization

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

@@ -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.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <artifactId>dubbo-serialization</artifactId>
+        <groupId>org.apache.dubbo</groupId>
+        <version>2.7.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>dubbo-serialization-protobuf</artifactId>
+    <packaging>jar</packaging>
+    <name>${project.artifactId}</name>
+    <description>The protobuf serialization module of dubbo project</description>
+
+    <properties>
+        <protobuf.version>1.5.9</protobuf.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-serialization-api</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.protostuff</groupId>
+            <artifactId>protostuff-core</artifactId>
+            <version>${protobuf.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.protostuff</groupId>
+            <artifactId>protostuff-runtime</artifactId>
+            <version>${protobuf.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.objenesis</groupId>
+            <artifactId>objenesis</artifactId>
+            <version>2.6</version>
+        </dependency>
+    </dependencies>
+
+</project>

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

@@ -0,0 +1,158 @@
+/*
+ * 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;
+
+import io.protostuff.ProtobufIOUtil;
+import io.protostuff.Schema;
+import io.protostuff.runtime.RuntimeSchema;
+import org.apache.dubbo.common.serialize.ObjectInput;
+import org.apache.dubbo.common.serialize.protobuf.utils.WrapperUtils;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Type;
+
+public class ProtobufObjectInput implements ObjectInput {
+
+    private DataInputStream dis;
+
+    public ProtobufObjectInput(InputStream inputStream) {
+        dis = new DataInputStream(inputStream);
+    }
+
+    @SuppressWarnings("ResultOfMethodCallIgnored")
+    @Override
+    public Object readObject() throws IOException, ClassNotFoundException {
+        int classNameLength = dis.readInt();
+        int bytesLength = dis.readInt();
+
+        if (classNameLength < 0 || bytesLength < 0) {
+            throw new IOException();
+        }
+
+        byte[] classNameBytes = new byte[classNameLength];
+        dis.readFully(classNameBytes, 0, classNameLength);
+
+        byte[] bytes = new byte[bytesLength];
+        dis.readFully(bytes, 0, bytesLength);
+
+        String className = new String(classNameBytes);
+        Class clazz = Class.forName(className);
+
+        Object result;
+        if (WrapperUtils.needWrapper(clazz)) {
+            Schema<Wrapper> schema = RuntimeSchema.getSchema(Wrapper.class);
+            Wrapper wrapper = schema.newMessage();
+            ProtobufIOUtil.mergeFrom(bytes, wrapper, schema);
+            result = wrapper.getData();
+        } else {
+            Schema schema = RuntimeSchema.getSchema(clazz);
+            result = schema.newMessage();
+            ProtobufIOUtil.mergeFrom(bytes, result, schema);
+        }
+
+        return result;
+    }
+
+    @Override
+    public <T> T readObject(Class<T> clazz) throws IOException, ClassNotFoundException {
+        int classNameLength = dis.readInt();
+        int bytesLength = dis.readInt();
+
+        if (classNameLength < 0 || bytesLength < 0) {
+            throw new IOException();
+        }
+
+        byte[] classNameBytes = new byte[classNameLength];
+        dis.read(classNameBytes, 0, classNameLength);
+
+        byte[] bytes = new byte[bytesLength];
+        dis.read(bytes, 0, bytesLength);
+
+        T result;
+        if (WrapperUtils.needWrapper(clazz)) {
+            Schema<Wrapper> schema = RuntimeSchema.getSchema(Wrapper.class);
+            Wrapper wrapper = schema.newMessage();
+            ProtobufIOUtil.mergeFrom(bytes, wrapper, schema);
+            result = (T) wrapper.getData();
+        } else {
+            Schema<T> schema = RuntimeSchema.getSchema(clazz);
+            result = schema.newMessage();
+            ProtobufIOUtil.mergeFrom(bytes, result, schema);
+        }
+
+        return result;
+    }
+
+    @Override
+    public <T> T readObject(Class<T> cls, Type type) throws IOException, ClassNotFoundException {
+        return readObject(cls);
+    }
+
+    @Override
+    public boolean readBool() throws IOException {
+        return dis.readBoolean();
+    }
+
+    @Override
+    public byte readByte() throws IOException {
+        return dis.readByte();
+    }
+
+    @Override
+    public short readShort() throws IOException {
+        return dis.readShort();
+    }
+
+    @Override
+    public int readInt() throws IOException {
+        return dis.readInt();
+    }
+
+    @Override
+    public long readLong() throws IOException {
+        return dis.readLong();
+    }
+
+    @Override
+    public float readFloat() throws IOException {
+        return dis.readFloat();
+    }
+
+    @Override
+    public double readDouble() throws IOException {
+        return dis.readDouble();
+    }
+
+    @Override
+    public String readUTF() throws IOException {
+        int length = dis.readInt();
+        byte[] bytes = new byte[length];
+        dis.read(bytes, 0, length);
+        return new String(bytes);
+    }
+
+    @Override
+    public byte[] readBytes() throws IOException {
+        int length = dis.readInt();
+        byte[] bytes = new byte[length];
+        dis.read(bytes, 0, length);
+        return bytes;
+    }
+}

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

@@ -0,0 +1,121 @@
+/*
+ * 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;
+
+import io.protostuff.*;
+import io.protostuff.runtime.RuntimeSchema;
+import org.apache.dubbo.common.serialize.ObjectOutput;
+import org.apache.dubbo.common.serialize.protobuf.utils.WrapperUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+
+public class ProtobufObjectOutput implements ObjectOutput {
+
+    private DataOutputStream dos;
+
+    public ProtobufObjectOutput(OutputStream outputStream) {
+        dos = new DataOutputStream(outputStream);
+    }
+
+    @Override
+    public void writeObject(Object obj) throws IOException {
+        LinkedBuffer buffer = LinkedBuffer.allocate();
+
+        byte[] bytes;
+        byte[] classNameBytes;
+
+        if (WrapperUtils.needWrapper(obj)) {
+            Schema<Wrapper> schema = RuntimeSchema.getSchema(Wrapper.class);
+            Wrapper wrapper = new Wrapper(obj);
+            bytes = ProtobufIOUtil.toByteArray(wrapper, schema, buffer);
+            classNameBytes = Wrapper.class.getName().getBytes();
+        } else {
+            Schema schema = RuntimeSchema.getSchema(obj.getClass());
+            bytes = ProtobufIOUtil.toByteArray(obj, schema, buffer);
+            classNameBytes = obj.getClass().getName().getBytes();
+        }
+
+        dos.writeInt(classNameBytes.length);
+        dos.writeInt(bytes.length);
+        dos.write(classNameBytes);
+        dos.write(bytes);
+    }
+
+    @Override
+    public void writeBool(boolean v) throws IOException {
+        dos.writeBoolean(v);
+    }
+
+    @Override
+    public void writeByte(byte v) throws IOException {
+        dos.writeByte(v);
+    }
+
+    @Override
+    public void writeShort(short v) throws IOException {
+        dos.writeShort(v);
+    }
+
+    @Override
+    public void writeInt(int v) throws IOException {
+        dos.writeInt(v);
+    }
+
+    @Override
+    public void writeLong(long v) throws IOException {
+        dos.writeLong(v);
+    }
+
+    @Override
+    public void writeFloat(float v) throws IOException {
+        dos.writeFloat(v);
+    }
+
+    @Override
+    public void writeDouble(double v) throws IOException {
+        dos.writeDouble(v);
+    }
+
+    @Override
+    public void writeUTF(String v) throws IOException {
+        byte[] bytes = v.getBytes();
+        dos.writeInt(bytes.length);
+        dos.write(bytes);
+    }
+
+    @Override
+    public void writeBytes(byte[] v) throws IOException {
+        dos.writeInt(v.length);
+        dos.write(v);
+    }
+
+    @Override
+    public void writeBytes(byte[] v, int off, int len) throws IOException {
+        dos.writeInt(len);
+        byte[] bytes = new byte[len];
+        System.arraycopy(v, off, bytes, 0, len);
+        dos.write(bytes);
+    }
+
+    @Override
+    public void flushBuffer() throws IOException {
+        dos.flush();
+    }
+}

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

@@ -0,0 +1,49 @@
+/*
+ * 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;
+
+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.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+public class ProtobufSerialization implements Serialization {
+    @Override
+    public byte getContentTypeId() {
+        return 10;
+    }
+
+    @Override
+    public String getContentType() {
+        return "x-application/protobuf";
+    }
+
+    @Override
+    public ObjectOutput serialize(URL url, OutputStream output) throws IOException {
+        return new ProtobufObjectOutput(output);
+    }
+
+    @Override
+    public ObjectInput deserialize(URL url, InputStream input) throws IOException {
+        return new ProtobufObjectInput(input);
+    }
+}