natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'apache-3.1' into apache-3.2 

# Conflicts:
#	dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java
This commit is contained in:
Albumen Kevin 2022-12-16 12:56:56 +08:00
parent e40edb7081 705b327885
commit 80a15197e6
30 changed files with 273 additions and 439 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
dubbo-common/src/main/java/org/apache/dubbo/common/config/CompositeConfiguration.java
View File

@ -89,20 +89,4 @@ public class CompositeConfiguration implements Configuration {
return null;
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
for (Configuration config : configList) {
try {
Object value = config.getProperty(key, defaultValue);
if (!ConfigurationUtils.isEmptyValue(value)) {
return value;
}
} catch (Exception e) {
logger.error("Error when trying to get value for key " + key + " from " + config + ", " +
"will continue to try the next one.");
}
}
return defaultValue;
}
}

32
dubbo-common/src/main/java/org/apache/dubbo/common/config/Configuration.java
View File

@ -127,19 +127,18 @@ public interface Configuration {
* Gets a property from the configuration. The default value will return if the configuration doesn't contain
* the mapping for the specified key.
*
* @param key property to retrieve
* @param key property to retrieve
* @param defaultValue default value
* @return the value to which this configuration maps the specified key, or default value if the configuration
* contains no mapping for this key.
*/
default Object getProperty(String key, Object defaultValue) {
return getInternalProperty(key, defaultValue);
Object value = getInternalProperty(key);
return value != null ? value : defaultValue;
}
Object getInternalProperty(String key);
Object getInternalProperty(String key, Object defaultValue);
/**
* Check if the configuration contains the specified key.
*
@ -154,12 +153,9 @@ public interface Configuration {
default <T> T convert(Class<T> cls, String key, T defaultValue) {
// we only process String properties for now
Object value = getProperty(key, defaultValue);
String value = (String) getProperty(key);
if (!String.class.isInstance(value)) {
if (cls.isInstance(value)) {
return cls.cast(value);
}
if (value == null) {
return defaultValue;
}
@ -168,26 +164,24 @@ public interface Configuration {
return cls.cast(value);
}
String str = (String) value;
if (Boolean.class.equals(cls) || Boolean.TYPE.equals(cls)) {
obj = Boolean.valueOf(str);
obj = Boolean.valueOf(value);
} else if (Number.class.isAssignableFrom(cls) || cls.isPrimitive()) {
if (Integer.class.equals(cls) || Integer.TYPE.equals(cls)) {
obj = Integer.valueOf(str);
obj = Integer.valueOf(value);
} else if (Long.class.equals(cls) || Long.TYPE.equals(cls)) {
obj = Long.valueOf(str);
obj = Long.valueOf(value);
} else if (Byte.class.equals(cls) || Byte.TYPE.equals(cls)) {
obj = Byte.valueOf(str);
obj = Byte.valueOf(value);
} else if (Short.class.equals(cls) || Short.TYPE.equals(cls)) {
obj = Short.valueOf(str);
obj = Short.valueOf(value);
} else if (Float.class.equals(cls) || Float.TYPE.equals(cls)) {
obj = Float.valueOf(str);
obj = Float.valueOf(value);
} else if (Double.class.equals(cls) || Double.TYPE.equals(cls)) {
obj = Double.valueOf(str);
obj = Double.valueOf(value);
}
} else if (cls.isEnum()) {
obj = Enum.valueOf(cls.asSubclass(Enum.class), str);
obj = Enum.valueOf(cls.asSubclass(Enum.class), value);
}
return cls.cast(obj);

2
dubbo-common/src/main/java/org/apache/dubbo/common/config/Environment.java
View File

@ -81,7 +81,7 @@ public class Environment extends LifecycleAdapter implements ApplicationExt {
public void initialize() throws IllegalStateException {
if (initialized.compareAndSet(false, true)) {
this.propertiesConfiguration = new PropertiesConfiguration(scopeModel);
this.systemConfiguration = new SystemConfiguration(scopeModel);
this.systemConfiguration = new SystemConfiguration();
this.environmentConfiguration = new EnvironmentConfiguration();
this.externalConfiguration = new InmemoryConfiguration("ExternalConfig");
this.appExternalConfiguration = new InmemoryConfiguration("AppExternalConfig");

13
dubbo-common/src/main/java/org/apache/dubbo/common/config/EnvironmentConfiguration.java
View File

@ -34,19 +34,6 @@ public class EnvironmentConfiguration implements Configuration {
return value;
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
String value = System.getenv(key);
if (StringUtils.isEmpty(value)) {
value = System.getenv(StringUtils.toOSStyleKey(key));
}
if (StringUtils.isEmpty(value)) {
return defaultValue;
}
return value;
}
public Map<String, String> getProperties() {
return getenv();
}

10
dubbo-common/src/main/java/org/apache/dubbo/common/config/InmemoryConfiguration.java
View File

@ -45,16 +45,6 @@ public class InmemoryConfiguration implements Configuration {
return store.get(key);
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
Object v = store.get(key);
if (v != null) {
return v;
} else {
return defaultValue;
}
}
/**
* Add one property into the store, the previous value will be replaced if the key exists
*/

10
dubbo-common/src/main/java/org/apache/dubbo/common/config/OrderedPropertiesConfiguration.java
View File

@ -68,16 +68,6 @@ public class OrderedPropertiesConfiguration implements Configuration {
return properties.getProperty(key);
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
Object v = properties.getProperty(key);
if (v != null){
return v;
}else {
return defaultValue;
}
}
public void setProperty(String key, String value) {
properties.setProperty(key, value);
}

13
dubbo-common/src/main/java/org/apache/dubbo/common/config/PrefixedConfiguration.java
View File

@ -42,17 +42,4 @@ public class PrefixedConfiguration implements Configuration {
return null;
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
if (StringUtils.isBlank(prefix)) {
return origin.getInternalProperty(key, defaultValue);
}
Object value = origin.getInternalProperty(prefix + "." + key, defaultValue);
if (!ConfigurationUtils.isEmptyValue(value)) {
return value;
}
return null;
}
}

10
dubbo-common/src/main/java/org/apache/dubbo/common/config/PropertiesConfiguration.java
View File

@ -49,16 +49,6 @@ public class PropertiesConfiguration implements Configuration {
return properties.getProperty(key);
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
Object v = properties.getProperty(key);
if (v != null){
return v;
}else {
return defaultValue;
}
}
public void setProperty(String key, String value) {
properties.setProperty(key, value);
}

81
dubbo-common/src/main/java/org/apache/dubbo/common/config/SystemConfiguration.java
View File

@ -17,94 +17,19 @@
package org.apache.dubbo.common.config;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.rpc.model.ScopeModel;
import org.apache.dubbo.rpc.model.ScopeModelUtil;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* FIXME: is this really necessary? PropertiesConfiguration should have already covered this:
*
* @See ConfigUtils#getProperty(String)
* @see PropertiesConfiguration
* Configuration from system properties
*/
public class SystemConfiguration implements Configuration {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(SystemConfiguration.class);
private final Map<String, Object> cache = new ConcurrentHashMap<>();
private final ScheduledExecutorService sharedScheduledExecutor;
public SystemConfiguration(ScopeModel scopeModel) {
sharedScheduledExecutor = ScopeModelUtil.getFrameworkModel(scopeModel).getBeanFactory()
.getBean(FrameworkExecutorRepository.class).getSharedScheduledExecutor();
sharedScheduledExecutor.scheduleWithFixedDelay(() -> {
if (!cache.isEmpty()) {
Set<String> keys = cache.keySet();
keys.forEach((key) -> overwriteCache(key, System.getProperty(key)));
}
}, 60000, 60000, TimeUnit.MILLISECONDS);
}
@Override
public Object getInternalProperty(String key) {
if (cache.containsKey(key)) {
return cache.get(key);
} else {
Object val = System.getProperty(key);
if (val != null) {
cache.putIfAbsent(key, val);
}
return val;
}
return System.getProperty(key);
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
if (cache.containsKey(key)) {
return cache.get(key);
} else {
Object val = System.getProperty(key);
if (val != null) {
cache.putIfAbsent(key, val);
} else {
val = defaultValue;
if (defaultValue != null) {
cache.putIfAbsent(key, defaultValue);
}
}
return val;
}
}
public void overwriteCache(String key, Object value) {
if (value != null) {
cache.put(key, value);
}
}
public void clearCache() {
cache.clear();
}
public Map<String, String> getProperties() {
Properties properties = System.getProperties();
Map<String, String> res = new ConcurrentHashMap<>(properties.size());
try {
res.putAll((Map) properties);
} catch (Exception e) {
logger.warn("System property get failed", e);
}
return res;
return (Map) System.getProperties();
}
}

5
dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/AbstractDynamicConfiguration.java
View File

@ -120,11 +120,6 @@ public abstract class AbstractDynamicConfiguration implements DynamicConfigurati
return null;
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
return null;
}
@Override
public final void close() throws Exception {
try {

5
dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfiguration.java
View File

@ -36,11 +36,6 @@ public class NopDynamicConfiguration implements DynamicConfiguration {
return null;
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
return null;
}
@Override
public void addListener(String key, String group, ConfigurationListener listener) {
// no-op

5
dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/wrapper/CompositeDynamicConfiguration.java
View File

@ -75,11 +75,6 @@ public class CompositeDynamicConfiguration implements DynamicConfiguration {
return iterateConfigOperation(configuration -> configuration.getInternalProperty(key));
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
return iterateConfigOperation(configuration -> configuration.getInternalProperty(key, defaultValue));
}
@Override
public boolean publishConfig(String key, String group, String content) throws UnsupportedOperationException {
boolean publishedAll = true;

10
dubbo-common/src/main/java/org/apache/dubbo/config/context/ConfigConfigurationAdapter.java
View File

@ -42,16 +42,6 @@ public class ConfigConfigurationAdapter implements Configuration {
return metaData.get(key);
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
Object v = metaData.get(key);
if (v != null) {
return v;
} else {
return defaultValue;
}
}
public Map<String, String> getProperties() {
return metaData;
}

21
dubbo-common/src/test/java/org/apache/dubbo/common/config/SystemConfigurationTest.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.dubbo.common.config;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@ -45,7 +44,7 @@ class SystemConfigurationTest {
@BeforeEach
public void init() {
sysConfig = new SystemConfiguration(ApplicationModel.defaultModel().getDefaultModule());
sysConfig = new SystemConfiguration();
}
/**
@ -71,23 +70,23 @@ class SystemConfigurationTest {
void testConvert() {
Assertions.assertEquals(
MOCK_STRING_VALUE, sysConfig.convert(String.class, NOT_EXIST_KEY, MOCK_STRING_VALUE));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(MOCK_BOOL_VALUE));
System.setProperty(MOCK_KEY, String.valueOf(MOCK_BOOL_VALUE));
Assertions.assertEquals(MOCK_BOOL_VALUE, sysConfig.convert(Boolean.class, MOCK_KEY, null));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(MOCK_STRING_VALUE));
System.setProperty(MOCK_KEY, String.valueOf(MOCK_STRING_VALUE));
Assertions.assertEquals(MOCK_STRING_VALUE, sysConfig.convert(String.class, MOCK_KEY, null));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(MOCK_INT_VALUE));
System.setProperty(MOCK_KEY, String.valueOf(MOCK_INT_VALUE));
Assertions.assertEquals(MOCK_INT_VALUE, sysConfig.convert(Integer.class, MOCK_KEY, null));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(MOCK_LONG_VALUE));
System.setProperty(MOCK_KEY, String.valueOf(MOCK_LONG_VALUE));
Assertions.assertEquals(MOCK_LONG_VALUE, sysConfig.convert(Long.class, MOCK_KEY, null));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(MOCK_SHORT_VALUE));
System.setProperty(MOCK_KEY, String.valueOf(MOCK_SHORT_VALUE));
Assertions.assertEquals(MOCK_SHORT_VALUE, sysConfig.convert(Short.class, MOCK_KEY, null));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(MOCK_FLOAT_VALUE));
System.setProperty(MOCK_KEY, String.valueOf(MOCK_FLOAT_VALUE));
Assertions.assertEquals(MOCK_FLOAT_VALUE, sysConfig.convert(Float.class, MOCK_KEY, null));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(MOCK_DOUBLE_VALUE));
System.setProperty(MOCK_KEY, String.valueOf(MOCK_DOUBLE_VALUE));
Assertions.assertEquals(MOCK_DOUBLE_VALUE, sysConfig.convert(Double.class, MOCK_KEY, null));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(MOCK_BYTE_VALUE));
System.setProperty(MOCK_KEY, String.valueOf(MOCK_BYTE_VALUE));
Assertions.assertEquals(MOCK_BYTE_VALUE, sysConfig.convert(Byte.class, MOCK_KEY, null));
sysConfig.overwriteCache(MOCK_KEY, String.valueOf(ConfigMock.MockOne));
System.setProperty(MOCK_KEY, String.valueOf(ConfigMock.MockOne));
Assertions.assertEquals(ConfigMock.MockOne, sysConfig.convert(ConfigMock.class, MOCK_KEY, null));
}

32
dubbo-common/src/test/java/org/apache/dubbo/common/utils/SerializeClassCheckerTest.java
View File

@ -16,11 +16,9 @@
*/
package org.apache.dubbo.common.utils;
import org.apache.dubbo.common.config.SystemConfiguration;
import org.apache.dubbo.common.constants.CommonConstants;
import javassist.compiler.Javac;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@ -59,16 +57,14 @@ class SerializeClassCheckerTest {
serializeClassChecker.validateClass(int.class.getName().toUpperCase(Locale.ROOT));
}
Assertions.assertThrows(IllegalArgumentException.class, () -> {
Assertions.assertThrows(IllegalArgumentException.class, ()-> {
serializeClassChecker.validateClass(Socket.class.getName());
});
}
@Test
void testAddAllow() {
SystemConfiguration systemConfiguration = ApplicationModel.defaultModel().getModelEnvironment().getSystemConfiguration();
systemConfiguration.overwriteCache(CommonConstants.CLASS_DESERIALIZE_ALLOWED_LIST, Socket.class.getName() + "," + Javac.class.getName());
System.setProperty(CommonConstants.CLASS_DESERIALIZE_ALLOWED_LIST, Socket.class.getName() + "," + Javac.class.getName());
SerializeClassChecker serializeClassChecker = SerializeClassChecker.getInstance();
for (int i = 0; i < 10; i++) {
@ -76,44 +72,40 @@ class SerializeClassCheckerTest {
serializeClassChecker.validateClass(Javac.class.getName());
}
systemConfiguration.clearCache();
System.clearProperty(CommonConstants.CLASS_DESERIALIZE_ALLOWED_LIST);
}
@Test
void testAddBlock() {
SystemConfiguration systemConfiguration = ApplicationModel.defaultModel().getModelEnvironment().getSystemConfiguration();
systemConfiguration.overwriteCache(CommonConstants.CLASS_DESERIALIZE_BLOCKED_LIST, LinkedList.class.getName() + "," + Integer.class.getName());
System.setProperty(CommonConstants.CLASS_DESERIALIZE_BLOCKED_LIST, LinkedList.class.getName() + "," + Integer.class.getName());
SerializeClassChecker serializeClassChecker = SerializeClassChecker.getInstance();
for (int i = 0; i < 10; i++) {
Assertions.assertThrows(IllegalArgumentException.class, () -> {
Assertions.assertThrows(IllegalArgumentException.class, ()-> {
serializeClassChecker.validateClass(LinkedList.class.getName());
});
Assertions.assertThrows(IllegalArgumentException.class, () -> {
Assertions.assertThrows(IllegalArgumentException.class, ()-> {
serializeClassChecker.validateClass(Integer.class.getName());
});
}
systemConfiguration.clearCache();
System.clearProperty(CommonConstants.CLASS_DESERIALIZE_BLOCKED_LIST);
}
@Test
void testBlockAll() {
SystemConfiguration systemConfiguration = ApplicationModel.defaultModel().getModelEnvironment().getSystemConfiguration();
systemConfiguration.overwriteCache(CommonConstants.CLASS_DESERIALIZE_BLOCK_ALL, "true");
systemConfiguration.overwriteCache(CommonConstants.CLASS_DESERIALIZE_ALLOWED_LIST, LinkedList.class.getName());
System.setProperty(CommonConstants.CLASS_DESERIALIZE_BLOCK_ALL, "true");
System.setProperty(CommonConstants.CLASS_DESERIALIZE_ALLOWED_LIST, LinkedList.class.getName());
SerializeClassChecker serializeClassChecker = SerializeClassChecker.getInstance();
for (int i = 0; i < 10; i++) {
serializeClassChecker.validateClass(LinkedList.class.getName());
Assertions.assertThrows(IllegalArgumentException.class, () -> {
Assertions.assertThrows(IllegalArgumentException.class, ()-> {
serializeClassChecker.validateClass(Integer.class.getName());
});
}
systemConfiguration.clearCache();
System.clearProperty(CommonConstants.CLASS_DESERIALIZE_BLOCK_ALL);
System.clearProperty(CommonConstants.CLASS_DESERIALIZE_ALLOWED_LIST);
}
}

19
dubbo-common/src/test/java/org/apache/dubbo/config/context/ConfigManagerTest.java
View File

@ -393,33 +393,22 @@ class ConfigManagerTest {
@Test
void testLoadConfigsOfTypeFromProps() {
try {
System.setProperty("dubbo.protocols.dubbo1.port", "20880");
System.setProperty("dubbo.protocols.dubbo2.port", "20881");
System.setProperty("dubbo.protocols.rest1.port", "8080");
System.setProperty("dubbo.protocols.rest2.port", "8081");
ApplicationModel.defaultModel().destroy();
ApplicationModel applicationModel = ApplicationModel.defaultModel();
configManager = applicationModel.getApplicationConfigManager();
moduleConfigManager = applicationModel.getDefaultModule().getConfigManager();
// dubbo.application.enable-file-cache = false
configManager.loadConfigsOfTypeFromProps(ApplicationConfig.class);
Optional<ApplicationConfig> application = configManager.getApplication();
Assertions.assertTrue(application.isPresent());
configManager.removeConfig(application.get());
System.setProperty("dubbo.protocols.dubbo1.port", "20880");
System.setProperty("dubbo.protocols.dubbo2.port", "20881");
System.setProperty("dubbo.protocols.rest1.port", "8080");
System.setProperty("dubbo.protocols.rest2.port", "8081");
configManager.loadConfigsOfTypeFromProps(ProtocolConfig.class);
Collection<ProtocolConfig> protocols = configManager.getProtocols();
Assertions.assertEquals(protocols.size(), 4);
System.setProperty("dubbo.applications.app1.name", "app-demo1");
System.setProperty("dubbo.applications.app2.name", "app-demo2");
ApplicationModel.defaultModel().destroy();
applicationModel = ApplicationModel.defaultModel();
configManager = applicationModel.getApplicationConfigManager();
moduleConfigManager = applicationModel.getDefaultModule().getConfigManager();
try {
configManager.loadConfigsOfTypeFromProps(ApplicationConfig.class);
Assertions.fail();

40
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/bootstrap/MultiInstanceTest.java
View File

@ -120,7 +120,7 @@ class MultiInstanceTest {
@AfterEach
public void afterEach() {
SysProps.clear();
FrameworkModel.destroyAll();
DubboBootstrap.reset();
}
@Test
@ -152,12 +152,12 @@ class MultiInstanceTest {
@Test
void testDefaultProviderApplication() {
DubboBootstrap.reset();
DubboBootstrap dubboBootstrap = DubboBootstrap.getInstance();
try {
configProviderApp(dubboBootstrap).start();
} finally {
dubboBootstrap.destroy();
DubboBootstrap.reset();
}
}
@ -172,12 +172,13 @@ class MultiInstanceTest {
Assertions.assertTrue(e.toString().contains("No provider available"), StringUtils.toString(e));
} finally {
dubboBootstrap.destroy();
DubboBootstrap.reset();
SysProps.clear();
}
}
@Test
void testDefaultMixedApplication() {
DubboBootstrap.reset();
DubboBootstrap dubboBootstrap = DubboBootstrap.getInstance();
try {
dubboBootstrap.application("mixed-app");
@ -188,6 +189,7 @@ class MultiInstanceTest {
testConsumer(dubboBootstrap);
} finally {
dubboBootstrap.destroy();
DubboBootstrap.reset();
}
}
@ -217,8 +219,6 @@ class MultiInstanceTest {
void testMultiModuleApplication() throws InterruptedException {
//SysProps.setProperty(METADATA_PUBLISH_DELAY_KEY, "100");
FrameworkModel frameworkModel = new FrameworkModel();
String version1 = "1.0";
String version2 = "2.0";
String version3 = "3.0";
@ -228,7 +228,7 @@ class MultiInstanceTest {
try {
// provider app
providerBootstrap = DubboBootstrap.newInstance(frameworkModel);
providerBootstrap = DubboBootstrap.newInstance();
ServiceConfig serviceConfig1 = new ServiceConfig();
serviceConfig1.setInterface(DemoService.class);
@ -272,7 +272,7 @@ class MultiInstanceTest {
//Thread.sleep(200);
// consumer app
consumerBootstrap = DubboBootstrap.newInstance(frameworkModel);
consumerBootstrap = DubboBootstrap.newInstance();
consumerBootstrap.application("consumer-app")
.registry(registryConfig)
.reference(builder -> builder
@ -306,12 +306,10 @@ class MultiInstanceTest {
@Test
void testMultiProviderApplicationsStopOneByOne() {
DubboBootstrap.reset();
String version1 = "1.0";
String version2 = "2.0";
FrameworkModel frameworkModel = new FrameworkModel();
DubboBootstrap providerBootstrap1 = null;
DubboBootstrap providerBootstrap2 = null;
@ -328,7 +326,7 @@ class MultiInstanceTest {
ProtocolConfig protocolConfig1 = new ProtocolConfig("dubbo", NetUtils.getAvailablePort());
providerBootstrap1 = DubboBootstrap.newInstance(frameworkModel);
providerBootstrap1 = DubboBootstrap.getInstance();
providerBootstrap1.application("provider1")
.registry(new RegistryConfig(registryConfig.getAddress()))
.service(serviceConfig1)
@ -351,7 +349,7 @@ class MultiInstanceTest {
ProtocolConfig protocolConfig2 = new ProtocolConfig("dubbo", NetUtils.getAvailablePort());
providerBootstrap2 = DubboBootstrap.newInstance(frameworkModel);
providerBootstrap2 = DubboBootstrap.newInstance();
providerBootstrap2.application("provider2")
.registry(registryConfig2)
.service(serviceConfig2)
@ -374,7 +372,7 @@ class MultiInstanceTest {
// stop provider app 2 and check threads
providerBootstrap2.stop();
// shutdown register center after dubbo application to avoid unregister services blocking
checkUnclosedThreadsOfApp(stackTraces2, "Found unclosed threads of app 2: ", new String[]{"Dubbo-framework-shared-scheduler"});
checkUnclosedThreadsOfApp(stackTraces2, "Found unclosed threads of app 2: ", null);
} finally {
if (providerBootstrap1 != null) {
@ -430,7 +428,6 @@ class MultiInstanceTest {
String version2 = "2.0";
String version3 = "3.0";
FrameworkModel frameworkModel = new FrameworkModel();
String serviceKey1 = DemoService.class.getName() + ":" + version1;
String serviceKey2 = DemoService.class.getName() + ":" + version2;
String serviceKey3 = DemoService.class.getName() + ":" + version3;
@ -440,7 +437,7 @@ class MultiInstanceTest {
try {
// provider app
providerBootstrap = DubboBootstrap.newInstance(frameworkModel);
providerBootstrap = DubboBootstrap.newInstance();
ServiceConfig serviceConfig1 = new ServiceConfig();
serviceConfig1.setInterface(DemoService.class);
@ -478,7 +475,7 @@ class MultiInstanceTest {
Assertions.assertNull(frameworkServiceRepository.lookupExportedServiceWithoutGroup(serviceKey3));
// consumer module 1
consumerBootstrap = DubboBootstrap.newInstance(frameworkModel);
consumerBootstrap = DubboBootstrap.newInstance();
consumerBootstrap.application("consumer-app")
.registry(registryConfig)
.reference(builder -> builder
@ -573,7 +570,6 @@ class MultiInstanceTest {
String version2 = "2.0";
String version3 = "3.0";
FrameworkModel frameworkModel = new FrameworkModel();
String serviceKey1 = DemoService.class.getName() + ":" + version1;
String serviceKey2 = DemoService.class.getName() + ":" + version2;
String serviceKey3 = DemoService.class.getName() + ":" + version3;
@ -581,7 +577,7 @@ class MultiInstanceTest {
// provider app
DubboBootstrap providerBootstrap = null;
try {
providerBootstrap = DubboBootstrap.newInstance(frameworkModel);
providerBootstrap = DubboBootstrap.newInstance();
ServiceConfig serviceConfig1 = new ServiceConfig();
serviceConfig1.setInterface(DemoService.class);
@ -654,7 +650,6 @@ class MultiInstanceTest {
String version2 = "2.0";
String version3 = "3.0";
FrameworkModel frameworkModel = new FrameworkModel();
String serviceKey1 = DemoService.class.getName() + ":" + version1;
String serviceKey2 = DemoService.class.getName() + ":" + version2;
String serviceKey3 = DemoService.class.getName() + ":" + version3;
@ -662,7 +657,7 @@ class MultiInstanceTest {
// provider app
DubboBootstrap providerBootstrap = null;
try {
providerBootstrap = DubboBootstrap.newInstance(frameworkModel);
providerBootstrap = DubboBootstrap.newInstance();
ServiceConfig serviceConfig1 = new ServiceConfig();
serviceConfig1.setInterface(DemoService.class);
@ -709,7 +704,6 @@ class MultiInstanceTest {
@Test
void testOldApiDeploy() throws Exception {
DubboBootstrap.reset();
try {
// provider app
ApplicationModel providerApplicationModel = ApplicationModel.defaultModel();
@ -785,10 +779,8 @@ class MultiInstanceTest {
@Test
void testAsyncExportAndReferServices() throws ExecutionException, InterruptedException {
FrameworkModel frameworkModel = new FrameworkModel();
DubboBootstrap providerBootstrap = DubboBootstrap.newInstance(frameworkModel);
DubboBootstrap consumerBootstrap = DubboBootstrap.newInstance(frameworkModel);
DubboBootstrap providerBootstrap = DubboBootstrap.newInstance();
DubboBootstrap consumerBootstrap = DubboBootstrap.newInstance();
try {
ServiceConfig serviceConfig = new ServiceConfig();

12
dubbo-configcenter/dubbo-configcenter-apollo/src/main/java/org/apache/dubbo/configcenter/support/apollo/ApolloDynamicConfiguration.java
View File

@ -47,8 +47,8 @@ import static org.apache.dubbo.common.constants.CommonConstants.CHECK_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.CLUSTER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
import static org.apache.dubbo.common.constants.CommonConstants.CONFIG_NAMESPACE_KEY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_FAILED_CONNECT_REGISTRY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_FAILED_CLOSE_CONNECT_APOLLO;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_FAILED_CONNECT_REGISTRY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_NOT_EFFECT_EMPTY_RULE_APOLLO;
/**
@ -215,16 +215,6 @@ public class ApolloDynamicConfiguration implements DynamicConfiguration {
return dubboConfig.getProperty(key, null);
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
Object v = dubboConfig.getProperty(key, null);
if (v != null) {
return v;
} else {
return defaultValue;
}
}
/**
* Ignores the group parameter.
*

11
dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java
View File

@ -220,17 +220,6 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
return null;
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
Object v = defaultValue;
try {
v = configService.getConfig(key, DEFAULT_GROUP, getDefaultTimeout());
} catch (NacosException e) {
logger.error(e.getMessage());
}
return v;
}
@Override
public boolean publishConfig(String key, String group, String content) {
boolean published = false;

10
dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java
View File

@ -85,16 +85,6 @@ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration
return zkClient.getContent(buildPathKey("", key));
}
@Override
public Object getInternalProperty(String key, Object defaultValue) {
Object v = zkClient.getContent(buildPathKey("", key));
if (v != null) {
return v;
} else {
return defaultValue;
}
}
@Override
protected void doClose() throws Exception {
// remove data listener

11
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java
View File

@ -66,6 +66,11 @@ public class AsyncRpcResult implements Result {
private CompletableFuture<AppResponse> responseFuture;
/**
* Whether set future to Thread Local when invocation mode is sync
*/
private static final boolean setFutureWhenSync = Boolean.parseBoolean(System.getProperty(CommonConstants.SET_FUTURE_IN_SYNC_MODE, "true"));
public AsyncRpcResult(CompletableFuture<AppResponse> future, Invocation invocation) {
this.responseFuture = future;
this.invocation = invocation;
@ -211,11 +216,7 @@ public class AsyncRpcResult implements Result {
fn.accept(v, t);
});
// Whether set future to Thread Local when invocation mode is sync
String setFutureWhenSync = invocation.getModuleModel().getModelEnvironment().getSystemConfiguration()
.getString(CommonConstants.SET_FUTURE_IN_SYNC_MODE, "true");
if (Boolean.parseBoolean(setFutureWhenSync) || ((RpcInvocation) invocation).getInvokeMode() != InvokeMode.SYNC) {
if (setFutureWhenSync || ((RpcInvocation) invocation).getInvokeMode() != InvokeMode.SYNC) {
// Necessary! update future in context, see https://github.com/apache/dubbo/issues/9461
RpcContext.getServiceContext().setFuture(new FutureAdapter<>(this.responseFuture));
}

4
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AccessLogFilter.java
View File

@ -81,7 +81,7 @@ public class AccessLogFilter implements Filter {
private AtomicBoolean scheduled = new AtomicBoolean();
private final String LINE_SEPARATOR = System.getProperty("line.separator");
private static final String LINE_SEPARATOR = "line.separator";
/**
* Default constructor initialize demon thread for writing into access log file with names with access log key
@ -173,7 +173,7 @@ public class AccessLogFilter implements Filter {
try {
while (!logQueue.isEmpty()) {
writer.write(logQueue.poll().getLogMessage());
writer.write(LINE_SEPARATOR);
writer.write(System.getProperty(LINE_SEPARATOR));
}
} finally {
writer.flush();

11
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java
View File

@ -85,6 +85,11 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
*/
private boolean destroyed = false;
/**
* Whether set future to Thread Local when invocation mode is sync
*/
private static final boolean setFutureWhenSync = Boolean.parseBoolean(System.getProperty(CommonConstants.SET_FUTURE_IN_SYNC_MODE, "true"));
// -- Constructor
public AbstractInvoker(Class<T> type, URL url) {
@ -249,11 +254,7 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
}
// Whether set future to Thread Local when invocation mode is sync
String setFutureWhenSync = invocation.getModuleModel().getModelEnvironment().getSystemConfiguration()
.getString(CommonConstants.SET_FUTURE_IN_SYNC_MODE, "true");
if (Boolean.parseBoolean(setFutureWhenSync) || invocation.getInvokeMode() != InvokeMode.SYNC) {
if (setFutureWhenSync || invocation.getInvokeMode() != InvokeMode.SYNC) {
// set server context
RpcContext.getServiceContext().setFuture(new FutureAdapter<>(asyncResult.getResponseFuture()));
}

14
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java
View File

@ -17,7 +17,6 @@
package org.apache.dubbo.rpc.protocol.dubbo;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.utils.CacheableSupplier;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
@ -39,7 +38,6 @@ import org.apache.dubbo.rpc.model.FrameworkServiceRepository;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.ModuleModel;
import org.apache.dubbo.rpc.model.ProviderModel;
import org.apache.dubbo.rpc.model.ScopeModel;
import org.apache.dubbo.rpc.model.ServiceDescriptor;
import org.apache.dubbo.rpc.support.RpcUtils;
@ -139,17 +137,7 @@ public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Dec
ClassLoader originClassLoader = Thread.currentThread().getContextClassLoader();
try {
ScopeModel scopeModel = channel.getUrl().getScopeModel();
if (scopeModel instanceof ModuleModel) {
scopeModel = scopeModel.getParent();
} else {
scopeModel = ApplicationModel.defaultModel();
}
String serializationSecurityCheck = ConfigurationUtils.getSystemConfiguration(
scopeModel).getString(SERIALIZATION_SECURITY_CHECK_KEY, "true");
if (Boolean.parseBoolean(serializationSecurityCheck)) {
if (Boolean.parseBoolean(System.getProperty(SERIALIZATION_SECURITY_CHECK_KEY, "true"))) {
CodecSupport.checkSerialization(frameworkModel.getServiceRepository(), path, version, serializationType);
}
Object[] args = DubboCodec.EMPTY_OBJECT_ARRAY;

187
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/decode/DubboTelnetDecodeTest.java
View File

@ -16,9 +16,6 @@
*/
package org.apache.dubbo.rpc.protocol.dubbo.decode;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.url.component.ServiceConfigURL;
@ -41,6 +38,10 @@ import org.apache.dubbo.rpc.model.ModuleServiceRepository;
import org.apache.dubbo.rpc.protocol.dubbo.DecodeableRpcInvocation;
import org.apache.dubbo.rpc.protocol.dubbo.DubboCodec;
import org.apache.dubbo.rpc.protocol.dubbo.support.DemoService;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@ -77,13 +78,13 @@ class DubboTelnetDecodeTest {
// disable org.apache.dubbo.remoting.transport.CodecSupport.checkSerialization to avoid error:
// java.io.IOException: Service org.apache.dubbo.rpc.protocol.dubbo.support.DemoService with version 0.0.0 not found, invocation rejected.
ApplicationModel.defaultModel().getModelEnvironment().getSystemConfiguration().overwriteCache(SERIALIZATION_SECURITY_CHECK_KEY, "false");
System.setProperty(SERIALIZATION_SECURITY_CHECK_KEY, "false");
}
@AfterAll
public static void teardown() {
FrameworkModel.defaultModel().destroy();
ApplicationModel.defaultModel().getModelEnvironment().getSystemConfiguration().clearCache();
System.clearProperty(SERIALIZATION_SECURITY_CHECK_KEY);
}
@ -100,26 +101,25 @@ class DubboTelnetDecodeTest {
try {
Codec2 codec = ExtensionLoader.getExtensionLoader(Codec2.class).getExtension("dubbo");
URL url = new ServiceConfigURL("dubbo", "localhost", 22226);
url = url.setScopeModel(ApplicationModel.defaultModel().newModule());
NettyCodecAdapter adapter = new NettyCodecAdapter(codec, url, new MockChannelHandler());
MockHandler mockHandler = new MockHandler(null,
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
if (checkDubboDecoded(msg)) {
dubbo.incrementAndGet();
}
return getDefaultFuture();
}
}))));
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
if (checkDubboDecoded(msg)) {
dubbo.incrementAndGet();
}
return getDefaultFuture();
}
}))));
ch = new LocalEmbeddedChannel();
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
ch.writeInbound(dubboByteBuf);
} catch (Exception e) {
@ -148,7 +148,6 @@ class DubboTelnetDecodeTest {
try {
Codec2 codec = ExtensionLoader.getExtensionLoader(Codec2.class).getExtension("dubbo");
URL url = new ServiceConfigURL("dubbo", "localhost", 22226);
url = url.setScopeModel(ApplicationModel.defaultModel().newModule());
NettyCodecAdapter adapter = new NettyCodecAdapter(codec, url, new MockChannelHandler());
MockHandler mockHandler = new MockHandler((msg) -> {
@ -156,19 +155,19 @@ class DubboTelnetDecodeTest {
telnet.incrementAndGet();
}
},
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
return getDefaultFuture();
}
}))));
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
return getDefaultFuture();
}
}))));
ch = new LocalEmbeddedChannel();
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
ch.writeInbound(telnetByteBuf);
} catch (Exception e) {
@ -193,13 +192,13 @@ class DubboTelnetDecodeTest {
* | telnet(incomplete) |
* +--------------------------------------------------+
* <p>
* <p>
*
* Second ByteBuf:
* +--------------------------++----------------------+
* | telnet(the remaining) || dubbo(complete) |
* +--------------------------++----------------------+
* ||
* Magic Code
* ||
* Magic Code
*
* @throws InterruptedException
*/
@ -212,7 +211,6 @@ class DubboTelnetDecodeTest {
try {
Codec2 codec = ExtensionLoader.getExtensionLoader(Codec2.class).getExtension("dubbo");
URL url = new ServiceConfigURL("dubbo", "localhost", 22226);
url = url.setScopeModel(ApplicationModel.defaultModel().newModule());
NettyCodecAdapter adapter = new NettyCodecAdapter(codec, url, new MockChannelHandler());
MockHandler mockHandler = new MockHandler((msg) -> {
@ -220,23 +218,23 @@ class DubboTelnetDecodeTest {
telnetDubbo.incrementAndGet();
}
},
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
if (checkDubboDecoded(msg)) {
telnetDubbo.incrementAndGet();
}
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
if (checkDubboDecoded(msg)) {
telnetDubbo.incrementAndGet();
}
return getDefaultFuture();
}
}))));
return getDefaultFuture();
}
}))));
ch = new LocalEmbeddedChannel();
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
ch.writeInbound(telnetByteBuf);
ch.writeInbound(Unpooled.wrappedBuffer(Unpooled.wrappedBuffer("\n".getBytes()), dubboByteBuf));
@ -267,7 +265,7 @@ class DubboTelnetDecodeTest {
* | telnet(incomplete) |
* +--------------------------------------------------+
* <p>
* <p>
*
* Second ByteBuf (secondByteBuf):
* +--------------------------------------------------+
* | telnet(the remaining) | telnet(complete) |
@ -285,7 +283,6 @@ class DubboTelnetDecodeTest {
try {
Codec2 codec = ExtensionLoader.getExtensionLoader(Codec2.class).getExtension("dubbo");
URL url = new ServiceConfigURL("dubbo", "localhost", 22226);
url = url.setScopeModel(ApplicationModel.defaultModel().newModule());
NettyCodecAdapter adapter = new NettyCodecAdapter(codec, url, new MockChannelHandler());
MockHandler mockHandler = new MockHandler((msg) -> {
@ -293,19 +290,19 @@ class DubboTelnetDecodeTest {
telnetTelnet.incrementAndGet();
}
},
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
return getDefaultFuture();
}
}))));
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
return getDefaultFuture();
}
}))));
ch = new LocalEmbeddedChannel();
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
ch.writeInbound(firstByteBuf);
ch.writeInbound(secondByteBuf);
@ -339,8 +336,8 @@ class DubboTelnetDecodeTest {
* +-------------------------++-----------------------+
* | dubbo(the remaining) || dubbo(complete) |
* +-------------------------++-----------------------+
* ||
* Magic Code
* ||
* Magic Code
*
* @throws InterruptedException
*/
@ -357,26 +354,25 @@ class DubboTelnetDecodeTest {
try {
Codec2 codec = ExtensionLoader.getExtensionLoader(Codec2.class).getExtension("dubbo");
URL url = new ServiceConfigURL("dubbo", "localhost", 22226);
url = url.setScopeModel(ApplicationModel.defaultModel().newModule());
NettyCodecAdapter adapter = new NettyCodecAdapter(codec, url, new MockChannelHandler());
MockHandler mockHandler = new MockHandler(null,
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
if (checkDubboDecoded(msg)) {
dubboDubbo.incrementAndGet();
}
return getDefaultFuture();
}
}))));
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
if (checkDubboDecoded(msg)) {
dubboDubbo.incrementAndGet();
}
return getDefaultFuture();
}
}))));
ch = new LocalEmbeddedChannel();
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
ch.writeInbound(firstDubboByteBuf);
ch.writeInbound(secondDubboByteBuf);
@ -425,7 +421,6 @@ class DubboTelnetDecodeTest {
try {
Codec2 codec = ExtensionLoader.getExtensionLoader(Codec2.class).getExtension("dubbo");
URL url = new ServiceConfigURL("dubbo", "localhost", 22226);
url = url.setScopeModel(ApplicationModel.defaultModel().newModule());
NettyCodecAdapter adapter = new NettyCodecAdapter(codec, url, new MockChannelHandler());
MockHandler mockHandler = new MockHandler((msg) -> {
@ -433,22 +428,22 @@ class DubboTelnetDecodeTest {
dubboTelnet.incrementAndGet();
}
},
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
if (checkDubboDecoded(msg)) {
dubboTelnet.incrementAndGet();
}
return getDefaultFuture();
}
}))));
new MultiMessageHandler(
new DecodeHandler(
new HeaderExchangeHandler(new ExchangeHandlerAdapter() {
@Override
public CompletableFuture<Object> reply(ExchangeChannel channel, Object msg) {
if (checkDubboDecoded(msg)) {
dubboTelnet.incrementAndGet();
}
return getDefaultFuture();
}
}))));
ch = new LocalEmbeddedChannel();
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
.addLast("decoder", adapter.getDecoder())
.addLast("handler", mockHandler);
ch.writeInbound(firstDubboByteBuf);
ch.writeInbound(secondByteBuf);
@ -501,13 +496,13 @@ class DubboTelnetDecodeTest {
if (msg instanceof DecodeableRpcInvocation) {
DecodeableRpcInvocation invocation = (DecodeableRpcInvocation) msg;
if ("sayHello".equals(invocation.getMethodName())
&& invocation.getParameterTypes().length == 1
&& String.class.equals(invocation.getParameterTypes()[0])
&& invocation.getArguments().length == 1
&& "dubbo".equals(invocation.getArguments()[0])
&& DemoService.class.getName().equals(invocation.getAttachment("path"))
&& DemoService.class.getName().equals(invocation.getAttachment("interface"))
&& "0.0.0".equals(invocation.getAttachment("version"))) {
&& invocation.getParameterTypes().length == 1
&& String.class.equals(invocation.getParameterTypes()[0])
&& invocation.getArguments().length == 1
&& "dubbo".equals(invocation.getArguments()[0])
&& DemoService.class.getName().equals(invocation.getAttachment("path"))
&& DemoService.class.getName().equals(invocation.getAttachment("interface"))
&& "0.0.0".equals(invocation.getAttachment("version"))) {
return true;
}
}

7
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java
View File

@ -19,6 +19,8 @@ package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.constants.CommonConstants;
import io.netty.handler.codec.http2.Http2Headers;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@ -61,6 +63,11 @@ public enum TripleHeaderEnum {
excludeAttachmentsSet.add(CommonConstants.APPLICATION_KEY);
excludeAttachmentsSet.add(TripleConstant.SERIALIZATION_KEY);
excludeAttachmentsSet.add(TripleConstant.TE_KEY);
for (Http2Headers.PseudoHeaderName value : Http2Headers.PseudoHeaderName.values()) {
excludeAttachmentsSet.add(value.value().toString());
}
}
private final String header;

36
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java
View File

@ -20,12 +20,12 @@ package org.apache.dubbo.rpc.protocol.tri.stream;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.JsonUtils;
import org.apache.dubbo.common.utils.LRU2Cache;
import org.apache.dubbo.rpc.TriRpcStatus;
import org.apache.dubbo.rpc.protocol.tri.TripleConstant;
import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.Http2Headers;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
@ -33,7 +33,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import java.util.stream.Collectors;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_UNSUPPORTED;
@ -44,6 +43,11 @@ public class StreamUtils {
private static final Base64.Decoder BASE64_DECODER = Base64.getDecoder();
private static final Base64.Encoder BASE64_ENCODER = Base64.getEncoder().withoutPadding();
private static final int MAX_LRU_HEADER_MAP_SIZE = 10000;
private static final Map<String, String> lruHeaderMap = new LRU2Cache<>(MAX_LRU_HEADER_MAP_SIZE);
public static String encodeBase64ASCII(byte[] in) {
byte[] bytes = encodeBase64(in);
return new String(bytes, StandardCharsets.US_ASCII);
@ -63,9 +67,6 @@ public class StreamUtils {
}
Map<String, Object> res = new HashMap<>(origin.size());
origin.forEach((k, v) -> {
if (Http2Headers.PseudoHeaderName.isPseudoHeader(k)) {
return;
}
if (TripleHeaderEnum.containsExcludeAttachments(k)) {
return;
}
@ -88,27 +89,26 @@ public class StreamUtils {
if (attachments == null) {
return;
}
Map<String, String> needConvertKey = new HashMap<>();
for (Map.Entry<String, Object> entry : attachments.entrySet()) {
final String key = entry.getKey().toLowerCase(Locale.ROOT);
if (Http2Headers.PseudoHeaderName.isPseudoHeader(key)) {
continue;
String key = lruHeaderMap.get(entry.getKey());
if (key == null) {
final String lowerCaseKey = entry.getKey().toLowerCase(Locale.ROOT);
lruHeaderMap.put(entry.getKey(), lowerCaseKey);
key = lowerCaseKey;
}
if (TripleHeaderEnum.containsExcludeAttachments(key)) {
continue;
}
if (needConvertHeaderKey && !key.equals(entry.getKey())) {
needConvertKey.put(key, entry.getKey());
}
final Object v = entry.getValue();
convertSingleAttachment(headers, key, v);
}
if (needConvertHeaderKey) {
Map<String, String> needConvertKey = attachments.entrySet()
.stream()
.filter(it -> !headers.contains(it.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, it -> it.getKey().toLowerCase(Locale.ROOT)));
if (!needConvertKey.isEmpty()) {
String needConvertJson = JsonUtils.getJson().toJson(needConvertKey);
headers.add(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader(), TriRpcStatus.encodeMessage(needConvertJson));
}
if (!needConvertKey.isEmpty()) {
String needConvertJson = JsonUtils.getJson().toJson(needConvertKey);
headers.add(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader(), TriRpcStatus.encodeMessage(needConvertJson));
}
}

81
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtilsTest.java
View File

@ -17,13 +17,25 @@
package org.apache.dubbo.rpc.protocol.tri.stream;
import org.apache.dubbo.common.utils.JsonUtils;
import org.apache.dubbo.rpc.TriRpcStatus;
import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class StreamUtilsTest {
@Test
void encodeBase64ASCII() {
String content = "😯";
@ -31,4 +43,73 @@ class StreamUtilsTest {
StreamUtils.decodeASCIIByte(StreamUtils.encodeBase64ASCII(content.getBytes(
StandardCharsets.UTF_8))));
}
@Test
void testConvertAttachment() throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
DefaultHttp2Headers headers = new DefaultHttp2Headers();
headers.add("key", "value");
Map<String, Object> attachments = new HashMap<>();
attachments.put(TripleHeaderEnum.PATH_KEY.getHeader(), "value");
attachments.put("key1111", "value");
attachments.put("Upper", "Upper");
attachments.put("obj", new Object());
StreamUtils.convertAttachment(headers, attachments, false);
Assertions.assertNull(headers.get(TripleHeaderEnum.PATH_KEY.getHeader()));
Assertions.assertNull(headers.get("Upper"));
Assertions.assertNull(headers.get("obj"));
headers = new DefaultHttp2Headers();
headers.add("key", "value");
StreamUtils.convertAttachment(headers, attachments, true);
Assertions.assertNull(headers.get(TripleHeaderEnum.PATH_KEY.getHeader()));
Assertions.assertNull(headers.get("Upper"));
Assertions.assertNull(headers.get("obj"));
String jsonRaw = headers.get(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader()).toString();
String json = TriRpcStatus.decodeMessage(jsonRaw);
System.out.println(jsonRaw + "---" + json);
Map<String, String> upperMap = JsonUtils.getJson().toJavaObject(json, Map.class);
Assertions.assertArrayEquals("Upper".getBytes(StandardCharsets.UTF_8), upperMap.get("upper").getBytes(StandardCharsets.UTF_8));
int count = 10000;
CountDownLatch latch = new CountDownLatch(count);
for (int i = 0; i < count; i++) {
String randomKey = "key" + i;
String randomValue = "value" + i;
Map<String, Object> attachments2 = new HashMap<>();
attachments2.put(TripleHeaderEnum.PATH_KEY.getHeader(), "value");
attachments2.put("key1111", "value");
attachments2.put("Upper", "Upper");
attachments2.put("obj", new Object());
attachments2.put(randomKey, randomValue);
executorService.execute(() -> {
DefaultHttp2Headers headers2 = new DefaultHttp2Headers();
headers2.add("key", "value");
StreamUtils.convertAttachment(headers2, attachments2, true);
if (headers2.get(TripleHeaderEnum.PATH_KEY.getHeader()) != null) {
return;
}
if (headers2.get("Upper") != null) {
return;
}
if (headers2.get("obj") != null) {
return;
}
if (!headers2.get(randomKey).toString().equals(randomValue)) {
return;
}
latch.countDown();
});
}
latch.await(10, TimeUnit.SECONDS);
Assertions.assertEquals(0, latch.getCount());
executorService.shutdown();
}
}

2
dubbo-serialization/dubbo-serialization-hessian2/src/main/java/org/apache/dubbo/common/serialize/hessian2/dubbo/DefaultHessian2FactoryInitializer.java
View File

@ -24,7 +24,7 @@ public class DefaultHessian2FactoryInitializer extends AbstractHessian2FactoryIn
@Override
protected SerializerFactory createSerializerFactory() {
Hessian2SerializerFactory hessian2SerializerFactory = new Hessian2SerializerFactory();
hessian2SerializerFactory.setAllowNonSerializable(Boolean.parseBoolean(ALLOW_NON_SERIALIZABLE));
hessian2SerializerFactory.setAllowNonSerializable(Boolean.parseBoolean(System.getProperty("dubbo.hessian.allowNonSerializable", "false")));
hessian2SerializerFactory.getClassFactory().allow("org.apache.dubbo.*");
return hessian2SerializerFactory;
}

2
dubbo-serialization/dubbo-serialization-hessian2/src/main/java/org/apache/dubbo/common/serialize/hessian2/dubbo/Hessian2FactoryInitializer.java
View File

@ -27,9 +27,7 @@ import com.alibaba.com.caucho.hessian.io.SerializerFactory;
@SPI(value = "default", scope = ExtensionScope.FRAMEWORK)
public interface Hessian2FactoryInitializer {
String ALLOW = System.getProperty("dubbo.application.hessian2.allow");
String DENY = System.getProperty("dubbo.application.hessian2.deny");
String WHITELIST = System.getProperty("dubbo.application.hessian2.whitelist");
String ALLOW_NON_SERIALIZABLE = System.getProperty("dubbo.hessian.allowNonSerializable", "false");