watchService;
+
+ /**
+ * Is Pooling Based Watch Service
+ *
+ * @see #detectPoolingBasedWatchService(Optional)
+ */
+ private static final boolean basedPoolingWatchService;
+
+ private static final WatchEvent.Modifier[] modifiers;
+
+ /**
+ * the delay to action in seconds. If null, execute indirectly
+ */
+ private static final Integer delay;
+
+ /**
+ * The thread pool for {@link WatchEvent WatchEvents} loop
+ * It's optional if there is not any {@link ConfigurationListener} registration
+ *
+ * @see ThreadPoolExecutor
+ */
+ private static final ThreadPoolExecutor watchEventsLoopThreadPool;
+
+ // static initialization
+ static {
+ watchService = newWatchService();
+ basedPoolingWatchService = detectPoolingBasedWatchService(watchService);
+ modifiers = initWatchEventModifiers();
+ delay = initDelay(modifiers);
+ watchEventsLoopThreadPool = newWatchEventsLoopThreadPool();
+ }
+
+ /**
+ * The Root Directory for config center
+ */
+ private final File rootDirectory;
+
+ private final String encoding;
+
+ /**
+ * The thread pool for workers who executes the tasks
+ */
+ private final ThreadPoolExecutor workersThreadPool;
+
+ /**
+ * The {@link Set} of {@link #configDirectory(String) directories} that may be processing,
+ *
+ * if {@link #isBasedPoolingWatchService()} is false, this properties will be
+ * {@link Collections#emptySet() empty}
+ *
+ * @see #initProcessingDirectories()
+ */
+ private final Set processingDirectories;
+
+ private final Map> listenersRepository;
+
+ public FileSystemDynamicConfiguration(URL url) {
+ this(initDirectory(url), getEncoding(url), getThreadPoolPrefixName(url), getThreadPoolSize(url));
+ }
+
+ public FileSystemDynamicConfiguration(File rootDirectory, String encoding,
+ String threadPoolPrefixName,
+ int threadPoolSize
+ ) {
+ this.rootDirectory = rootDirectory;
+ this.encoding = encoding;
+ this.workersThreadPool = initWorkersThreadPool(threadPoolPrefixName, threadPoolSize);
+ this.processingDirectories = initProcessingDirectories();
+ this.listenersRepository = new LinkedHashMap<>();
+ }
+
+ private Set initProcessingDirectories() {
+ return isBasedPoolingWatchService() ? new LinkedHashSet<>() : emptySet();
+ }
+
+ @Override
+ public void addListener(String key, String group, ConfigurationListener listener) {
+ doInListener(key, group, (configFilePath, listeners) -> {
+
+ if (listeners.isEmpty()) { // If no element, it indicates watchService was registered before
+ ThrowableConsumer.execute(configFilePath, configFile -> {
+ FileUtils.forceMkdirParent(configFile);
+ // A rootDirectory to be watched
+ File configDirectory = configFile.getParentFile();
+ if (configDirectory != null) {
+ // Register the configDirectory
+ configDirectory.toPath().register(watchService.get(), INTEREST_PATH_KINDS, modifiers);
+ }
+ });
+ }
+
+ // Add into cache
+ listeners.add(listener);
+ });
+ }
+
+ @Override
+ public void removeListener(String key, String group, ConfigurationListener listener) {
+ doInListener(key, group, (file, listeners) -> {
+ // Remove into cache
+ listeners.remove(listener);
+ });
+ }
+
+ protected File configDirectory(String group) {
+ String actualGroup = isBlank(group) ? DEFAULT_GROUP : group;
+ return new File(rootDirectory, actualGroup);
+ }
+
+ protected File configFile(String key, String group) {
+ return new File(configDirectory(group), key);
+ }
+
+ private void doInListener(String key, String group, BiConsumer> consumer) {
+ watchService.ifPresent(watchService -> {
+ File configFile = configFile(key, group);
+ executeMutually(configFile.getParentFile(), () -> {
+ // process the WatchEvents if not start
+ if (!isProcessingWatchEvents()) {
+ processWatchEvents(watchService);
+ }
+
+ List listeners = getListeners(configFile);
+ consumer.accept(configFile, listeners);
+
+ // Nothing to return
+ return null;
+ });
+ });
+ }
+
+ private static boolean isProcessingWatchEvents() {
+ return getWatchEventsLoopThreadPool().getActiveCount() > 0;
+ }
+
+ /**
+ * Process the {@link WatchEvent WatchEvents} loop in async execution
+ *
+ * @param watchService {@link WatchService}
+ */
+ private void processWatchEvents(WatchService watchService) {
+ getWatchEventsLoopThreadPool().execute(() -> { // WatchEvents Loop
+ while (true) {
+ WatchKey watchKey = null;
+ try {
+ watchKey = watchService.take();
+ if (watchKey.isValid()) {
+ for (WatchEvent event : watchKey.pollEvents()) {
+ WatchEvent.Kind kind = event.kind();
+ // configChangeType's key to match WatchEvent's Kind
+ ConfigChangeType configChangeType = CONFIG_CHANGE_TYPES_MAP.get(kind.name());
+ if (configChangeType != null) {
+ Path configDirectoryPath = (Path) watchKey.watchable();
+ Path currentPath = (Path) event.context();
+ Path configFilePath = configDirectoryPath.resolve(currentPath);
+ File configDirectory = configDirectoryPath.toFile();
+ executeMutually(configDirectory, () -> {
+ fireConfigChangeEvent(configFilePath.toFile(), configChangeType);
+ signalConfigDirectory(configDirectory);
+ return null;
+ });
+ }
+ }
+ }
+ } catch (Exception e) {
+ return;
+ } finally {
+ if (watchKey != null) {
+ // reset
+ watchKey.reset();
+ }
+ }
+ }
+ });
+ }
+
+ private void signalConfigDirectory(File configDirectory) {
+ if (isBasedPoolingWatchService()) {
+ // remove configDirectory from processing set because it's done
+ removeProcessingDirectory(configDirectory);
+ // notify configDirectory
+ notifyProcessingDirectory(configDirectory);
+ if (logger.isDebugEnabled()) {
+ logger.debug(format("The config rootDirectory[%s] is signalled...", configDirectory.getName()));
+ }
+ }
+ }
+
+ private void removeProcessingDirectory(File configDirectory) {
+ processingDirectories.remove(configDirectory);
+ }
+
+ private void notifyProcessingDirectory(File configDirectory) {
+ configDirectory.notifyAll();
+ }
+
+ private List getListeners(File configFile) {
+ return listenersRepository.computeIfAbsent(configFile, p -> new LinkedList<>());
+ }
+
+ private void fireConfigChangeEvent(File configFile, ConfigChangeType configChangeType) {
+ String key = configFile.getName();
+ String value = getConfig(configFile, -1L);
+ // fire ConfigChangeEvent one by one
+ getListeners(configFile).forEach(listener -> {
+ try {
+ listener.process(new ConfigChangeEvent(key, value, configChangeType));
+ } catch (Throwable e) {
+ if (logger.isErrorEnabled()) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+ });
+ }
+
+ @Override
+ public String getConfig(String key, String group, long timeout) throws IllegalStateException {
+ File configFile = configFile(key, group);
+ return getConfig(configFile, timeout);
+ }
+
+ protected String getConfig(File configFile, long timeout) {
+ return canRead(configFile) ? execute(() -> readFileToString(configFile, getEncoding()), timeout) : null;
+ }
+
+ private boolean canRead(File file) {
+ return file.exists() && file.canRead();
+ }
+
+ @Override
+ public String getConfigs(String key, String group, long timeout) throws IllegalStateException {
+ return getConfig(key, group, timeout);
+ }
+
+ @Override
+ public Object getInternalProperty(String key) {
+ return null;
+ }
+
+ @Override
+ public boolean publishConfig(String key, String group, String content) {
+ return delay(key, group, configFile -> {
+ FileUtils.write(configFile, content, getEncoding());
+ return true;
+ });
+ }
+
+ @Override
+ public String removeConfig(String key, String group) {
+ return delay(key, group, configFile -> {
+
+ String content = getConfig(configFile, -1L);
+
+ FileUtils.deleteQuietly(configFile);
+
+ return content;
+ });
+ }
+
+ private ThreadPoolExecutor initWorkersThreadPool(String prefix, int size) {
+ return (ThreadPoolExecutor) newFixedThreadPool(size, new NamedThreadFactory(prefix));
+ }
+
+ /**
+ * Delay action for {@link #configFile(String, String) config file}
+ *
+ * @param key the key to represent a configuration
+ * @param group the group where the key belongs to
+ * @param function the customized {@link Function function} with {@link File}
+ * @param the computed value
+ * @return
+ */
+ protected V delay(String key, String group, ThrowableFunction function) {
+ File configFile = configFile(key, group);
+ // Must be based on PoolingWatchService and has listeners under config file
+ if (isBasedPoolingWatchService()) {
+ File configDirectory = configFile.getParentFile();
+ executeMutually(configDirectory, () -> {
+ if (hasListeners(configFile) && isProcessing(configDirectory)) {
+ Integer delay = getDelay();
+ if (delay != null) {
+ // wait for delay in seconds
+ long timeout = SECONDS.toMillis(delay);
+ if (logger.isDebugEnabled()) {
+ logger.debug(format("The config[key : %s, group : %s] is about to delay in %d ms.",
+ key, group, timeout));
+ }
+ configDirectory.wait(timeout);
+ }
+ }
+ addProcessing(configDirectory);
+ return null;
+ });
+ }
+
+ V value = null;
+
+ try {
+ value = function.apply(configFile);
+ } catch (Throwable e) {
+ if (logger.isErrorEnabled()) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ return value;
+ }
+
+ private boolean hasListeners(File configFile) {
+ return getListeners(configFile).size() > 0;
+ }
+
+ /**
+ * Is processing on {@link #configDirectory(String) config rootDirectory}
+ *
+ * @param configDirectory {@link #configDirectory(String) config rootDirectory}
+ * @return if processing , return true, or false
+ */
+ private boolean isProcessing(File configDirectory) {
+ return processingDirectories.contains(configDirectory);
+ }
+
+ private void addProcessing(File configDirectory) {
+ processingDirectories.add(configDirectory);
+ }
+
+ @Override
+ public Set getConfigKeys(String group) {
+ return Stream.of(configDirectory(group).listFiles(File::isFile))
+ .map(File::getName)
+ .collect(Collectors.toSet());
+ }
+
+
+ @Override
+ public Set getConfigGroups() {
+ return Stream.of(getRootDirectory().listFiles())
+ .filter(File::isDirectory)
+ .map(File::getName)
+ .collect(Collectors.toSet());
+ }
+
+ @Override
+ public Map getConfigs(String group) throws UnsupportedOperationException {
+ return getConfigs(group, -1);
+ }
+
+ @Override
+ public void close() throws Exception {
+ // TODO
+ }
+
+ private V execute(Callable task, long timeout) {
+ V value = null;
+ try {
+
+ if (timeout < 1) { // less or equal 0
+ value = task.call();
+ } else {
+ Future future = workersThreadPool.submit(task);
+ value = future.get(timeout, TimeUnit.MILLISECONDS);
+ }
+ } catch (Exception e) {
+ if (logger.isErrorEnabled()) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+ return value;
+ }
+
+ protected File getRootDirectory() {
+ return rootDirectory;
+ }
+
+ protected ThreadPoolExecutor getWorkersThreadPool() {
+ return workersThreadPool;
+ }
+
+ protected String getEncoding() {
+ return encoding;
+ }
+
+ protected Integer getDelay() {
+ return delay;
+ }
+
+ /**
+ * It's whether the implementation of {@link WatchService} is based on {@linkplain sun.nio.fs.PollingWatchService}
+ * or not.
+ *
+ *
+ * @return if based, return true, or false
+ * @see #detectPoolingBasedWatchService(Optional)
+ */
+ protected static boolean isBasedPoolingWatchService() {
+ return basedPoolingWatchService;
+ }
+
+ private static String getThreadPoolPrefixName(URL url) {
+ return getParameter(url, THREAD_POOL_PREFIX_PARAM_NAME, DEFAULT_THREAD_POOL_PREFIX);
+ }
+
+ protected static ThreadPoolExecutor getWatchEventsLoopThreadPool() {
+ return watchEventsLoopThreadPool;
+ }
+
+ private V executeMutually(Object mutex, Callable callable) {
+ V value = null;
+ synchronized (mutex) {
+ try {
+ value = callable.call();
+ } catch (Exception e) {
+ if (logger.isErrorEnabled()) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+ }
+ return value;
+ }
+
+ private static T[] of(T... values) {
+ return values;
+ }
+
+ private static Integer initDelay(WatchEvent.Modifier[] modifiers) {
+ return Stream.of(modifiers)
+ .filter(modifier -> modifier instanceof SensitivityWatchEventModifier)
+ .map(SensitivityWatchEventModifier.class::cast)
+ .map(SensitivityWatchEventModifier::sensitivityValueInSeconds)
+ .max(Integer::compareTo)
+ .orElse(null);
+ }
+
+ private static WatchEvent.Modifier[] initWatchEventModifiers() {
+ if (isBasedPoolingWatchService()) { // If based on PollingWatchService, High sensitivity will be used
+ return of(SensitivityWatchEventModifier.HIGH);
+ } else {
+ return of();
+ }
+ }
+
+ /**
+ * Detect the argument of {@link WatchService} is based on {@linkplain sun.nio.fs.PollingWatchService}
+ * or not.
+ *
+ * Some platforms do not provide the native implementation of {@link WatchService}, just use
+ * {@linkplain sun.nio.fs.PollingWatchService} in periodic poll file modifications.
+ *
+ * @param watchService the instance of {@link WatchService}
+ * @return if based, return true, or false
+ */
+ private static boolean detectPoolingBasedWatchService(Optional watchService) {
+ String className = watchService.map(Object::getClass).map(Class::getName).orElse(null);
+ return POLLING_WATCH_SERVICE_CLASS_NAME.equals(className);
+ }
+
+ private static Optional newWatchService() {
+ Optional watchService = null;
+ FileSystem fileSystem = FileSystems.getDefault();
+ try {
+ watchService = Optional.of(fileSystem.newWatchService());
+ } catch (IOException e) {
+ if (logger.isErrorEnabled()) {
+ logger.error(e.getMessage(), e);
+ }
+ watchService = Optional.empty();
+ }
+ return watchService;
+ }
+
+ private static File initDirectory(URL url) {
+ String directoryPath = getParameter(url, CONFIG_CENTER_DIR_PARAM_NAME, DEFAULT_CONFIG_CENTER_DIR_PATH);
+ File rootDirectory = new File(getParameter(url, CONFIG_CENTER_DIR_PARAM_NAME, DEFAULT_CONFIG_CENTER_DIR_PATH));
+ if (!rootDirectory.exists() && !rootDirectory.mkdirs()) {
+ throw new IllegalStateException(format("Dubbo config center rootDirectory[%s] can't be created!",
+ directoryPath));
+ }
+ return rootDirectory;
+ }
+
+ private static String getParameter(URL url, String name, String defaultValue) {
+ if (url != null) {
+ return url.getParameter(name, defaultValue);
+ }
+ return defaultValue;
+ }
+
+ private static String getEncoding(URL url) {
+ return getParameter(url, CONFIG_CENTER_ENCODING_PARAM_NAME, DEFAULT_CONFIG_CENTER_ENCODING);
+ }
+
+ private static int getThreadPoolSize(URL url) {
+ return Integer.parseInt(getParameter(url, THREAD_POOL_SIZE_PARAM_NAME, DEFAULT_THREAD_POOL_SIZE));
+ }
+
+ private static ThreadPoolExecutor newWatchEventsLoopThreadPool() {
+ return new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE,
+ 0L, MILLISECONDS,
+ new SynchronousQueue(),
+ new NamedThreadFactory("dubbo-config-center-watch-events-loop", true));
+ }
+}
\ No newline at end of file
diff --git a/dubbo-bootstrap/src/main/java/org/apache/dubbo/bootstrap/AbstractSettings.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfigurationFactory.java
similarity index 57%
rename from dubbo-bootstrap/src/main/java/org/apache/dubbo/bootstrap/AbstractSettings.java
rename to dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfigurationFactory.java
index 7ed1de5498..f2a13322c6 100644
--- a/dubbo-bootstrap/src/main/java/org/apache/dubbo/bootstrap/AbstractSettings.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfigurationFactory.java
@@ -14,23 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dubbo.bootstrap;
+package org.apache.dubbo.common.config.configcenter.file;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.configcenter.AbstractDynamicConfigurationFactory;
+import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
+import org.apache.dubbo.common.config.configcenter.DynamicConfigurationFactory;
/**
- * Abstract {@link Settings}
+ * File-System based {@link DynamicConfigurationFactory} implementation
*
* @since 2.7.4
*/
-public class AbstractSettings implements Settings {
-
- private final DubboBootstrap dubboBootstrap;
-
- public AbstractSettings(DubboBootstrap dubboBootstrap) {
- this.dubboBootstrap = dubboBootstrap;
- }
+public class FileSystemDynamicConfigurationFactory extends AbstractDynamicConfigurationFactory {
@Override
- public DubboBootstrap next() {
- return dubboBootstrap;
+ protected DynamicConfiguration createDynamicConfiguration(URL url) {
+ return new FileSystemDynamicConfiguration(url);
}
}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfiguration.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfiguration.java
index 14accb28a2..a69f115ab7 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfiguration.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfiguration.java
@@ -28,6 +28,7 @@ import static java.util.Collections.emptySortedSet;
* The default extension of {@link DynamicConfiguration}. If user does not specify a config centre, or specifies one
* that is not a valid extension, it will default to this one.
*/
+@Deprecated
public class NopDynamicConfiguration implements DynamicConfiguration {
public NopDynamicConfiguration(URL url) {
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfigurationFactory.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfigurationFactory.java
index bd45e7f736..487f70da6b 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfigurationFactory.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/nop/NopDynamicConfigurationFactory.java
@@ -23,6 +23,7 @@ import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
/**
*
*/
+@Deprecated
public class NopDynamicConfigurationFactory extends AbstractDynamicConfigurationFactory {
@Override
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/wrapper/CompositeDynamicConfiguration.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/wrapper/CompositeDynamicConfiguration.java
index 7f98d383d7..016265df87 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/wrapper/CompositeDynamicConfiguration.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/wrapper/CompositeDynamicConfiguration.java
@@ -89,9 +89,6 @@ public class CompositeDynamicConfiguration implements DynamicConfiguration {
Object value = null;
for (DynamicConfiguration configuration : configurations) {
value = func.apply(configuration);
- if (value != null) {
- break;
- }
}
return value;
}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/utils/ReflectUtils.java b/dubbo-common/src/main/java/org/apache/dubbo/common/utils/ReflectUtils.java
index 27202299d2..f8e987c0a4 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/utils/ReflectUtils.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/utils/ReflectUtils.java
@@ -21,6 +21,9 @@ import javassist.CtConstructor;
import javassist.CtMethod;
import javassist.NotFoundException;
+import java.beans.BeanInfo;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
import java.lang.reflect.Array;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
@@ -48,6 +51,7 @@ import java.util.concurrent.Future;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static java.util.Arrays.asList;
import static java.util.Collections.unmodifiableSet;
@@ -1127,7 +1131,7 @@ public final class ReflectUtils {
}
return new Type[]{returnType, genericReturnType};
}
-
+
/**
* Find the {@link Set} of {@link ParameterizedType}
*
@@ -1184,4 +1188,28 @@ public final class ReflectUtils {
return unmodifiableSet(hierarchicalTypes);
}
+
+ public static T getProperty(Object bean, String propertyName) {
+ Class beanClass = bean.getClass();
+ BeanInfo beanInfo = null;
+ T propertyValue = null;
+ try {
+ beanInfo = Introspector.getBeanInfo(beanClass);
+ propertyValue = (T) Stream.of(beanInfo.getPropertyDescriptors())
+ .filter(propertyDescriptor -> propertyName.equals(propertyDescriptor.getName()))
+ .map(PropertyDescriptor::getReadMethod)
+ .findFirst()
+ .map(method -> {
+ try {
+ return method.invoke(bean);
+ } catch (Exception e) {
+ }
+ return null;
+ }).get();
+ } catch (Exception e) {
+
+ }
+ return propertyValue;
+ }
+
}
\ No newline at end of file
diff --git a/dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.config.configcenter.DynamicConfigurationFactory b/dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.config.configcenter.DynamicConfigurationFactory
index 42d6a25112..f6fe6d68fc 100644
--- a/dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.config.configcenter.DynamicConfigurationFactory
+++ b/dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.config.configcenter.DynamicConfigurationFactory
@@ -1 +1,2 @@
-nop=org.apache.dubbo.common.config.configcenter.nop.NopDynamicConfigurationFactory
\ No newline at end of file
+nop=org.apache.dubbo.common.config.configcenter.nop.NopDynamicConfigurationFactory
+file=org.apache.dubbo.common.config.configcenter.file.FileSystemDynamicConfigurationFactory
\ No newline at end of file
diff --git a/dubbo-bootstrap/src/main/java/org/apache/dubbo/bootstrap/Settings.java b/dubbo-common/src/test/java/org/apache/dubbo/common/config/configcenter/DynamicConfigurationFactoryTest.java
similarity index 51%
rename from dubbo-bootstrap/src/main/java/org/apache/dubbo/bootstrap/Settings.java
rename to dubbo-common/src/test/java/org/apache/dubbo/common/config/configcenter/DynamicConfigurationFactoryTest.java
index 290bb7e4b9..eacd3eed1e 100644
--- a/dubbo-bootstrap/src/main/java/org/apache/dubbo/bootstrap/Settings.java
+++ b/dubbo-common/src/test/java/org/apache/dubbo/common/config/configcenter/DynamicConfigurationFactoryTest.java
@@ -14,19 +14,26 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dubbo.bootstrap;
+package org.apache.dubbo.common.config.configcenter;
+
+import org.apache.dubbo.common.config.configcenter.file.FileSystemDynamicConfigurationFactory;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.dubbo.common.extension.ExtensionLoader.getExtensionLoader;
+import static org.junit.jupiter.api.Assertions.assertEquals;
/**
- * The Dubbo settings
+ * {@link DynamicConfigurationFactory} Test
*
* @since 2.7.4
*/
-public interface Settings {
+public class DynamicConfigurationFactoryTest {
- /**
- * Go next settings
- *
- * @return {@link DubboBootstrap}
- */
- DubboBootstrap next();
+ @Test
+ public void testDefaultExtension() {
+ DynamicConfigurationFactory factory = getExtensionLoader(DynamicConfigurationFactory.class).getDefaultExtension();
+ assertEquals(FileSystemDynamicConfigurationFactory.class, factory.getClass());
+ assertEquals(factory, getExtensionLoader(DynamicConfigurationFactory.class).getExtension("file"));
+ }
}
diff --git a/dubbo-common/src/test/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfigurationTest.java b/dubbo-common/src/test/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfigurationTest.java
new file mode 100644
index 0000000000..c68cd4c6d2
--- /dev/null
+++ b/dubbo-common/src/test/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfigurationTest.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.common.config.configcenter.file;
+
+import org.apache.dubbo.common.URL;
+
+import org.apache.commons.io.FileUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static java.util.Collections.singleton;
+import static org.apache.commons.io.FileUtils.deleteQuietly;
+import static org.apache.dubbo.common.URL.valueOf;
+import static org.apache.dubbo.common.config.configcenter.DynamicConfiguration.DEFAULT_GROUP;
+import static org.apache.dubbo.common.config.configcenter.file.FileSystemDynamicConfiguration.CONFIG_CENTER_DIR_PARAM_NAME;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * {@link FileSystemDynamicConfiguration} Test
+ */
+public class FileSystemDynamicConfigurationTest {
+
+ private FileSystemDynamicConfiguration configuration;
+
+ private static final String KEY = "abc-def-ghi";
+
+ private static final String CONTENT = "Hello,World";
+
+ @BeforeEach
+ public void init() {
+ String classPath = getClassPath();
+ URL url = valueOf("dubbo://127.0.0.1:20880").addParameter(CONFIG_CENTER_DIR_PARAM_NAME, classPath + File.separator + "config-center");
+ configuration = new FileSystemDynamicConfiguration(url);
+ deleteQuietly(configuration.getRootDirectory());
+ }
+
+ private String getClassPath() {
+ return getClass().getProtectionDomain().getCodeSource().getLocation().getPath();
+ }
+
+ @Test
+ public void testInit() {
+
+ assertEquals(new File(getClassPath(), "config-center"), configuration.getRootDirectory());
+ assertEquals("UTF-8", configuration.getEncoding());
+ assertEquals(ThreadPoolExecutor.class, configuration.getWorkersThreadPool().getClass());
+ assertEquals(1, (configuration.getWorkersThreadPool()).getCorePoolSize());
+ assertEquals(1, (configuration.getWorkersThreadPool()).getMaximumPoolSize());
+ assertNotNull(configuration.getWatchEventsLoopThreadPool());
+ assertEquals(1, (configuration.getWatchEventsLoopThreadPool()).getCorePoolSize());
+ assertEquals(1, (configuration.getWatchEventsLoopThreadPool()).getMaximumPoolSize());
+
+ if (configuration.isBasedPoolingWatchService()) {
+ assertEquals(2, configuration.getDelay());
+ } else {
+ assertNull(configuration.getDelay());
+ }
+ }
+
+ @Test
+ public void testPublishAndGetConfig() {
+ assertTrue(configuration.publishConfig(KEY, CONTENT));
+ assertTrue(configuration.publishConfig(KEY, CONTENT));
+ assertTrue(configuration.publishConfig(KEY, CONTENT));
+ assertEquals(CONTENT, configuration.getConfig(KEY));
+ assertTrue(configuration.getConfigs(null).size() > 0);
+ }
+
+ @Test
+ public void testPublishAndRemoveConfig() throws InterruptedException {
+ assertTrue(configuration.publishConfig(KEY, CONTENT));
+ configuration.addListener(KEY, event -> {
+ System.out.printf("[%s] " + event + "\n", Thread.currentThread().getName());
+
+ });
+ assertTrue(configuration.publishConfig(KEY, CONTENT));
+ assertEquals(CONTENT, configuration.removeConfig(KEY));
+ Thread.sleep(configuration.getDelay() * 1000);
+ }
+
+ @Test
+ public void testGetConfigsAndGroups() {
+ assertTrue(configuration.publishConfig(KEY, CONTENT));
+ assertEquals(singleton(KEY), configuration.getConfigKeys(DEFAULT_GROUP));
+ assertEquals(singleton(DEFAULT_GROUP), configuration.getConfigGroups());
+
+ assertTrue(configuration.publishConfig(KEY, "test", CONTENT));
+ assertEquals(singleton(KEY), configuration.getConfigKeys(DEFAULT_GROUP));
+ assertTrue(configuration.getConfigGroups().contains(DEFAULT_GROUP));
+ assertTrue(configuration.getConfigGroups().contains("test"));
+ }
+
+ @Test
+ public void testAddAndRemoveListener() throws InterruptedException {
+
+ configuration.publishConfig(KEY, "A");
+
+ AtomicBoolean processedEvent = new AtomicBoolean();
+
+ configuration.addListener(KEY, event -> {
+
+ processedEvent.set(true);
+ assertEquals(KEY, event.getKey());
+ System.out.printf("[%s] " + event + "\n", Thread.currentThread().getName());
+ });
+
+
+ configuration.publishConfig(KEY, "B");
+ while (!processedEvent.get()) {
+ Thread.sleep(1 * 1000L);
+ }
+
+ processedEvent.set(false);
+ configuration.publishConfig(KEY, "C");
+ while (!processedEvent.get()) {
+ Thread.sleep(1 * 1000L);
+ }
+
+ processedEvent.set(false);
+ configuration.publishConfig(KEY, "D");
+ while (!processedEvent.get()) {
+ Thread.sleep(1 * 1000L);
+ }
+
+ configuration.addListener("test", "test", event -> {
+ processedEvent.set(true);
+ assertEquals("test", event.getKey());
+ System.out.printf("[%s] " + event + "\n", Thread.currentThread().getName());
+ });
+ processedEvent.set(false);
+ configuration.publishConfig("test", "test", "TEST");
+ while (!processedEvent.get()) {
+ Thread.sleep(1 * 1000L);
+ }
+
+ configuration.publishConfig("test", "test", "TEST");
+ configuration.publishConfig("test", "test", "TEST");
+ configuration.publishConfig("test", "test", "TEST");
+
+
+ processedEvent.set(false);
+ File keyFile = configuration.configFile(KEY, DEFAULT_GROUP);
+ FileUtils.deleteQuietly(keyFile);
+ while (!processedEvent.get()) {
+ Thread.sleep(1 * 1000L);
+ }
+ }
+}
diff --git a/dubbo-common/src/test/resources/log4j.xml b/dubbo-common/src/test/resources/log4j.xml
index bfb523c0cc..21ea447138 100644
--- a/dubbo-common/src/test/resources/log4j.xml
+++ b/dubbo-common/src/test/resources/log4j.xml
@@ -21,14 +21,14 @@
-
+
-
+
\ No newline at end of file
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/AbstractInterfaceConfig.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/AbstractInterfaceConfig.java
index 1ad15ecb72..874e38b968 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/AbstractInterfaceConfig.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/AbstractInterfaceConfig.java
@@ -255,7 +255,6 @@ public abstract class AbstractInterfaceConfig extends AbstractMethodConfig {
}
/**
- *
* Load the registry and conversion it to {@link URL}, the priority order is: system property > dubbo registry config
*
* @param provider whether it is the provider side
@@ -298,7 +297,6 @@ public abstract class AbstractInterfaceConfig extends AbstractMethodConfig {
}
/**
- *
* Load the monitor config from the system properties and conversation it to {@link URL}
*
* @param registryURL
@@ -358,7 +356,7 @@ public abstract class AbstractInterfaceConfig extends AbstractMethodConfig {
* methods configured in the configuration file are included in the interface of remote service
*
* @param interfaceClass the interface of remote service
- * @param methods the methods configured
+ * @param methods the methods configured
*/
protected void checkInterfaceAndMethods(Class interfaceClass, List methods) {
// interface cannot be null
@@ -466,15 +464,14 @@ public abstract class AbstractInterfaceConfig extends AbstractMethodConfig {
private void convertRegistryIdsToRegistries() {
if (StringUtils.isEmpty(registryIds)) {
if (CollectionUtils.isEmpty(registries)) {
- setRegistries(
- ConfigManager.getInstance().getDefaultRegistries()
- .filter(CollectionUtils::isNotEmpty)
- .orElseGet(() -> {
- RegistryConfig registryConfig = new RegistryConfig();
- registryConfig.refresh();
- return Arrays.asList(registryConfig);
- })
- );
+ List registryConfigs = ConfigManager.getInstance().getDefaultRegistries();
+ if (registryConfigs.isEmpty()) {
+ registryConfigs = new ArrayList<>();
+ RegistryConfig registryConfig = new RegistryConfig();
+ registryConfig.refresh();
+ registryConfigs.add(registryConfig);
+ }
+ setRegistries(registryConfigs);
}
} else {
String[] ids = COMMA_SPLIT_PATTERN.split(registryIds);
@@ -668,7 +665,7 @@ public abstract class AbstractInterfaceConfig extends AbstractMethodConfig {
@SuppressWarnings({"unchecked"})
public void setRegistries(List registries) {
- ConfigManager.getInstance().addRegistries((List) registries, false);
+ ConfigManager.getInstance().addRegistries((List) registries);
this.registries = (List) registries;
}
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java
index c9a96cb2a5..a5196df614 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java
@@ -16,6 +16,7 @@
*/
package org.apache.dubbo.config;
+import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.support.Parameter;
import org.apache.dubbo.remoting.Constants;
@@ -179,9 +180,14 @@ public class RegistryConfig extends AbstractConfig {
public void setAddress(String address) {
this.address = address;
if (address != null) {
- int i = address.indexOf("://");
- if (i > 0) {
- this.updateIdIfAbsent(address.substring(0, i));
+ try {
+ URL url = URL.valueOf(address);
+ setUsername(url.getUsername());
+ setPassword(url.getPassword());
+ setProtocol(url.getProtocol());
+ setPort(url.getPort());
+ setParameters(url.getParameters());
+ } catch (Exception ignored) {
}
}
}
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
index fb9d247d30..c851ca133e 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
@@ -882,15 +882,14 @@ public class ServiceConfig extends AbstractServiceConfig {
private void convertProtocolIdsToProtocols() {
if (StringUtils.isEmpty(protocolIds)) {
if (CollectionUtils.isEmpty(protocols)) {
- setProtocols(
- ConfigManager.getInstance().getDefaultProtocols()
- .filter(CollectionUtils::isNotEmpty)
- .orElseGet(() -> {
- ProtocolConfig protocolConfig = new ProtocolConfig();
- protocolConfig.refresh();
- return new ArrayList<>(Arrays.asList(protocolConfig));
- })
- );
+ List protocolConfigs = ConfigManager.getInstance().getDefaultProtocols();
+ if (protocolConfigs.isEmpty()) {
+ protocolConfigs = new ArrayList<>(1);
+ ProtocolConfig protocolConfig = new ProtocolConfig();
+ protocolConfig.refresh();
+ protocolConfigs.add(protocolConfig);
+ }
+ setProtocols(protocolConfigs);
}
} else {
String[] arr = COMMA_SPLIT_PATTERN.split(protocolIds);
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/builders/AbstractBuilder.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/builders/AbstractBuilder.java
index 12d09bf05c..1f51f1bc69 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/builders/AbstractBuilder.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/builders/AbstractBuilder.java
@@ -36,7 +36,7 @@ public abstract class AbstractBuilder protocols = new ConcurrentHashMap<>();
- private Map registries = new ConcurrentHashMap<>();
- private Map