[Dubbo-3829] support google pb generic invocation

dubbo-all/pom.xml

@@ -387,6 +387,13 @@
             <scope>compile</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-serialization-googlePb</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-configcenter-api</artifactId>
@@ -585,6 +592,7 @@
                                     <include>org.apache.dubbo:dubbo-serialization-jdk</include>
                                     <include>org.apache.dubbo:dubbo-serialization-protostuff</include>
                                     <include>org.apache.dubbo:dubbo-serialization-gson</include>
+                                    <include>org.apache.dubbo:dubbo-serialization-googlePb</include>
                                     <include>org.apache.dubbo:dubbo-configcenter-api</include>
                                     <include>org.apache.dubbo:dubbo-configcenter-definition</include>
                                     <include>org.apache.dubbo:dubbo-configcenter-apollo</include>

dubbo-bom/pom.xml

@@ -342,6 +342,11 @@
                 <artifactId>dubbo-serialization-gson</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.dubbo</groupId>
+                <artifactId>dubbo-serialization-googlePb</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.dubbo</groupId>
                 <artifactId>dubbo-compatible</artifactId>

dubbo-common/src/main/java/org/apache/dubbo/common/Constants.java

@@ -727,6 +727,8 @@ public class Constants {
 
     public static final String GENERIC_SERIALIZATION_BEAN = "bean";
 
+    public static final String GENERIC_SERIALIZATION_PROTOBUF = "gernericprotobuf";
+
     public static final String DUBBO_IP_TO_REGISTRY = "DUBBO_IP_TO_REGISTRY";
 
     public static final String DUBBO_PORT_TO_REGISTRY = "DUBBO_PORT_TO_REGISTRY";

dubbo-dependencies-bom/pom.xml

@@ -108,6 +108,8 @@
         <cxf_version>3.1.15</cxf_version>
         <thrift_version>0.12.0</thrift_version>
         <hessian_version>4.0.38</hessian_version>
+        <protobuf-java>3.6.0</protobuf-java>
+        <protobuf.java.util>3.6.0</protobuf.java.util>
         <servlet_version>3.1.0</servlet_version>
         <jetty_version>9.4.11.v20180605</jetty_version>
         <validation_version>1.1.0.Final</validation_version>
@@ -271,6 +273,16 @@
                 <artifactId>hessian-lite</artifactId>
                 <version>${hessian_lite_version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.google.protobuf</groupId>
+                <artifactId>protobuf-java</artifactId>
+                <version>${protobuf-java}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.google.protobuf</groupId>
+                <artifactId>protobuf-java-util</artifactId>
+                <version>${protobuf.java.util}</version>
+            </dependency>
             <dependency>
                 <groupId>javax.servlet</groupId>
                 <artifactId>javax.servlet-api</artifactId>

dubbo-distribution/pom.xml

@@ -335,6 +335,11 @@
             <artifactId>dubbo-serialization-gson</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-serialization-googlePb</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo</artifactId>

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java

@@ -76,7 +76,7 @@ public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
                 } else if (ProtocolUtils.isJavaGenericSerialization(generic)) {
                     for (int i = 0; i < args.length; i++) {
                         if (byte[].class == args[i].getClass()) {
-                            try(UnsafeByteArrayInputStream is = new UnsafeByteArrayInputStream((byte[]) args[i])) {
+                            try (UnsafeByteArrayInputStream is = new UnsafeByteArrayInputStream((byte[]) args[i])) {
                                 args[i] = ExtensionLoader.getExtensionLoader(Serialization.class)
                                         .getExtension(Constants.GENERIC_SERIALIZATION_NATIVE_JAVA)
                                         .deserialize(null, is).readObject();
@@ -107,6 +107,22 @@ public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
                                             args[i].getClass().getName());
                         }
                     }
+                } else if (ProtocolUtils.isProtobufGenericSerialization(generic)) {
+                    // as proto3 only accept one protobuf parameter
+                    if (args.length == 1 && args[0] instanceof String) {
+                        try (UnsafeByteArrayInputStream is = new UnsafeByteArrayInputStream(((String) args[0]).getBytes())) {
+                            args[0] = ExtensionLoader.getExtensionLoader(Serialization.class)
+                                    .getExtension("" + Constants.GENERIC_SERIALIZATION_PROTOBUF)
+                                    .deserialize(null, is).readObject(method.getParameterTypes()[0]);
+                        } catch (Exception e) {
+                            throw new RpcException("Deserialize argument failed.", e);
+                        }
+                    } else {
+                        throw new RpcException(
+                                new StringBuilder("Generic serialization [").append(Constants.GENERIC_SERIALIZATION_PROTOBUF)
+                                        .append("] only support one").append(String.class.getName()).append(" argument ")
+                                        .append(" and your message size is ").append(args.length).append(" and type is").append(args[0].getClass().getName()).toString());
+                    }
                 }
                 Result result = invoker.invoke(new RpcInvocation(method, args, inv.getAttachments()));
                 if (result.hasException()
@@ -121,10 +137,22 @@ public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
                                 .serialize(null, os).writeObject(result.getValue());
                         return new RpcResult(os.toByteArray());
                     } catch (IOException e) {
-                        throw new RpcException("Serialize result failed.", e);
+                        throw new RpcException(new StringBuilder("Generic serialization [").append(Constants.GENERIC_SERIALIZATION_NATIVE_JAVA)
+                                .append("] serialize result failed.").toString(), e);
                     }
                 } else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
                     return new RpcResult(JavaBeanSerializeUtil.serialize(result.getValue(), JavaBeanAccessor.METHOD));
+                } else if (ProtocolUtils.isProtobufGenericSerialization(generic)) {
+                    try {
+                        UnsafeByteArrayOutputStream os = new UnsafeByteArrayOutputStream(512);
+                        ExtensionLoader.getExtensionLoader(Serialization.class)
+                                .getExtension(Constants.GENERIC_SERIALIZATION_PROTOBUF)
+                                .serialize(null, os).writeObject(result.getValue());
+                        return new RpcResult(os.toString());
+                    } catch (IOException e) {
+                        throw new RpcException(new StringBuilder("Generic serialization [").append(Constants.GENERIC_SERIALIZATION_PROTOBUF)
+                                .append("] serialize result failed.").toString(), e);
+                    }
                 } else {
                     return new RpcResult(PojoUtils.generalize(result.getValue()));
                 }

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/ProtocolUtils.java

@@ -51,7 +51,8 @@ public static boolean isGeneric(String generic) {
                 && !"".equals(generic)
                 && (Constants.GENERIC_SERIALIZATION_DEFAULT.equalsIgnoreCase(generic)  /* Normal generalization cal */
                 || Constants.GENERIC_SERIALIZATION_NATIVE_JAVA.equalsIgnoreCase(generic) /* Streaming generalization call supporting jdk serialization */
-                || Constants.GENERIC_SERIALIZATION_BEAN.equalsIgnoreCase(generic));
+                || Constants.GENERIC_SERIALIZATION_BEAN.equalsIgnoreCase(generic)
+                || Constants.GENERIC_SERIALIZATION_PROTOBUF.equalsIgnoreCase(generic));
     }
 
     public static boolean isDefaultGenericSerialization(String generic) {
@@ -67,4 +68,8 @@ public static boolean isJavaGenericSerialization(String generic) {
     public static boolean isBeanGenericSerialization(String generic) {
         return isGeneric(generic) && Constants.GENERIC_SERIALIZATION_BEAN.equals(generic);
     }
+
+    public static boolean isProtobufGenericSerialization(String generic) {
+        return isGeneric(generic) && Constants.GENERIC_SERIALIZATION_PROTOBUF.equals(generic);
+    }
 }

dubbo-serialization/dubbo-serialization-googlePb/pom.xml

@@ -0,0 +1,46 @@
+<!--
+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>${revision}</version>
+    </parent>
+    <artifactId>dubbo-serialization-googlePb</artifactId>
+    <packaging>jar</packaging>
+    <name>${project.artifactId}</name>
+    <description>The protobuf serialization module of dubbo project</description>
+    <properties>
+        <skip_maven_deploy>false</skip_maven_deploy>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-serialization-api</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java-util</artifactId>
+        </dependency>
+    </dependencies>
+</project>

dubbo-serialization/dubbo-serialization-googlePb/src/main/java/org/apache/dubbo/common/serialize/protobuf/support/GenericProtobufObjectInput.java

@@ -0,0 +1,115 @@
+/*
+ * 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 java.io.BufferedReader;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.lang.reflect.Type;
+
+/**
+ * GenericGoogleProtobuf object input implementation
+ */
+public class GenericProtobufObjectInput implements ObjectInput {
+    private final BufferedReader reader;
+
+    public GenericProtobufObjectInput(InputStream in) {
+        this.reader = new BufferedReader(new InputStreamReader(in));
+    }
+
+    @Override
+    public boolean readBool() throws IOException {
+        return read(boolean.class);
+    }
+
+    @Override
+    public byte readByte() throws IOException {
+        return read(byte.class);
+    }
+
+    @Override
+    public short readShort() throws IOException {
+        return read(short.class);
+    }
+
+    @Override
+    public int readInt() throws IOException {
+        return read(int.class);
+    }
+
+    @Override
+    public long readLong() throws IOException {
+        return read(long.class);
+    }
+
+    @Override
+    public float readFloat() throws IOException {
+        return read(float.class);
+    }
+
+    @Override
+    public double readDouble() throws IOException {
+        return read(double.class);
+    }
+
+    @Override
+    public String readUTF() throws IOException {
+        return read(String.class);
+    }
+
+    @Override
+    public byte[] readBytes() throws IOException {
+        return readLine().getBytes();
+    }
+
+    @Override
+    public Object readObject() throws IOException {
+        return read(String.class);
+    }
+
+    @Override
+    public <T> T readObject(Class<T> cls) throws IOException {
+        return read(cls);
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public <T> T readObject(Class<T> cls, Type type) throws IOException {
+        return readObject(cls);
+    }
+
+    private String readLine() throws IOException {
+        String line = reader.readLine();
+        if (line == null || line.trim().length() == 0) {
+            throw new EOFException();
+        }
+        return line;
+    }
+
+    private <T> T read(Class<T> cls) throws IOException {
+        if (!ProtobufUtils.isSupported(cls)) {
+            throw new IllegalArgumentException("This serialization only support google protobuf entity, the class is :" + cls.getName());
+        }
+
+        String json = readLine();
+        return ProtobufUtils.deserialize(json, cls);
+    }
+}