This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.2 by this push:
new f2cee0ab89 :sparkles: Enable consumers to request service providers by
priority … (#10533)
f2cee0ab89 is described below
commit f2cee0ab893bf0f0f22d64227f075d4a154a6175
Author: XS <[email protected]>
AuthorDate: Sat Oct 8 11:23:01 2022 +0800
:sparkles: Enable consumers to request service providers by priority …
(#10533)
* :sparkles: Enable consumers to request service providers by priority
using the appropriate serialization type
* :pencil2: Fixing PR specification
* :pencil2: Optimization of import
* :tada: Optimization of import and comment
* :tada: Optimization of import
* :tada: Optimization of import
* :tada: fix Junit
* :tada: Default prefer serialization is `hessian2`
* fix comment
---
.../dubbo/common/constants/ProviderConstants.java | 27 +++-----
.../apache/dubbo/config/AbstractServiceConfig.java | 26 ++++++++
.../bootstrap/builders/AbstractServiceBuilder.java | 23 ++++++-
.../dubbo/config/AbstractServiceConfigTest.java | 7 ++
.../registry/integration/RegistryProtocol.java | 5 +-
.../java/org/apache/dubbo/remoting/Constants.java | 5 ++
.../dubbo/remoting/transport/CodecSupport.java | 39 ++++++-----
.../org/apache/dubbo/remoting/utils/UrlUtils.java | 78 ++++++++++++++++++++++
.../apache/dubbo/rpc/filter/CompatibleFilter.java | 5 +-
.../apache/dubbo/rpc/protocol/AbstractInvoker.java | 49 ++++++++------
.../protocol/dubbo/DecodeableRpcInvocation.java | 2 +-
.../rpc/protocol/dubbo/DecodeableRpcResult.java | 4 +-
.../rpc/protocol/dubbo/DubboCodecSupport.java | 9 +--
.../protocol/dubbo/DubboInvokerAvailableTest.java | 20 ++++++
.../protocol/injvm/DefaultParamDeepCopyUtil.java | 5 +-
.../rpc/protocol/tri/ReflectionPackableMethod.java | 23 +++----
16 files changed, 239 insertions(+), 88 deletions(-)
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/ProviderConstants.java
similarity index 52%
copy from
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java
copy to
dubbo-common/src/main/java/org/apache/dubbo/common/constants/ProviderConstants.java
index 9b38c7b002..36aa6dff05 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/ProviderConstants.java
@@ -14,24 +14,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.dubbo.common.constants;
-package org.apache.dubbo.remoting.utils;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.remoting.Constants;
-
-public class UrlUtils {
- public static int getIdleTimeout(URL url) {
- int heartBeat = getHeartbeat(url);
- // idleTimeout should be at least more than twice heartBeat because
possible retries of client.
- int idleTimeout = url.getParameter(Constants.HEARTBEAT_TIMEOUT_KEY,
heartBeat * 3);
- if (idleTimeout < heartBeat * 2) {
- throw new IllegalStateException("idleTimeout < heartbeatInterval *
2");
- }
- return idleTimeout;
- }
+/**
+ * Provider Constants
+ */
+public interface ProviderConstants {
- public static int getHeartbeat(URL url) {
- return url.getParameter(Constants.HEARTBEAT_KEY,
Constants.DEFAULT_HEARTBEAT);
- }
+ /**
+ * Default prefer serialization,multiple separated by commas
+ */
+ String DEFAULT_PREFER_SERIALIZATION = "hessian2";
}
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/config/AbstractServiceConfig.java
b/dubbo-common/src/main/java/org/apache/dubbo/config/AbstractServiceConfig.java
index ac863c27bb..e5f00d9a57 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/config/AbstractServiceConfig.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/config/AbstractServiceConfig.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.config;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.support.Parameter;
import org.apache.dubbo.rpc.model.ModuleModel;
@@ -27,6 +28,7 @@ import java.util.List;
import static
org.apache.dubbo.common.constants.CommonConstants.EXPORTER_LISTENER_KEY;
import static
org.apache.dubbo.common.constants.CommonConstants.EXPORT_ASYNC_KEY;
import static
org.apache.dubbo.common.constants.CommonConstants.SERVICE_FILTER_KEY;
+import static
org.apache.dubbo.common.constants.ProviderConstants.DEFAULT_PREFER_SERIALIZATION;
/**
* AbstractServiceConfig
@@ -121,6 +123,18 @@ public abstract class AbstractServiceConfig extends
AbstractInterfaceConfig {
*/
private String serialization;
+ /**
+ * If the parameter has a value, the consumer will read the parameter
first.
+ * If the Dubbo Sdk you are using contains the serialization type, the
serialization method specified by the argument is used.
+ * <p>
+ * When this parameter is null or the serialization type specified by this
parameter does not exist in the Dubbo SDK, the serialization type specified by
serialization is used.
+ * If the Dubbo SDK if still does not exist, the default type of the Dubbo
SDK is used.
+ * For Dubbo SDK >= 3.1, <code>preferSerialization</code> takes precedence
over <code>serialization</code>
+ * <p>
+ * The configuration supports multiple, which are separated by commas.Such
as:<code>fastjson2,fastjson,hessian2</code>
+ */
+ private String preferSerialization; // default:hessian2
+
/**
* Weather the service is export asynchronously
* @deprecated
@@ -145,6 +159,10 @@ public abstract class AbstractServiceConfig extends
AbstractInterfaceConfig {
if (dynamic == null) {
dynamic = true;
}
+
+ if (StringUtils.isBlank(preferSerialization)) {
+ preferSerialization = DEFAULT_PREFER_SERIALIZATION;
+ }
}
@Override
@@ -323,6 +341,14 @@ public abstract class AbstractServiceConfig extends
AbstractInterfaceConfig {
this.serialization = serialization;
}
+ public String getPreferSerialization() {
+ return preferSerialization;
+ }
+
+ public void setPreferSerialization(String preferSerialization) {
+ this.preferSerialization = preferSerialization;
+ }
+
@Deprecated
@Parameter(key = EXPORT_ASYNC_KEY)
public Boolean getExportAsync() {
diff --git
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/AbstractServiceBuilder.java
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/AbstractServiceBuilder.java
index a4f5644523..c3ddc76e66 100644
---
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/AbstractServiceBuilder.java
+++
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/AbstractServiceBuilder.java
@@ -29,7 +29,7 @@ import java.util.List;
* @since 2.7
*/
public abstract class AbstractServiceBuilder<T extends AbstractServiceConfig,
B extends AbstractServiceBuilder<T, B>>
- extends AbstractInterfaceBuilder<T, B> {
+ extends AbstractInterfaceBuilder<T, B> {
/**
* The service version
@@ -107,6 +107,11 @@ public abstract class AbstractServiceBuilder<T extends
AbstractServiceConfig, B
*/
private String serialization;
+ /**
+ * The prefer serialization type
+ */
+ private String preferSerialization;
+
public B version(String version) {
this.version = version;
return getThis();
@@ -211,11 +216,22 @@ public abstract class AbstractServiceBuilder<T extends
AbstractServiceConfig, B
return getThis();
}
- public B serialization(String serialization) {
+ public B serialization(String serialization) {
this.serialization = serialization;
return getThis();
}
+ /**
+ * The prefer serialization type
+ *
+ * @param preferSerialization prefer serialization type
+ * @return {@link B}
+ */
+ public B preferSerialization(String preferSerialization) {
+ this.preferSerialization = preferSerialization;
+ return getThis();
+ }
+
@Override
public void build(T instance) {
super.build(instance);
@@ -268,5 +284,8 @@ public abstract class AbstractServiceBuilder<T extends
AbstractServiceConfig, B
if (!StringUtils.isEmpty(serialization)) {
instance.setSerialization(serialization);
}
+ if (StringUtils.isNotBlank(preferSerialization)) {
+ instance.setPreferSerialization(preferSerialization);
+ }
}
}
diff --git
a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractServiceConfigTest.java
b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractServiceConfigTest.java
index bd0e7cddbf..dd27a14ca6 100644
---
a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractServiceConfigTest.java
+++
b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractServiceConfigTest.java
@@ -177,6 +177,13 @@ public class AbstractServiceConfigTest {
assertThat(serviceConfig.getSerialization(), equalTo("serialization"));
}
+ @Test
+ public void testPreferSerialization() throws Exception {
+ ServiceConfig serviceConfig = new ServiceConfig();
+ serviceConfig.setPreferSerialization("preferSerialization");
+ assertThat(serviceConfig.getPreferSerialization(),
equalTo("preferSerialization"));
+ }
+
private static class ServiceConfig extends AbstractServiceConfig {
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java
index 05387f55a7..caf5f5d1ee 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java
@@ -118,6 +118,7 @@ import static org.apache.dubbo.remoting.Constants.CODEC_KEY;
import static org.apache.dubbo.remoting.Constants.CONNECTIONS_KEY;
import static org.apache.dubbo.remoting.Constants.EXCHANGER_KEY;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
+import static org.apache.dubbo.remoting.Constants.PREFER_SERIALIZATION_KEY;
import static org.apache.dubbo.rpc.Constants.DEPRECATED_KEY;
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
import static org.apache.dubbo.rpc.Constants.INTERFACES;
@@ -135,7 +136,7 @@ import static
org.apache.dubbo.rpc.model.ScopeModelUtil.getApplicationModel;
*/
public class RegistryProtocol implements Protocol, ScopeModelAware {
public static final String[] DEFAULT_REGISTER_PROVIDER_KEYS = {
- APPLICATION_KEY, CODEC_KEY, EXCHANGER_KEY, SERIALIZATION_KEY,
CLUSTER_KEY, CONNECTIONS_KEY, DEPRECATED_KEY,
+ APPLICATION_KEY, CODEC_KEY, EXCHANGER_KEY, SERIALIZATION_KEY,
PREFER_SERIALIZATION_KEY, CLUSTER_KEY, CONNECTIONS_KEY, DEPRECATED_KEY,
GROUP_KEY, LOADBALANCE_KEY, MOCK_KEY, PATH_KEY, TIMEOUT_KEY,
TOKEN_KEY, VERSION_KEY, WARMUP_KEY,
WEIGHT_KEY, DUBBO_VERSION_KEY, RELEASE_KEY, SIDE_KEY, IPV6_KEY
};
@@ -487,7 +488,7 @@ public class RegistryProtocol implements Protocol,
ScopeModelAware {
Map<String, Object> consumerAttribute = new
HashMap<>(url.getAttributes());
consumerAttribute.remove(REFER_KEY);
String p = isEmpty(parameters.get(PROTOCOL_KEY)) ? CONSUMER :
parameters.get(PROTOCOL_KEY);
- URL consumerUrl = new ServiceConfigURL (
+ URL consumerUrl = new ServiceConfigURL(
p,
null,
null,
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java
index 02186be735..42d9a06a9a 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java
+++
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java
@@ -79,6 +79,11 @@ public interface Constants {
String SERIALIZATION_KEY = "serialization";
+ /**
+ * Prefer serialization
+ */
+ String PREFER_SERIALIZATION_KEY = "prefer_serialization";
+
String DEFAULT_REMOTING_SERIALIZATION_PROPERTY_KEY =
"DUBBO_DEFAULT_SERIALIZATION";
String CODEC_KEY = "codec";
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
index 5f27886342..bdbbb57707 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
+++
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
@@ -24,9 +24,8 @@ import org.apache.dubbo.common.logger.LoggerFactory;
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.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.common.utils.CollectionUtils;
-import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.FrameworkServiceRepository;
@@ -83,11 +82,10 @@ public class CodecSupport {
}
public static Serialization getSerialization(URL url) {
- return
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
- url.getParameter(Constants.SERIALIZATION_KEY,
DefaultSerializationSelector.getDefaultRemotingSerialization()));
+ return
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(UrlUtils.serializationOrDefault(url));
}
- public static Serialization getSerialization(URL url, Byte id) throws
IOException {
+ public static Serialization getSerialization(Byte id) throws IOException {
Serialization result = getSerializationById(id);
if (result == null) {
throw new IOException("Unrecognized serialize type from consumer:
" + id);
@@ -96,7 +94,7 @@ public class CodecSupport {
}
public static ObjectInput deserialize(URL url, InputStream is, byte proto)
throws IOException {
- Serialization s = getSerialization(url, proto);
+ Serialization s = getSerialization(proto);
return s.deserialize(url, is);
}
@@ -167,20 +165,29 @@ public class CodecSupport {
if (CollectionUtils.isEmpty(urls)) {
throw new IOException("Service " + path + " with version " +
version + " not found, invocation rejected.");
} else {
- boolean match = false;
- for (URL url : urls) {
- String serializationName =
url.getParameter(org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY,
DefaultSerializationSelector.getDefaultRemotingSerialization());
- Byte localId = SERIALIZATIONNAME_ID_MAP.get(serializationName);
- if (localId != null && localId.equals(id)) {
- match = true;
- }
- }
- if(!match) {
+ boolean match = urls.stream().anyMatch(url -> isMatch(url, id));
+ if (!match) {
throw new IOException("Unexpected serialization id:" + id + "
received from network, please check if the peer send the right id.");
}
}
}
-
+ /**
+ * Is Match
+ *
+ * @param url url
+ * @param id id
+ * @return boolean
+ */
+ private static boolean isMatch(URL url, Byte id) {
+ Byte localId;
+ for (String serialization : UrlUtils.allSerializations(url)) {
+ localId = SERIALIZATIONNAME_ID_MAP.get(serialization);
+ if (id.equals(localId)) {
+ return true;
+ }
+ }
+ return false;
+ }
}
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java
index 9b38c7b002..5fb9af867f 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java
+++
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java
@@ -18,7 +18,21 @@
package org.apache.dubbo.remoting.utils;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
+import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.transport.CodecSupport;
+
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.dubbo.remoting.Constants.PREFER_SERIALIZATION_KEY;
+import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
public class UrlUtils {
public static int getIdleTimeout(URL url) {
@@ -34,4 +48,68 @@ public class UrlUtils {
public static int getHeartbeat(URL url) {
return url.getParameter(Constants.HEARTBEAT_KEY,
Constants.DEFAULT_HEARTBEAT);
}
+
+ /**
+ * Get the serialization id
+ *
+ * @param url url
+ * @return {@link Byte}
+ */
+ public static Byte serializationId(URL url) {
+ Byte serializationId;
+ // Obtain the value from prefer_serialization. Such
as:fastjson2,hessian2
+ List<String> preferSerials = preferSerialization(url);
+ for (String preferSerial : preferSerials) {
+ if ((serializationId = CodecSupport.getIDByName(preferSerial)) !=
null) {
+ return serializationId;
+ }
+ }
+
+ // Secondly, obtain the value from serialization
+ if ((serializationId =
CodecSupport.getIDByName(url.getParameter(SERIALIZATION_KEY))) != null) {
+ return serializationId;
+ }
+
+ // Finally, use the default serialization type
+ return
CodecSupport.getIDByName(DefaultSerializationSelector.getDefaultRemotingSerialization());
+ }
+
+ /**
+ * Get the serialization or default serialization
+ *
+ * @param url url
+ * @return {@link String}
+ */
+ public static String serializationOrDefault(URL url) {
+ return allSerializations(url).stream().findFirst().get();
+ }
+
+ /**
+ * Get the all serializations,ensure insertion order
+ *
+ * @param url url
+ * @return {@link List}<{@link String}>
+ */
+ public static Collection<String> allSerializations(URL url) {
+ // preferSerialization -> serialization -> default serialization
+ Set<String> serializations = new LinkedHashSet<>();
+ UrlUtils.preferSerialization(url).forEach(serializations::add);
+
Optional.ofNullable(url.getParameter(SERIALIZATION_KEY)).filter(StringUtils::isNotBlank).ifPresent(serializations::add);
+
serializations.add(DefaultSerializationSelector.getDefaultRemotingSerialization());
+ return Collections.unmodifiableSet(serializations);
+ }
+
+ /**
+ * Prefer Serialization
+ *
+ * @param url url
+ * @return {@link List}<{@link String}>
+ */
+ public static List<String> preferSerialization(URL url) {
+ String preferSerialization =
url.getParameter(PREFER_SERIALIZATION_KEY);
+ if (StringUtils.isNotBlank(preferSerialization)) {
+ return
Collections.unmodifiableList(StringUtils.splitToList(preferSerialization, ','));
+ }
+ return Collections.emptyList();
+ }
}
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/CompatibleFilter.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/CompatibleFilter.java
index b7b368f47b..d736e8ae11 100644
---
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/CompatibleFilter.java
+++
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/CompatibleFilter.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CompatibleTypeUtils;
import org.apache.dubbo.common.utils.PojoUtils;
+import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@@ -29,8 +30,6 @@ import org.apache.dubbo.rpc.RpcException;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
-import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
-
/**
* CompatibleFilter make the remote method's return value compatible to
invoker's version of object.
* To make return object compatible it does
@@ -63,7 +62,7 @@ public class CompatibleFilter implements Filter,
Filter.Listener {
Method method =
invoker.getInterface().getMethod(invocation.getMethodName(),
invocation.getParameterTypes());
Class<?> type = method.getReturnType();
Object newValue;
- String serialization =
invoker.getUrl().getParameter(SERIALIZATION_KEY);
+ String serialization =
UrlUtils.serializationOrDefault(invoker.getUrl());
if ("json".equals(serialization) ||
"fastjson".equals(serialization)) {
// If the serialization key is json or fastjson
Type gtype = method.getGenericReturnType();
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java
index be560156d5..eacc24f09e 100644
---
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java
+++
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java
@@ -22,7 +22,6 @@ import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.common.threadpool.ThreadlessExecutor;
import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
import org.apache.dubbo.common.utils.ArrayUtils;
@@ -30,7 +29,7 @@ import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.TimeoutException;
-import org.apache.dubbo.remoting.transport.CodecSupport;
+import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.AsyncRpcResult;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.InvokeMode;
@@ -51,7 +50,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
-import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
import static org.apache.dubbo.rpc.Constants.SERIALIZATION_ID_KEY;
/**
@@ -110,9 +108,7 @@ public abstract class AbstractInvoker<T> implements
Invoker<T> {
}
this.type = type;
this.url = url;
- this.attachment = attachment == null
- ? null
- : Collections.unmodifiableMap(attachment);
+ this.attachment = attachment == null ? null :
Collections.unmodifiableMap(attachment);
}
private static Map<String, Object> convertAttachment(URL url, String[]
keys) {
@@ -169,8 +165,7 @@ public abstract class AbstractInvoker<T> implements
Invoker<T> {
public Result invoke(Invocation inv) throws RpcException {
// if invoker is destroyed due to address refresh from registry, let's
allow the current invoke to proceed
if (isDestroyed()) {
- logger.warn("Invoker for service " + this + " on consumer " +
NetUtils.getLocalHost() + " is destroyed, "
- + ", dubbo version is " + Version.getVersion() + ", this
invoker should not be used any longer");
+ logger.warn("Invoker for service " + this + " on consumer " +
NetUtils.getLocalHost() + " is destroyed, " + ", dubbo version is " +
Version.getVersion() + ", this invoker should not be used any longer");
}
RpcInvocation invocation = (RpcInvocation) inv;
@@ -196,7 +191,24 @@ public abstract class AbstractInvoker<T> implements
Invoker<T> {
RpcUtils.attachInvocationIdIfAsync(getUrl(), inv);
- Byte serializationId =
CodecSupport.getIDByName(getUrl().getParameter(SERIALIZATION_KEY,
DefaultSerializationSelector.getDefaultRemotingSerialization()));
+ attachInvocationSerializationId(inv);
+ }
+
+ /**
+ * Attach Invocation Serialization id
+ * <p>
+ * <ol>
+ * <li>Obtain the value from <code>prefer_serialization</code></li>
+ * <li>If the preceding information is not obtained, obtain the
value from <code>serialization</code></li>
+ * <li>If neither is obtained, use the default value</li>
+ * </ol>
+ * </p>
+ *
+ * @param inv inv
+ */
+ private void attachInvocationSerializationId(RpcInvocation inv) {
+ Byte serializationId = UrlUtils.serializationId(getUrl());
+
if (serializationId != null) {
inv.put(SERIALIZATION_ID_KEY, serializationId);
}
@@ -267,23 +279,18 @@ public abstract class AbstractInvoker<T> implements
Invoker<T> {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- throw new RpcException("Interrupted unexpectedly while waiting for
remote result to return! method: " +
- invocation.getMethodName() + ", provider: " + getUrl() + ",
cause: " + e.getMessage(), e);
+ throw new RpcException("Interrupted unexpectedly while waiting for
remote result to return! method: " + invocation.getMethodName() + ", provider:
" + getUrl() + ", cause: " + e.getMessage(), e);
} catch (ExecutionException e) {
Throwable rootCause = e.getCause();
if (rootCause instanceof TimeoutException) {
- throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke
remote method timeout. method: " +
- invocation.getMethodName() + ", provider: " + getUrl() +
", cause: " + e.getMessage(), e);
+ throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke
remote method timeout. method: " + invocation.getMethodName() + ", provider: "
+ getUrl() + ", cause: " + e.getMessage(), e);
} else if (rootCause instanceof RemotingException) {
- throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed
to invoke remote method: " +
- invocation.getMethodName() + ", provider: " + getUrl() +
", cause: " + e.getMessage(), e);
+ throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed
to invoke remote method: " + invocation.getMethodName() + ", provider: " +
getUrl() + ", cause: " + e.getMessage(), e);
} else {
- throw new RpcException(RpcException.UNKNOWN_EXCEPTION, "Fail
to invoke remote method: " +
- invocation.getMethodName() + ", provider: " + getUrl() +
", cause: " + e.getMessage(), e);
+ throw new RpcException(RpcException.UNKNOWN_EXCEPTION, "Fail
to invoke remote method: " + invocation.getMethodName() + ", provider: " +
getUrl() + ", cause: " + e.getMessage(), e);
}
} catch (java.util.concurrent.TimeoutException e) {
- throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke
remote method timeout. method: " +
- invocation.getMethodName() + ", provider: " + getUrl() + ",
cause: " + e.getMessage(), e);
+ throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke
remote method timeout. method: " + invocation.getMethodName() + ", provider: "
+ getUrl() + ", cause: " + e.getMessage(), e);
} catch (Throwable e) {
throw new RpcException(e.getMessage(), e);
}
@@ -295,9 +302,7 @@ public abstract class AbstractInvoker<T> implements
Invoker<T> {
if (InvokeMode.SYNC == RpcUtils.getInvokeMode(getUrl(), inv)) {
return new ThreadlessExecutor();
}
- return
url.getOrDefaultApplicationModel().getExtensionLoader(ExecutorRepository.class)
- .getDefaultExtension()
- .getExecutor(url);
+ return
url.getOrDefaultApplicationModel().getExtensionLoader(ExecutorRepository.class).getDefaultExtension().getExecutor(url);
}
/**
diff --git
a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java
b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java
index 1f7de2e538..700687e881 100644
---
a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java
+++
b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java
@@ -113,7 +113,7 @@ public class DecodeableRpcInvocation extends RpcInvocation
implements Codec, Dec
@Override
public Object decode(Channel channel, InputStream input) throws
IOException {
- ObjectInput in = CodecSupport.getSerialization(channel.getUrl(),
serializationType)
+ ObjectInput in = CodecSupport.getSerialization(serializationType)
.deserialize(channel.getUrl(), input);
this.put(SERIALIZATION_ID_KEY, serializationType);
diff --git
a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcResult.java
b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcResult.java
index 3c33fcad3f..e95cc2f56b 100644
---
a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcResult.java
+++
b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcResult.java
@@ -86,8 +86,8 @@ public class DecodeableRpcResult extends AppResponse
implements Codec, Decodeabl
if (invocation != null && invocation.getServiceModel() != null) {
Thread.currentThread().setContextClassLoader(invocation.getServiceModel().getClassLoader());
}
- ObjectInput in = CodecSupport.getSerialization(channel.getUrl(),
serializationType)
- .deserialize(channel.getUrl(), input);
+ ObjectInput in = CodecSupport.getSerialization(serializationType)
+ .deserialize(channel.getUrl(), input);
byte flag = in.readByte();
switch (flag) {
diff --git
a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodecSupport.java
b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodecSupport.java
index d0914c0a94..97dccf5232 100644
---
a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodecSupport.java
+++
b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodecSupport.java
@@ -18,9 +18,8 @@ package org.apache.dubbo.rpc.protocol.dubbo;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.serialize.Serialization;
-import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
-import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.transport.CodecSupport;
+import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
@@ -34,8 +33,7 @@ public class DubboCodecSupport {
if (serializationTypeObj != null) {
return CodecSupport.getSerializationById((byte)
serializationTypeObj);
}
- return
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
-
url.getParameter(org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY,
DefaultSerializationSelector.getDefaultRemotingSerialization()));
+ return
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(UrlUtils.serializationOrDefault(url));
}
public static Serialization getResponseSerialization(URL url, AppResponse
appResponse) {
@@ -47,7 +45,6 @@ public class DubboCodecSupport {
return CodecSupport.getSerializationById((byte)
serializationTypeObj);
}
}
- return
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
- url.getParameter(Constants.SERIALIZATION_KEY,
DefaultSerializationSelector.getDefaultRemotingSerialization()));
+ return
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(UrlUtils.serializationOrDefault(url));
}
}
diff --git
a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvokerAvailableTest.java
b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvokerAvailableTest.java
index 1897f4bde0..e83fd1014d 100644
---
a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvokerAvailableTest.java
+++
b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvokerAvailableTest.java
@@ -151,6 +151,26 @@ public class DubboInvokerAvailableTest {
Assertions.assertFalse(invoker.isAvailable());
}
+ /**
+ * The test prefer serialization
+ *
+ * @throws Exception Exception
+ */
+ @Test
+ public void testPreferSerialization() throws Exception {
+ int port = NetUtils.getAvailablePort();
+ URL url = URL.valueOf("dubbo://127.0.0.1:" + port +
"/org.apache.dubbo.rpc.protocol.dubbo.IDemoService?lazy=true&connections=1&timeout=10000&serialization=fastjson&prefer_serialization=fastjson2,hessian2");
+ ProtocolUtils.export(new DemoServiceImpl(), IDemoService.class, url);
+
+ Invoker<?> invoker = protocol.refer(IDemoService.class, url);
+ Assertions.assertTrue(invoker.isAvailable());
+ ExchangeClient exchangeClient = getClients((DubboInvoker<?>)
invoker)[0];
+ Assertions.assertFalse(exchangeClient.isClosed());
+ //invoke method --> init client
+ IDemoService service = (IDemoService) proxy.getProxy(invoker);
+ Assertions.assertEquals("ok", service.get());
+ }
+
private ExchangeClient[] getClients(DubboInvoker<?> invoker) throws
Exception {
Field field = DubboInvoker.class.getDeclaredField("clients");
field.setAccessible(true);
diff --git
a/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/DefaultParamDeepCopyUtil.java
b/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/DefaultParamDeepCopyUtil.java
index 232b586031..7ff20aadc8 100644
---
a/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/DefaultParamDeepCopyUtil.java
+++
b/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/DefaultParamDeepCopyUtil.java
@@ -22,8 +22,7 @@ import org.apache.dubbo.common.logger.LoggerFactory;
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.serialize.support.DefaultSerializationSelector;
-import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.utils.UrlUtils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -38,7 +37,7 @@ public class DefaultParamDeepCopyUtil implements
ParamDeepCopyUtil {
@SuppressWarnings({"unchecked"})
public <T> T copy(URL url, Object src, Class<T> targetClass) {
Serialization serialization =
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
- url.getParameter(Constants.SERIALIZATION_KEY,
DefaultSerializationSelector.getDefaultRemotingSerialization()));
+ UrlUtils.serializationOrDefault(url));
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream())
{
ObjectOutput objectOutput = serialization.serialize(url,
outputStream);
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
index 7ee9e0e66e..ff40386286 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
@@ -17,19 +17,18 @@
package org.apache.dubbo.rpc.protocol.tri;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.Message;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.serialize.MultipleSerialization;
-import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.config.Constants;
+import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.PackableMethod;
import org.apache.dubbo.triple.TripleWrapper;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.Message;
-
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@@ -39,7 +38,6 @@ import java.util.stream.Stream;
import static org.apache.dubbo.common.constants.CommonConstants.$ECHO;
import static
org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
-import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
import static
org.apache.dubbo.rpc.protocol.tri.TripleProtocol.METHOD_ATTR_PACK;
public class ReflectionPackableMethod implements PackableMethod {
@@ -104,8 +102,7 @@ public class ReflectionPackableMethod implements
PackableMethod {
}
public static ReflectionPackableMethod init(MethodDescriptor
methodDescriptor, URL url) {
- final String serializeName = url.getParameter(SERIALIZATION_KEY,
- DefaultSerializationSelector.getDefaultRemotingSerialization());
+ final String serializeName = UrlUtils.serializationOrDefault(url);
Object stored = methodDescriptor.getAttribute(METHOD_ATTR_PACK);
if (stored != null) {
return (ReflectionPackableMethod) stored;
@@ -127,7 +124,7 @@ public class ReflectionPackableMethod implements
PackableMethod {
* @return true if the request and response object is not generated by
protobuf
*/
static boolean needWrap(MethodDescriptor methodDescriptor, Class<?>[]
parameterClasses,
- Class<?> returnClass) {
+ Class<?> returnClass) {
String methodName = methodDescriptor.getMethodName();
// generic call must be wrapped
if (CommonConstants.$INVOKE.equals(methodName) ||
CommonConstants.$INVOKE_ASYNC.equals(
@@ -316,7 +313,7 @@ public class ReflectionPackableMethod implements
PackableMethod {
String serialize;
private WrapResponsePack(MultipleSerialization multipleSerialization,
URL url,
- String returnType) {
+ String returnType) {
this.multipleSerialization = multipleSerialization;
this.url = url;
this.returnType = returnType;
@@ -364,10 +361,10 @@ public class ReflectionPackableMethod implements
PackableMethod {
private final boolean singleArgument;
private WrapRequestPack(MultipleSerialization multipleSerialization,
- URL url,
- String serialize,
- String[] argumentsType,
- boolean singleArgument) {
+ URL url,
+ String serialize,
+ String[] argumentsType,
+ boolean singleArgument) {
this.url = url;
this.serialize = convertHessianToWrapper(serialize);
this.multipleSerialization = multipleSerialization;
