Fix protobuf class deep copy issue

dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java

@@ -279,7 +279,6 @@ public interface CommonConstants {
     String RELEASE_KEY = "release";
 
     String PROTOBUF_MESSAGE_CLASS_NAME = "com.google.protobuf.Message";
-
     int MAX_PROXY_COUNT = 65535;
 
     String MONITOR_KEY = "monitor";

dubbo-common/src/main/java/org/apache/dubbo/config/AbstractServiceConfig.java

@@ -397,11 +397,11 @@ public void setPayload(Integer payload) {
         this.payload = payload;
     }
 
+    @Parameter(excluded = true)
     public Boolean getUseJavaPackageAsPath() {
         return useJavaPackageAsPath;
     }
 
-    @Parameter(excluded = true)
     public void setUseJavaPackageAsPath(Boolean useJavaPackageAsPath) {
         this.useJavaPackageAsPath = useJavaPackageAsPath;
     }

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java

@@ -17,6 +17,7 @@
 package org.apache.dubbo.remoting.utils;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
 import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
 import org.apache.dubbo.common.utils.StringUtils;
 import org.apache.dubbo.remoting.Constants;
@@ -111,11 +112,18 @@ public static Byte serializationId(URL url) {
      * @return {@link String}
      */
     public static String serializationOrDefault(URL url) {
+        if (useNativeStubSerialization(url)) {
+            return CommonConstants.NATIVE_STUB;
+        }
         //noinspection OptionalGetWithoutIsPresent
         Optional<String> serializations = allSerializations(url).stream().findFirst();
         return serializations.orElseGet(DefaultSerializationSelector::getDefaultRemotingSerialization);
     }
 
+    public static boolean useNativeStubSerialization(URL url) {
+        return CommonConstants.NATIVE_STUB.equals(url.getParameter("proxy", CommonConstants.DEFAULT_PROXY));
+    }
+
     /**
      * Get the all serializations,ensure insertion order
      *

dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/DefaultParamDeepCopyUtil.java → dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/injvm/DefaultParamDeepCopyUtil.java

@@ -22,7 +22,9 @@
 import org.apache.dubbo.common.serialize.ObjectInput;
 import org.apache.dubbo.common.serialize.ObjectOutput;
 import org.apache.dubbo.common.serialize.Serialization;
+import org.apache.dubbo.common.utils.ProtobufUtils;
 import org.apache.dubbo.remoting.utils.UrlUtils;
+import org.apache.dubbo.rpc.model.FrameworkModel;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -37,13 +39,31 @@ public class DefaultParamDeepCopyUtil implements ParamDeepCopyUtil {
 
     public static final String NAME = "default";
 
+    private static final String PB_UTIL_NAME = "protobuf";
+
+    private ParamDeepCopyUtil protobufDeepCopyUtil;
+
+    public DefaultParamDeepCopyUtil() {}
+
+    public void setFrameworkModel(FrameworkModel frameworkModel) {
+        try {
+            this.protobufDeepCopyUtil =
+                    frameworkModel.getExtensionLoader(ParamDeepCopyUtil.class).getExtension(PB_UTIL_NAME);
+        } catch (IllegalStateException illegalStateException) {
+            this.protobufDeepCopyUtil = null;
+        }
+    }
+
     @Override
     @SuppressWarnings({"unchecked"})
     public <T> T copy(URL url, Object src, Class<T> targetClass, Type type) {
+        //TODO: maybe we have better way to do this
+        if(src != null && ProtobufUtils.isProtobufClass(src.getClass())){
+            return (T)src;
+        }
         Serialization serialization = url.getOrDefaultFrameworkModel()
                 .getExtensionLoader(Serialization.class)
                 .getExtension(UrlUtils.serializationOrDefault(url));
-
         try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
             ObjectOutput objectOutput = serialization.serialize(url, outputStream);
             objectOutput.writeObject(src);

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/serialize/ProtobufObjectInput.java

@@ -0,0 +1,101 @@
+/*
+ * 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.rpc.protocol.tri.serialize;
+
+import org.apache.dubbo.common.serialize.ObjectInput;
+import org.apache.dubbo.rpc.protocol.tri.SingleProtobufUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Type;
+
+import com.google.protobuf.BoolValue;
+import com.google.protobuf.BytesValue;
+import com.google.protobuf.DoubleValue;
+import com.google.protobuf.FloatValue;
+import com.google.protobuf.Int32Value;
+import com.google.protobuf.Int64Value;
+import com.google.protobuf.StringValue;
+
+public class ProtobufObjectInput implements ObjectInput {
+
+    private final InputStream input;
+
+    public ProtobufObjectInput(InputStream input) {
+        this.input = input;
+    }
+
+    @Override
+    public Object readObject() {
+        throw new UnsupportedOperationException("Protobuf serialization does not support readObject()");
+    }
+
+    @Override
+    public <T> T readObject(Class<T> cls) throws IOException {
+        return SingleProtobufUtils.deserialize(input, cls);
+    }
+
+    @Override
+    public <T> T readObject(Class<T> cls, Type type) throws IOException {
+        return SingleProtobufUtils.deserialize(input, cls);
+    }
+
+    @Override
+    public boolean readBool() throws IOException {
+        return BoolValue.parseFrom(input).getValue();
+    }
+
+    @Override
+    public byte readByte() throws IOException {
+        return (byte) input.read();
+    }
+
+    @Override
+    public short readShort() throws IOException {
+        return (short) Int32Value.parseFrom(input).getValue();
+    }
+
+    @Override
+    public int readInt() throws IOException {
+        return Int32Value.parseFrom(input).getValue();
+    }
+
+    @Override
+    public long readLong() throws IOException {
+        return Int64Value.parseFrom(input).getValue();
+    }
+
+    @Override
+    public float readFloat() throws IOException {
+        return FloatValue.parseFrom(input).getValue();
+    }
+
+    @Override
+    public double readDouble() throws IOException {
+        return DoubleValue.parseFrom(input).getValue();
+    }
+
+    @Override
+    public String readUTF() throws IOException {
+        return StringValue.parseFrom(input).getValue();
+    }
+
+    @Override
+    public byte[] readBytes() throws IOException {
+        return BytesValue.parseFrom(input).getValue().toByteArray();
+    }
+}

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/serialize/ProtobufObjectOutput.java

@@ -0,0 +1,105 @@
+/*
+ * 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.rpc.protocol.tri.serialize;
+
+import org.apache.dubbo.common.serialize.ObjectOutput;
+import org.apache.dubbo.rpc.protocol.tri.SingleProtobufUtils;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import com.google.protobuf.BoolValue;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.BytesValue;
+import com.google.protobuf.DoubleValue;
+import com.google.protobuf.FloatValue;
+import com.google.protobuf.Int32Value;
+import com.google.protobuf.Int64Value;
+import com.google.protobuf.StringValue;
+
+public class ProtobufObjectOutput implements ObjectOutput {
+
+    private final OutputStream output;
+
+    public ProtobufObjectOutput(OutputStream output) {
+        this.output = output;
+    }
+
+    @Override
+    public void writeObject(Object obj) throws IOException {
+        SingleProtobufUtils.serialize(obj, output);
+    }
+
+    @Override
+    public void writeBool(boolean v) throws IOException {
+        output.write(BoolValue.newBuilder().setValue(v).build().toByteArray());
+    }
+
+    @Override
+    public void writeByte(byte v) throws IOException {
+        output.write(v);
+    }
+
+    @Override
+    public void writeShort(short v) throws IOException {
+        output.write(Int32Value.newBuilder().setValue(v).build().toByteArray());
+    }
+
+    @Override
+    public void writeInt(int v) throws IOException {
+        output.write(Int32Value.newBuilder().setValue(v).build().toByteArray());
+    }
+
+    @Override
+    public void writeLong(long v) throws IOException {
+        output.write(Int64Value.newBuilder().setValue(v).build().toByteArray());
+    }
+
+    @Override
+    public void writeFloat(float v) throws IOException {
+        output.write(FloatValue.newBuilder().setValue(v).build().toByteArray());
+    }
+
+    @Override
+    public void writeDouble(double v) throws IOException {
+        output.write(DoubleValue.newBuilder().setValue(v).build().toByteArray());
+    }
+
+    @Override
+    public void writeUTF(String v) throws IOException {
+        output.write(StringValue.newBuilder().setValue(v).build().toByteArray());
+    }
+
+    @Override
+    public void writeBytes(byte[] v) throws IOException {
+        output.write(
+                BytesValue.newBuilder().setValue(ByteString.copyFrom(v)).build().toByteArray());
+    }
+
+    @Override
+    public void writeBytes(byte[] v, int off, int len) throws IOException {
+        output.write(BytesValue.newBuilder()
+                .setValue(ByteString.copyFrom(v, off, len))
+                .build()
+                .toByteArray());
+    }
+
+    @Override
+    public void flushBuffer() throws IOException {
+        output.flush();
+    }
+}

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/serialize/ProtobufSerialization.java

@@ -0,0 +1,52 @@
+/*
+ * 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.rpc.protocol.tri.serialize;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.serialize.Constants;
+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 {
+
+    private static final String NAME = "protobuf";
+
+    @Override
+    public byte getContentTypeId() {
+        return Constants.PROTOBUF_SERIALIZATION_ID;
+    }
+
+    @Override
+    public String getContentType() {
+        return "application/protobuf";
+    }
+
+    @Override
+    public ObjectOutput serialize(URL url, OutputStream output) throws IOException {
+        return new ProtobufObjectOutput(output);
+    }
+
+    @Override
+    public ObjectInput deserialize(URL url, InputStream input) {
+        return new ProtobufObjectInput(input);
+    }
+}

dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.serialize.Serialization

@@ -0,0 +1 @@
+nativestub=org.apache.dubbo.rpc.protocol.tri.serialize.ProtobufSerialization