You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shenyu.apache.org by xi...@apache.org on 2021/07/15 09:43:21 UTC
[incubator-shenyu] branch master updated: [ISSUE #1538] data sync
support consul (#1766)
This is an automated email from the ASF dual-hosted git repository.
xiaoyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new d9c7dc1 [ISSUE #1538] data sync support consul (#1766)
d9c7dc1 is described below
commit d9c7dc1c96d43850a82fcadb5cc59d1b26b4f6d2
Author: ttttangzhen <28...@qq.com>
AuthorDate: Thu Jul 15 17:43:14 2021 +0800
[ISSUE #1538] data sync support consul (#1766)
* [ISSUE #1538] data sync support consul
* [ISSUE #1538] fix test consul sync failed
* [ISSUE #1538] fix consul-api multi version
---
pom.xml | 1 +
shenyu-admin/pom.xml | 6 +
.../shenyu/admin/config/DataSyncConfiguration.java | 48 ++++
.../admin/config/properties/ConsulProperties.java | 32 +++
.../listener/consul/ConsulDataChangedListener.java | 304 +++++++++++++++++++++
.../admin/listener/consul/ConsulDataInit.java | 59 ++++
shenyu-admin/src/main/resources/application.yml | 2 +
.../consul/ConsulDataChangedListenerTest.java | 208 ++++++++++++++
.../admin/listener/consul/ConsulDataInitTest.java | 64 +++++
shenyu-bootstrap/pom.xml | 7 +
.../src/main/resources/application-local.yml | 4 +
.../shenyu/common/constant/ConsulConstants.java | 61 +++++
.../pom.xml | 1 +
.../pom.xml | 29 +-
.../data/consul/ConsulSyncDataConfiguration.java | 89 ++++++
.../src/main/resources/META-INF/spring.factories | 19 ++
.../src/main/resources/META-INF/spring.provides | 18 ++
shenyu-sync-data-center/pom.xml | 1 +
.../shenyu-sync-data-consul}/pom.xml | 29 +-
.../sync/data/consul/ConsulSyncDataService.java | 155 +++++++++++
.../sync/data/consul/config/ConsulConfig.java | 34 +++
.../data/consul/handler/ConsulCacheHandler.java | 124 +++++++++
.../data/consul/ConsulSyncDataServiceTest.java | 93 +++++++
.../consul/handler/ConsulCacheHandlerTest.java | 215 +++++++++++++++
24 files changed, 1582 insertions(+), 21 deletions(-)
diff --git a/pom.xml b/pom.xml
index 22f64b7..fdffaf1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -81,6 +81,7 @@
<commons-collections4.version>4.1</commons-collections4.version>
<reactor-spring.version>1.0.1.RELEASE</reactor-spring.version>
<disruptor.version>3.4.0</disruptor.version>
+ <consul.api.version>1.4.5</consul.api.version>
<hessian.version>4.0.38</hessian.version>
<protostuff.version>1.0.8</protostuff.version>
<kryo.version>4.0.0</kryo.version>
diff --git a/shenyu-admin/pom.xml b/shenyu-admin/pom.xml
index 504a583..b0eedc0 100644
--- a/shenyu-admin/pom.xml
+++ b/shenyu-admin/pom.xml
@@ -165,6 +165,12 @@
</dependency>
<dependency>
+ <groupId>com.ecwid.consul</groupId>
+ <artifactId>consul-api</artifactId>
+ <version>${consul.api.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-ldap</artifactId>
</dependency>
diff --git a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/DataSyncConfiguration.java b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/DataSyncConfiguration.java
index a922c70..12236a6 100644
--- a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/DataSyncConfiguration.java
+++ b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/DataSyncConfiguration.java
@@ -18,12 +18,16 @@
package org.apache.shenyu.admin.config;
import com.alibaba.nacos.api.config.ConfigService;
+import com.ecwid.consul.v1.ConsulClient;
import io.etcd.jetcd.Client;
import org.I0Itec.zkclient.ZkClient;
+import org.apache.shenyu.admin.config.properties.ConsulProperties;
import org.apache.shenyu.admin.config.properties.EtcdProperties;
import org.apache.shenyu.admin.config.properties.HttpSyncProperties;
import org.apache.shenyu.admin.config.properties.WebsocketSyncProperties;
import org.apache.shenyu.admin.listener.DataChangedListener;
+import org.apache.shenyu.admin.listener.consul.ConsulDataChangedListener;
+import org.apache.shenyu.admin.listener.consul.ConsulDataInit;
import org.apache.shenyu.admin.listener.etcd.EtcdClient;
import org.apache.shenyu.admin.listener.etcd.EtcdDataDataChangedListener;
import org.apache.shenyu.admin.listener.etcd.EtcdDataInit;
@@ -215,5 +219,49 @@ public class DataSyncConfiguration {
return new EtcdDataInit(etcdClient, syncDataService);
}
}
+
+ /**
+ * The type Consul listener.
+ */
+ @Configuration
+ @ConditionalOnProperty(prefix = "shenyu.sync.consul", name = "url")
+ @EnableConfigurationProperties(ConsulProperties.class)
+ static class ConsulListener {
+
+ /**
+ * init Consul client.
+ * @param consulProperties the consul properties
+ * @return Consul client
+ */
+ @Bean
+ public ConsulClient consulClient(final ConsulProperties consulProperties) {
+ return new ConsulClient(consulProperties.getUrl());
+ }
+
+ /**
+ * Config event listener data changed listener.
+ *
+ * @param consulClient the consul client
+ * @return the data changed listener
+ */
+ @Bean
+ @ConditionalOnMissingBean(ConsulDataChangedListener.class)
+ public DataChangedListener consulDataChangedListener(final ConsulClient consulClient) {
+ return new ConsulDataChangedListener(consulClient);
+ }
+
+ /**
+ * Consul data init.
+ *
+ * @param consulClient the consul client
+ * @param syncDataService the sync data service
+ * @return the consul data init
+ */
+ @Bean
+ @ConditionalOnMissingBean(ConsulDataInit.class)
+ public ConsulDataInit consulDataInit(final ConsulClient consulClient, final SyncDataService syncDataService) {
+ return new ConsulDataInit(consulClient, syncDataService);
+ }
+ }
}
diff --git a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/ConsulProperties.java b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/ConsulProperties.java
new file mode 100644
index 0000000..f1123d3
--- /dev/null
+++ b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/ConsulProperties.java
@@ -0,0 +1,32 @@
+/*
+ * 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.shenyu.admin.config.properties;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Consul config.
+ */
+@Data
+@ConfigurationProperties(prefix = "shenyu.sync.consul")
+public class ConsulProperties {
+
+ private String url;
+
+}
diff --git a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/consul/ConsulDataChangedListener.java b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/consul/ConsulDataChangedListener.java
new file mode 100644
index 0000000..15641c6
--- /dev/null
+++ b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/consul/ConsulDataChangedListener.java
@@ -0,0 +1,304 @@
+/*
+ * 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.shenyu.admin.listener.consul;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.kv.model.GetValue;
+import com.google.common.collect.Maps;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shenyu.admin.listener.DataChangedListener;
+import org.apache.shenyu.common.constant.ConsulConstants;
+import org.apache.shenyu.common.dto.AppAuthData;
+import org.apache.shenyu.common.dto.MetaData;
+import org.apache.shenyu.common.dto.PluginData;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.enums.DataEventTypeEnum;
+import org.apache.shenyu.common.utils.GsonUtils;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.stream.Collectors;
+
+/**
+ * Use Consul to push data changes.
+ */
+@Slf4j
+public class ConsulDataChangedListener implements DataChangedListener {
+
+ private static final ConcurrentMap<String, PluginData> PLUGIN_MAP = Maps.newConcurrentMap();
+
+ private static final ConcurrentMap<String, List<SelectorData>> SELECTOR_MAP = Maps.newConcurrentMap();
+
+ private static final ConcurrentMap<String, List<RuleData>> RULE_MAP = Maps.newConcurrentMap();
+
+ private static final ConcurrentMap<String, AppAuthData> AUTH_MAP = Maps.newConcurrentMap();
+
+ private static final ConcurrentMap<String, MetaData> META_DATA = Maps.newConcurrentMap();
+
+ private static final Comparator<SelectorData> SELECTOR_DATA_COMPARATOR = Comparator.comparing(SelectorData::getSort);
+
+ private static final Comparator<RuleData> RULE_DATA_COMPARATOR = Comparator.comparing(RuleData::getSort);
+
+ private final ConsulClient consulClient;
+
+ public ConsulDataChangedListener(final ConsulClient consulClient) {
+ this.consulClient = consulClient;
+ }
+
+ @Override
+ public void onAppAuthChanged(final List<AppAuthData> changed, final DataEventTypeEnum eventType) {
+ updateAuthMap(getConfig(ConsulConstants.AUTH_DATA));
+ switch (eventType) {
+ case DELETE:
+ changed.forEach(appAuth -> AUTH_MAP.remove(appAuth.getAppKey()));
+ break;
+ case REFRESH:
+ case MYSELF:
+ Set<String> set = new HashSet<>(AUTH_MAP.keySet());
+ changed.forEach(appAuth -> {
+ set.remove(appAuth.getAppKey());
+ AUTH_MAP.put(appAuth.getAppKey(), appAuth);
+ });
+ AUTH_MAP.keySet().removeAll(set);
+ break;
+ default:
+ changed.forEach(appAuth -> AUTH_MAP.put(appAuth.getAppKey(), appAuth));
+ break;
+ }
+ publishConfig(ConsulConstants.AUTH_DATA, AUTH_MAP);
+ }
+
+ @Override
+ public void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {
+ updatePluginMap(getConfig(ConsulConstants.PLUGIN_DATA));
+ switch (eventType) {
+ case DELETE:
+ changed.forEach(plugin -> PLUGIN_MAP.remove(plugin.getName()));
+ break;
+ case REFRESH:
+ case MYSELF:
+ Set<String> set = new HashSet<>(PLUGIN_MAP.keySet());
+ changed.forEach(plugin -> {
+ set.remove(plugin.getName());
+ PLUGIN_MAP.put(plugin.getName(), plugin);
+ });
+ PLUGIN_MAP.keySet().removeAll(set);
+ break;
+ default:
+ changed.forEach(plugin -> PLUGIN_MAP.put(plugin.getName(), plugin));
+ break;
+ }
+ publishConfig(ConsulConstants.PLUGIN_DATA, PLUGIN_MAP);
+ }
+
+ @Override
+ public void onSelectorChanged(final List<SelectorData> changed, final DataEventTypeEnum eventType) {
+ updateSelectorMap(getConfig(ConsulConstants.SELECTOR_DATA));
+ switch (eventType) {
+ case DELETE:
+ changed.forEach(selector -> {
+ List<SelectorData> ls = SELECTOR_MAP
+ .getOrDefault(selector.getPluginName(), new ArrayList<>())
+ .stream()
+ .filter(s -> !s.getId().equals(selector.getId()))
+ .sorted(SELECTOR_DATA_COMPARATOR)
+ .collect(Collectors.toList());
+ SELECTOR_MAP.put(selector.getPluginName(), ls);
+ });
+ break;
+ case REFRESH:
+ case MYSELF:
+ SELECTOR_MAP.keySet().removeAll(SELECTOR_MAP.keySet());
+ changed.forEach(selector -> {
+ List<SelectorData> ls = SELECTOR_MAP
+ .getOrDefault(selector.getPluginName(), new ArrayList<>())
+ .stream()
+ .sorted(SELECTOR_DATA_COMPARATOR)
+ .collect(Collectors.toList());
+ ls.add(selector);
+ SELECTOR_MAP.put(selector.getPluginName(), ls);
+ });
+ break;
+ default:
+ changed.forEach(selector -> {
+ List<SelectorData> ls = SELECTOR_MAP
+ .getOrDefault(selector.getPluginName(), new ArrayList<>())
+ .stream()
+ .filter(s -> !s.getId().equals(selector.getId()))
+ .sorted(SELECTOR_DATA_COMPARATOR)
+ .collect(Collectors.toList());
+ ls.add(selector);
+ SELECTOR_MAP.put(selector.getPluginName(), ls);
+ });
+ break;
+ }
+ publishConfig(ConsulConstants.SELECTOR_DATA, SELECTOR_MAP);
+ }
+
+ @Override
+ public void onMetaDataChanged(final List<MetaData> changed, final DataEventTypeEnum eventType) {
+ updateMetaDataMap(getConfig(ConsulConstants.META_DATA));
+ switch (eventType) {
+ case DELETE:
+ changed.forEach(meta -> META_DATA.remove(meta.getPath()));
+ break;
+ case REFRESH:
+ case MYSELF:
+ Set<String> set = new HashSet<>(META_DATA.keySet());
+ changed.forEach(meta -> {
+ set.remove(meta.getPath());
+ META_DATA.put(meta.getPath(), meta);
+ });
+ META_DATA.keySet().removeAll(set);
+ break;
+ default:
+ changed.forEach(meta -> {
+ META_DATA
+ .values()
+ .stream()
+ .filter(md -> Objects.equals(md.getId(), meta.getId()))
+ .forEach(md -> META_DATA.remove(md.getPath()));
+
+ META_DATA.put(meta.getPath(), meta);
+ });
+ break;
+ }
+ publishConfig(ConsulConstants.META_DATA, META_DATA);
+ }
+
+ @Override
+ public void onRuleChanged(final List<RuleData> changed, final DataEventTypeEnum eventType) {
+ updateRuleMap(getConfig(ConsulConstants.RULE_DATA));
+ switch (eventType) {
+ case DELETE:
+ changed.forEach(rule -> {
+ List<RuleData> ls = RULE_MAP
+ .getOrDefault(rule.getSelectorId(), new ArrayList<>())
+ .stream()
+ .filter(s -> !s.getId().equals(rule.getId()))
+ .sorted(RULE_DATA_COMPARATOR)
+ .collect(Collectors.toList());
+ RULE_MAP.put(rule.getSelectorId(), ls);
+ });
+ break;
+ case REFRESH:
+ case MYSELF:
+ RULE_MAP.keySet().removeAll(RULE_MAP.keySet());
+ changed.forEach(rule -> {
+ List<RuleData> ls = RULE_MAP
+ .getOrDefault(rule.getSelectorId(), new ArrayList<>())
+ .stream()
+ .sorted(RULE_DATA_COMPARATOR)
+ .collect(Collectors.toList());
+ ls.add(rule);
+ RULE_MAP.put(rule.getSelectorId(), ls);
+ });
+ break;
+ default:
+ changed.forEach(rule -> {
+ List<RuleData> ls = RULE_MAP
+ .getOrDefault(rule.getSelectorId(), new ArrayList<>())
+ .stream()
+ .filter(s -> !s.getId().equals(rule.getId()))
+ .collect(Collectors.toList());
+ ls.add(rule);
+ ls.sort(RULE_DATA_COMPARATOR);
+ RULE_MAP.put(rule.getSelectorId(), ls);
+ });
+ break;
+ }
+
+ publishConfig(ConsulConstants.RULE_DATA, RULE_MAP);
+ }
+
+ @SneakyThrows
+ private void publishConfig(final String dataKey, final Object data) {
+ consulClient.setKVValue(dataKey, GsonUtils.getInstance().toJson(data));
+ }
+
+ @SneakyThrows
+ private String getConfig(final String dataKey) {
+ Response<GetValue> kvValue = consulClient.getKVValue(dataKey);
+ return Objects.nonNull(kvValue.getValue()) ? kvValue.getValue().getDecodedValue() : ConsulConstants.EMPTY_CONFIG_DEFAULT_VALUE;
+ }
+
+ private void updateAuthMap(final String configInfo) {
+ JsonObject jo = GsonUtils.getInstance().fromJson(configInfo, JsonObject.class);
+ Set<String> set = new HashSet<>(AUTH_MAP.keySet());
+ for (Map.Entry<String, JsonElement> e : jo.entrySet()) {
+ set.remove(e.getKey());
+ AUTH_MAP.put(e.getKey(), GsonUtils.getInstance().fromJson(e.getValue(), AppAuthData.class));
+ }
+ AUTH_MAP.keySet().removeAll(set);
+ }
+
+ private void updatePluginMap(final String configInfo) {
+ JsonObject jo = GsonUtils.getInstance().fromJson(configInfo, JsonObject.class);
+ Set<String> set = new HashSet<>(PLUGIN_MAP.keySet());
+ for (Map.Entry<String, JsonElement> e : jo.entrySet()) {
+ set.remove(e.getKey());
+ PLUGIN_MAP.put(e.getKey(), GsonUtils.getInstance().fromJson(e.getValue(), PluginData.class));
+ }
+ PLUGIN_MAP.keySet().removeAll(set);
+ }
+
+ private void updateSelectorMap(final String configInfo) {
+ JsonObject jo = GsonUtils.getInstance().fromJson(configInfo, JsonObject.class);
+ Set<String> set = new HashSet<>(SELECTOR_MAP.keySet());
+ for (Map.Entry<String, JsonElement> e : jo.entrySet()) {
+ set.remove(e.getKey());
+ List<SelectorData> ls = new ArrayList<>();
+ e.getValue().getAsJsonArray().forEach(je -> ls.add(GsonUtils.getInstance().fromJson(je, SelectorData.class)));
+ SELECTOR_MAP.put(e.getKey(), ls);
+ }
+ SELECTOR_MAP.keySet().removeAll(set);
+ }
+
+ private void updateMetaDataMap(final String configInfo) {
+ JsonObject jo = GsonUtils.getInstance().fromJson(configInfo, JsonObject.class);
+ Set<String> set = new HashSet<>(META_DATA.keySet());
+ for (Map.Entry<String, JsonElement> e : jo.entrySet()) {
+ set.remove(e.getKey());
+ META_DATA.put(e.getKey(), GsonUtils.getInstance().fromJson(e.getValue(), MetaData.class));
+ }
+ META_DATA.keySet().removeAll(set);
+ }
+
+ private void updateRuleMap(final String configInfo) {
+ JsonObject jo = GsonUtils.getInstance().fromJson(configInfo, JsonObject.class);
+ Set<String> set = new HashSet<>(RULE_MAP.keySet());
+ for (Map.Entry<String, JsonElement> e : jo.entrySet()) {
+ set.remove(e.getKey());
+ List<RuleData> ls = new ArrayList<>();
+ e.getValue().getAsJsonArray().forEach(je -> ls.add(GsonUtils.getInstance().fromJson(je, RuleData.class)));
+ RULE_MAP.put(e.getKey(), ls);
+ }
+ RULE_MAP.keySet().removeAll(set);
+ }
+}
diff --git a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/consul/ConsulDataInit.java b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/consul/ConsulDataInit.java
new file mode 100644
index 0000000..a99b0eb
--- /dev/null
+++ b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/consul/ConsulDataInit.java
@@ -0,0 +1,59 @@
+/*
+ * 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.shenyu.admin.listener.consul;
+
+import com.ecwid.consul.v1.ConsulClient;
+import lombok.SneakyThrows;
+import org.apache.shenyu.admin.service.SyncDataService;
+import org.apache.shenyu.common.constant.ConsulConstants;
+import org.apache.shenyu.common.enums.DataEventTypeEnum;
+import org.springframework.boot.CommandLineRunner;
+
+/**
+ * The type Consul data init.
+ */
+public class ConsulDataInit implements CommandLineRunner {
+ private final ConsulClient consulClient;
+
+ private final SyncDataService syncDataService;
+
+ /**
+ * Instantiates a new Consul data init.
+ * @param consulClient the Consul client
+ * @param syncDataService the sync data service
+ */
+ public ConsulDataInit(final ConsulClient consulClient, final SyncDataService syncDataService) {
+ this.consulClient = consulClient;
+ this.syncDataService = syncDataService;
+ }
+
+ @Override
+ public void run(final String... args) {
+ String pluginData = ConsulConstants.PLUGIN_DATA;
+ String authData = ConsulConstants.AUTH_DATA;
+ String metaData = ConsulConstants.META_DATA;
+ if (dataKeyNotExist(pluginData) && dataKeyNotExist(authData) && dataKeyNotExist(metaData)) {
+ syncDataService.syncAll(DataEventTypeEnum.REFRESH);
+ }
+ }
+
+ @SneakyThrows
+ private boolean dataKeyNotExist(final String dataKey) {
+ return consulClient.getKVValue(dataKey).getValue() == null;
+ }
+}
diff --git a/shenyu-admin/src/main/resources/application.yml b/shenyu-admin/src/main/resources/application.yml
index 71eedd5..c755339 100644
--- a/shenyu-admin/src/main/resources/application.yml
+++ b/shenyu-admin/src/main/resources/application.yml
@@ -73,6 +73,8 @@ shenyu:
# secretKey:
# etcd:
# url: http://localhost:2379
+# consul:
+# url: http://localhost:8500
aes:
secret:
key: 2095132720951327
diff --git a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/consul/ConsulDataChangedListenerTest.java b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/consul/ConsulDataChangedListenerTest.java
new file mode 100644
index 0000000..d212308
--- /dev/null
+++ b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/consul/ConsulDataChangedListenerTest.java
@@ -0,0 +1,208 @@
+/*
+ * 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.shenyu.admin.listener.consul;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.kv.model.GetValue;
+import com.google.common.collect.ImmutableList;
+import lombok.SneakyThrows;
+import org.apache.shenyu.common.constant.ConsulConstants;
+import org.apache.shenyu.common.dto.AppAuthData;
+import org.apache.shenyu.common.dto.MetaData;
+import org.apache.shenyu.common.dto.PluginData;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.enums.DataEventTypeEnum;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The testCase for {@link ConsulDataChangedListener}.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ConsulDataChangedListenerTest {
+
+ private static final String MOCK_APP_KEY = "MOCK_APP_KEY";
+
+ private static final String MOCK_APP_SECRET = "MOCK_APP_SECRET";
+
+ private static final String MOCK_ID = "MOCK_ID";
+
+ private static final String MOCK_PATH = "MOCK_PATH";
+
+ private static final String MOCK_APP_NAME = "MOCK_APP_NAME";
+
+ private static final String MOCK_NAME = "MOCK_NAME";
+
+ private static final String MOCK_CONFIG = "MOCK_CONFIG";
+
+ private static final String MOCK_PLUGIN_NAME = "MOCK_PLUGIN_NAME";
+
+ private static final String MOCK_SELECTOR_ID = "MOCK_SELECTOR_ID";
+
+ @Mock
+ private ConsulClient consulClient;
+
+ @InjectMocks
+ private ConsulDataChangedListener consulDataChangedListener;
+
+ @Test
+ @SneakyThrows
+ public void testOnAppAuthChanged() {
+ String config = "{\"divide\":{\"appKey\":\"appKey\",\"appSecret\":\"appSecret\",\"open\":true}}";
+ final AppAuthData appAuthData = AppAuthData.builder().appKey(MOCK_APP_KEY).appSecret(MOCK_APP_SECRET).build();
+
+ Response<GetValue> response = mock(Response.class);
+ GetValue getValueModel = mock(GetValue.class);
+ when(consulClient.getKVValue(anyString())).thenReturn(response);
+ when(response.getValue()).thenReturn(getValueModel);
+ when(getValueModel.getDecodedValue()).thenReturn(config);
+
+ consulDataChangedListener.onAppAuthChanged(ImmutableList.of(appAuthData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onAppAuthChanged(ImmutableList.of(appAuthData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onAppAuthChanged(ImmutableList.of(appAuthData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onAppAuthChanged(ImmutableList.of(appAuthData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(4)).setKVValue(any(String.class), any(String.class));
+
+ getValueModel.setValue(ConsulConstants.EMPTY_CONFIG_DEFAULT_VALUE);
+ consulDataChangedListener.onAppAuthChanged(ImmutableList.of(appAuthData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onAppAuthChanged(ImmutableList.of(appAuthData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onAppAuthChanged(ImmutableList.of(appAuthData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onAppAuthChanged(ImmutableList.of(appAuthData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(8)).setKVValue(any(String.class), any(String.class));
+ }
+
+ @Test
+ @SneakyThrows
+ public void testOnPluginChanged() {
+ String config = "{\"divide\":{\"id\":\"id\",\"name\":\"name\",\"enabled\":true}}";
+ final PluginData pluginData = PluginData.builder().id(MOCK_ID).name(MOCK_NAME).config(MOCK_CONFIG).build();
+ Response<GetValue> response = mock(Response.class);
+ GetValue getValueModel = mock(GetValue.class);
+ when(consulClient.getKVValue(anyString())).thenReturn(response);
+ when(response.getValue()).thenReturn(getValueModel);
+ when(getValueModel.getDecodedValue()).thenReturn(config);
+
+ consulDataChangedListener.onPluginChanged(ImmutableList.of(pluginData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onPluginChanged(ImmutableList.of(pluginData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onPluginChanged(ImmutableList.of(pluginData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onPluginChanged(ImmutableList.of(pluginData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(4)).setKVValue(any(String.class), any(String.class));
+
+ getValueModel.setValue(ConsulConstants.EMPTY_CONFIG_DEFAULT_VALUE);
+ consulDataChangedListener.onPluginChanged(ImmutableList.of(pluginData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onPluginChanged(ImmutableList.of(pluginData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onPluginChanged(ImmutableList.of(pluginData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onPluginChanged(ImmutableList.of(pluginData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(8)).setKVValue(any(String.class), any(String.class));
+ }
+
+ @Test
+ @SneakyThrows
+ public void testOnSelectorChanged() {
+ String config = "{\"divide\":[{\"id\":\"id\",\"name\":\"name\",\"enabled\":true}]}";
+ final SelectorData selectorData = SelectorData.builder().id(MOCK_ID).name(MOCK_NAME).pluginName(MOCK_PLUGIN_NAME).build();
+
+ Response<GetValue> response = mock(Response.class);
+ GetValue getValueModel = mock(GetValue.class);
+ when(consulClient.getKVValue(anyString())).thenReturn(response);
+ when(response.getValue()).thenReturn(getValueModel);
+ when(getValueModel.getDecodedValue()).thenReturn(config);
+
+ consulDataChangedListener.onSelectorChanged(ImmutableList.of(selectorData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onSelectorChanged(ImmutableList.of(selectorData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onSelectorChanged(ImmutableList.of(selectorData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onSelectorChanged(ImmutableList.of(selectorData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(4)).setKVValue(any(String.class), any(String.class));
+
+ getValueModel.setValue(ConsulConstants.EMPTY_CONFIG_DEFAULT_VALUE);
+ consulDataChangedListener.onSelectorChanged(ImmutableList.of(selectorData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onSelectorChanged(ImmutableList.of(selectorData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onSelectorChanged(ImmutableList.of(selectorData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onSelectorChanged(ImmutableList.of(selectorData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(8)).setKVValue(any(String.class), any(String.class));
+ }
+
+ @Test
+ @SneakyThrows
+ public void testOnMetaDataChanged() {
+ String config = "{\"divide\":{\"id\":\"id\",\"appName\":\"appName\",\"enabled\":true}}";
+ final MetaData metaData = MetaData.builder().id(MOCK_ID).path(MOCK_PATH).appName(MOCK_APP_NAME).build();
+
+ Response<GetValue> response = mock(Response.class);
+ GetValue getValueModel = mock(GetValue.class);
+ when(consulClient.getKVValue(anyString())).thenReturn(response);
+ when(response.getValue()).thenReturn(getValueModel);
+ when(getValueModel.getDecodedValue()).thenReturn(config);
+
+ consulDataChangedListener.onMetaDataChanged(ImmutableList.of(metaData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onMetaDataChanged(ImmutableList.of(metaData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onMetaDataChanged(ImmutableList.of(metaData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onMetaDataChanged(ImmutableList.of(metaData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(4)).setKVValue(any(String.class), any(String.class));
+
+ getValueModel.setValue(ConsulConstants.EMPTY_CONFIG_DEFAULT_VALUE);
+ consulDataChangedListener.onMetaDataChanged(ImmutableList.of(metaData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onMetaDataChanged(ImmutableList.of(metaData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onMetaDataChanged(ImmutableList.of(metaData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onMetaDataChanged(ImmutableList.of(metaData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(8)).setKVValue(any(String.class), any(String.class));
+ }
+
+ @Test
+ @SneakyThrows
+ public void testOnRuleChanged() {
+ String config = "{\"divide\":[{\"id\":\"id\",\"appName\":\"appName\",\"enabled\":true}]}";
+ final RuleData ruleData = RuleData.builder()
+ .id(MOCK_ID)
+ .name(MOCK_NAME)
+ .pluginName(MOCK_PLUGIN_NAME)
+ .selectorId(MOCK_SELECTOR_ID)
+ .build();
+
+ Response<GetValue> response = mock(Response.class);
+ GetValue getValueModel = mock(GetValue.class);
+ when(consulClient.getKVValue(anyString())).thenReturn(response);
+ when(response.getValue()).thenReturn(getValueModel);
+ when(getValueModel.getDecodedValue()).thenReturn(config);
+
+ consulDataChangedListener.onRuleChanged(ImmutableList.of(ruleData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onRuleChanged(ImmutableList.of(ruleData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onRuleChanged(ImmutableList.of(ruleData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onRuleChanged(ImmutableList.of(ruleData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(4)).setKVValue(any(String.class), any(String.class));
+
+ getValueModel.setValue(ConsulConstants.EMPTY_CONFIG_DEFAULT_VALUE);
+ consulDataChangedListener.onRuleChanged(ImmutableList.of(ruleData), DataEventTypeEnum.DELETE);
+ consulDataChangedListener.onRuleChanged(ImmutableList.of(ruleData), DataEventTypeEnum.REFRESH);
+ consulDataChangedListener.onRuleChanged(ImmutableList.of(ruleData), DataEventTypeEnum.MYSELF);
+ consulDataChangedListener.onRuleChanged(ImmutableList.of(ruleData), DataEventTypeEnum.CREATE);
+ verify(consulClient, times(8)).setKVValue(any(String.class), any(String.class));
+ }
+}
diff --git a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/consul/ConsulDataInitTest.java b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/consul/ConsulDataInitTest.java
new file mode 100644
index 0000000..431ec98
--- /dev/null
+++ b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/consul/ConsulDataInitTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.shenyu.admin.listener.consul;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.kv.model.GetValue;
+import lombok.SneakyThrows;
+import org.apache.shenyu.admin.service.SyncDataService;
+import org.apache.shenyu.common.enums.DataEventTypeEnum;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test cases for {@link ConsulDataInit}.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ConsulDataInitTest {
+
+ @Mock
+ private ConsulClient consulClient;
+
+ @Mock
+ private SyncDataService syncDataService;
+
+ @Test
+ @SneakyThrows
+ public void testRun() {
+ ConsulDataInit consulDataInit = new ConsulDataInit(consulClient, syncDataService);
+
+ Response<GetValue> response = mock(Response.class);
+ when(consulClient.getKVValue(anyString())).thenReturn(response);
+ when(response.getValue()).thenReturn(null);
+ consulDataInit.run();
+ when(response.getValue()).thenReturn(new GetValue());
+ consulDataInit.run();
+
+ verify(syncDataService, times(1)).syncAll(any(DataEventTypeEnum.class));
+ }
+}
diff --git a/shenyu-bootstrap/pom.xml b/shenyu-bootstrap/pom.xml
index 9fbc728..9fdabc7 100644
--- a/shenyu-bootstrap/pom.xml
+++ b/shenyu-bootstrap/pom.xml
@@ -344,6 +344,13 @@
</dependency>
<!--shenyu debug plugin end-->
+ <!--shenyu data sync start use consul-->
+ <dependency>
+ <groupId>org.apache.shenyu</groupId>
+ <artifactId>shenyu-spring-boot-starter-sync-data-consul</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<!-- shenyu modify response plugin start-->
<dependency>
<groupId>org.apache.shenyu</groupId>
diff --git a/shenyu-bootstrap/src/main/resources/application-local.yml b/shenyu-bootstrap/src/main/resources/application-local.yml
index 7b5a36d..ba75ed6 100644
--- a/shenyu-bootstrap/src/main/resources/application-local.yml
+++ b/shenyu-bootstrap/src/main/resources/application-local.yml
@@ -74,6 +74,10 @@ shenyu:
# secretKey:
# etcd:
# url: http://localhost:2379
+# consul:
+# url: http://localhost:8500
+# waitTime: 1000
+# watchDelay: 1000
exclude:
enabled: false
paths:
diff --git a/shenyu-common/src/main/java/org/apache/shenyu/common/constant/ConsulConstants.java b/shenyu-common/src/main/java/org/apache/shenyu/common/constant/ConsulConstants.java
new file mode 100644
index 0000000..a9862ba
--- /dev/null
+++ b/shenyu-common/src/main/java/org/apache/shenyu/common/constant/ConsulConstants.java
@@ -0,0 +1,61 @@
+/*
+ * 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.shenyu.common.constant;
+
+/**
+ * Consul constants.
+ */
+public final class ConsulConstants {
+
+ public static final String SYNC_PRE_FIX = "shenyu/sync";
+
+ /**
+ * plugin data path.
+ */
+ public static final String PLUGIN_DATA = SYNC_PRE_FIX + "/plugin";
+
+ /**
+ * selector data path.
+ */
+ public static final String SELECTOR_DATA = SYNC_PRE_FIX + "/selector";
+
+ /**
+ * rule data path.
+ */
+ public static final String RULE_DATA = SYNC_PRE_FIX + "/rule";
+
+ /**
+ * auth data path.
+ */
+ public static final String AUTH_DATA = SYNC_PRE_FIX + "/auth";
+
+ /**
+ * meta data path.
+ */
+ public static final String META_DATA = SYNC_PRE_FIX + "/meta";
+
+ /**
+ * default value of get config.
+ */
+ public static final String EMPTY_CONFIG_DEFAULT_VALUE = "{}";
+
+ /**
+ * default value of config version index.
+ */
+ public static final Long INIT_CONFIG_VERSION_INDEX = -1L;
+}
diff --git a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml
index 4f36e62..4fbf5e7 100644
--- a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml
+++ b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml
@@ -34,6 +34,7 @@
<module>shenyu-spring-boot-starter-sync-data-http</module>
<module>shenyu-spring-boot-starter-sync-data-nacos</module>
<module>shenyu-spring-boot-starter-sync-data-etcd</module>
+ <module>shenyu-spring-boot-starter-sync-data-consul</module>
</modules>
diff --git a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/pom.xml
similarity index 62%
copy from shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml
copy to shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/pom.xml
index 4f36e62..a45a6a2 100644
--- a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml
+++ b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/pom.xml
@@ -21,20 +21,27 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.shenyu</groupId>
- <artifactId>shenyu-spring-boot-starter</artifactId>
+ <artifactId>shenyu-spring-boot-starter-sync-data-center</artifactId>
<version>2.4.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>shenyu-spring-boot-starter-sync-data-center</artifactId>
- <packaging>pom</packaging>
-
- <modules>
- <module>shenyu-spring-boot-starter-sync-data-zookeeper</module>
- <module>shenyu-spring-boot-starter-sync-data-websocket</module>
- <module>shenyu-spring-boot-starter-sync-data-http</module>
- <module>shenyu-spring-boot-starter-sync-data-nacos</module>
- <module>shenyu-spring-boot-starter-sync-data-etcd</module>
- </modules>
+ <artifactId>shenyu-spring-boot-starter-sync-data-consul</artifactId>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.shenyu</groupId>
+ <artifactId>shenyu-sync-data-consul</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
</project>
\ No newline at end of file
diff --git a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/java/org/apache/shenyu/springboot/sync/data/consul/ConsulSyncDataConfiguration.java b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/java/org/apache/shenyu/springboot/sync/data/consul/ConsulSyncDataConfiguration.java
new file mode 100644
index 0000000..39b44f1
--- /dev/null
+++ b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/java/org/apache/shenyu/springboot/sync/data/consul/ConsulSyncDataConfiguration.java
@@ -0,0 +1,89 @@
+/*
+ * 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.shenyu.springboot.sync.data.consul;
+
+import com.ecwid.consul.v1.ConsulClient;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shenyu.sync.data.api.AuthDataSubscriber;
+import org.apache.shenyu.sync.data.api.MetaDataSubscriber;
+import org.apache.shenyu.sync.data.api.PluginDataSubscriber;
+import org.apache.shenyu.sync.data.api.SyncDataService;
+import org.apache.shenyu.sync.data.consul.ConsulSyncDataService;
+import org.apache.shenyu.sync.data.consul.config.ConsulConfig;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Consul sync data configuration for spring boot.
+ */
+@Configuration
+@ConditionalOnClass(ConsulSyncDataConfiguration.class)
+@ConditionalOnProperty(prefix = "shenyu.sync.consul", name = "url")
+@Slf4j
+public class ConsulSyncDataConfiguration {
+
+ /**
+ * Sync data service.
+ *
+ * @param consulClient the consul client
+ * @param consulConfig the consul config
+ * @param pluginSubscriber the plugin subscriber
+ * @param metaSubscribers the meta subscribers
+ * @param authSubscribers the auth subscribers
+ * @return the sync data service
+ */
+ @Bean
+ public SyncDataService syncDataService(final ObjectProvider<ConsulClient> consulClient,
+ final ObjectProvider<ConsulConfig> consulConfig,
+ final ObjectProvider<PluginDataSubscriber> pluginSubscriber,
+ final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers,
+ final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {
+ log.info("you use consul sync shenyu data.......");
+ return new ConsulSyncDataService(consulClient.getIfAvailable(), consulConfig.getIfAvailable(), pluginSubscriber.getIfAvailable(),
+ metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));
+ }
+
+ /**
+ * Consul config.
+ *
+ * @return the consul config
+ */
+ @Bean
+ @ConfigurationProperties(prefix = "shenyu.sync.consul")
+ public ConsulConfig consulConfig() {
+ return new ConsulConfig();
+ }
+
+
+ /**
+ * init Consul client.
+ * @param consulConfig the consul config
+ * @return Consul client
+ */
+ @Bean
+ public ConsulClient consulClient(final ConsulConfig consulConfig) {
+ return new ConsulClient(consulConfig.getUrl());
+ }
+}
diff --git a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/resources/META-INF/spring.factories b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/resources/META-INF/spring.factories
new file mode 100644
index 0000000..fb5f40f
--- /dev/null
+++ b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.apache.shenyu.springboot.sync.data.consul.ConsulSyncDataConfiguration
diff --git a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/resources/META-INF/spring.provides b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/resources/META-INF/spring.provides
new file mode 100644
index 0000000..7145d6a
--- /dev/null
+++ b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-consul/src/main/resources/META-INF/spring.provides
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+provides: shenyu-spring-boot-starter-sync-data-consul
diff --git a/shenyu-sync-data-center/pom.xml b/shenyu-sync-data-center/pom.xml
index cea5d87..d68be9c 100644
--- a/shenyu-sync-data-center/pom.xml
+++ b/shenyu-sync-data-center/pom.xml
@@ -35,5 +35,6 @@
<module>shenyu-sync-data-http</module>
<module>shenyu-sync-data-nacos</module>
<module>shenyu-sync-data-etcd</module>
+ <module>shenyu-sync-data-consul</module>
</modules>
</project>
diff --git a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml b/shenyu-sync-data-center/shenyu-sync-data-consul/pom.xml
similarity index 65%
copy from shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml
copy to shenyu-sync-data-center/shenyu-sync-data-consul/pom.xml
index 4f36e62..7beac7f 100644
--- a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/pom.xml
+++ b/shenyu-sync-data-center/shenyu-sync-data-consul/pom.xml
@@ -21,20 +21,29 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.shenyu</groupId>
- <artifactId>shenyu-spring-boot-starter</artifactId>
+ <artifactId>shenyu-sync-data-center</artifactId>
<version>2.4.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>shenyu-spring-boot-starter-sync-data-center</artifactId>
- <packaging>pom</packaging>
+ <artifactId>shenyu-sync-data-consul</artifactId>
- <modules>
- <module>shenyu-spring-boot-starter-sync-data-zookeeper</module>
- <module>shenyu-spring-boot-starter-sync-data-websocket</module>
- <module>shenyu-spring-boot-starter-sync-data-http</module>
- <module>shenyu-spring-boot-starter-sync-data-nacos</module>
- <module>shenyu-spring-boot-starter-sync-data-etcd</module>
- </modules>
+ <properties>
+ <consul.version>2.2.6.RELEASE</consul.version>
+ </properties>
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.shenyu</groupId>
+ <artifactId>shenyu-sync-data-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.ecwid.consul</groupId>
+ <artifactId>consul-api</artifactId>
+ <version>${consul.api.version}</version>
+ </dependency>
+ </dependencies>
</project>
\ No newline at end of file
diff --git a/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/ConsulSyncDataService.java b/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/ConsulSyncDataService.java
new file mode 100644
index 0000000..3747785
--- /dev/null
+++ b/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/ConsulSyncDataService.java
@@ -0,0 +1,155 @@
+/*
+ * 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.shenyu.sync.data.consul;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.QueryParams;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.kv.model.GetValue;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shenyu.common.concurrent.ShenyuThreadFactory;
+import org.apache.shenyu.common.constant.ConsulConstants;
+import org.apache.shenyu.sync.data.api.AuthDataSubscriber;
+import org.apache.shenyu.sync.data.api.MetaDataSubscriber;
+import org.apache.shenyu.sync.data.api.PluginDataSubscriber;
+import org.apache.shenyu.sync.data.api.SyncDataService;
+import org.apache.shenyu.sync.data.consul.config.ConsulConfig;
+import org.apache.shenyu.sync.data.consul.handler.ConsulCacheHandler;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Consul sync data service.
+ */
+@Slf4j
+public class ConsulSyncDataService extends ConsulCacheHandler implements AutoCloseable, SyncDataService {
+
+ private final Map<String, OnChange> groupMap = new HashMap<>();
+
+ private final Map<String, Long> consulIndexes = new HashMap<>();
+
+ private final ScheduledThreadPoolExecutor executor;
+
+ private ScheduledFuture<?> watchFuture;
+
+ private ConsulConfig consulConfig;
+
+ private ConsulClient consulClient;
+
+ private final AtomicBoolean running = new AtomicBoolean(false);
+
+ /**
+ * Instantiates a new Consul sync data service.
+ *
+ * @param pluginDataSubscriber the plugin data subscriber
+ * @param metaDataSubscribers the meta data subscribers
+ * @param authDataSubscribers the auth data subscribers
+ */
+ public ConsulSyncDataService(final ConsulClient consulClient, final ConsulConfig consulConfig, final PluginDataSubscriber pluginDataSubscriber,
+ final List<MetaDataSubscriber> metaDataSubscribers, final List<AuthDataSubscriber> authDataSubscribers) {
+ super(pluginDataSubscriber, metaDataSubscribers, authDataSubscribers);
+ this.consulClient = consulClient;
+ this.consulConfig = consulConfig;
+ this.executor = new ScheduledThreadPoolExecutor(1, ShenyuThreadFactory.create("consul-config-watch", true));
+ consulIndexes.put(ConsulConstants.SYNC_PRE_FIX, 0L);
+ initUpdateMap();
+ start();
+ }
+
+ /**
+ * init config key and update method mapping.
+ */
+ private void initUpdateMap() {
+ groupMap.put(ConsulConstants.PLUGIN_DATA, this::updatePluginData);
+ groupMap.put(ConsulConstants.SELECTOR_DATA, this::updateSelectorMap);
+ groupMap.put(ConsulConstants.RULE_DATA, this::updateRuleMap);
+ groupMap.put(ConsulConstants.META_DATA, this::updateMetaDataMap);
+ groupMap.put(ConsulConstants.AUTH_DATA, this::updateAuthMap);
+ }
+
+ private void watchConfigKeyValues() {
+ if (this.running.get()) {
+ for (String context : this.consulIndexes.keySet()) {
+ try {
+ Long currentIndex = this.consulIndexes.get(context);
+ if (currentIndex == null) {
+ currentIndex = ConsulConstants.INIT_CONFIG_VERSION_INDEX;
+ }
+ Response<List<GetValue>> response = this.consulClient.getKVValues(context, null,
+ new QueryParams(consulConfig.getWaitTime(), currentIndex));
+ if (response.getValue() == null || response.getValue().isEmpty()) {
+ if (log.isTraceEnabled()) {
+ log.trace("No value for context " + context);
+ }
+ continue;
+ }
+ Long newIndex = response.getConsulIndex();
+ if (newIndex == null || newIndex.equals(currentIndex)) {
+ if (log.isTraceEnabled()) {
+ log.trace("Same index for context " + context);
+ }
+ continue;
+ }
+ if (!this.consulIndexes.containsValue(newIndex)
+ && !currentIndex.equals(ConsulConstants.INIT_CONFIG_VERSION_INDEX)) {
+ if (log.isTraceEnabled()) {
+ log.trace("Context " + context + " has new index " + newIndex);
+ }
+ final Long lastIndex = currentIndex;
+ response.getValue().forEach(data -> {
+ if (data.getModifyIndex() == lastIndex) {
+ //data has not changed
+ return;
+ }
+ groupMap.get(data.getKey()).change(data.getDecodedValue());
+ });
+
+ } else if (log.isTraceEnabled()) {
+ log.info("Event for index already published for context " + context);
+ }
+ this.consulIndexes.put(context, newIndex);
+ } catch (Exception e) {
+ log.warn("Error querying consul Key/Values for context '" + context + "'. Message: " + e.getMessage());
+ }
+ }
+ }
+ }
+
+ /**
+ * Start.
+ */
+ public void start() {
+ if (this.running.compareAndSet(false, true)) {
+ this.watchFuture = this.executor.scheduleWithFixedDelay(this::watchConfigKeyValues,
+ 5, consulConfig.getWatchDelay(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (this.running.compareAndSet(true, false) && this.watchFuture != null) {
+ this.watchFuture.cancel(true);
+ }
+ }
+}
diff --git a/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/config/ConsulConfig.java b/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/config/ConsulConfig.java
new file mode 100644
index 0000000..6c85338
--- /dev/null
+++ b/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/config/ConsulConfig.java
@@ -0,0 +1,34 @@
+/*
+ * 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.shenyu.sync.data.consul.config;
+
+import lombok.Data;
+
+/**
+ * The type Consul config.
+ */
+@Data
+public class ConsulConfig {
+
+ private String url;
+
+ private int waitTime;
+
+ private int watchDelay;
+
+}
diff --git a/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/handler/ConsulCacheHandler.java b/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/handler/ConsulCacheHandler.java
new file mode 100644
index 0000000..d4bce59
--- /dev/null
+++ b/shenyu-sync-data-center/shenyu-sync-data-consul/src/main/java/org/apache/shenyu/sync/data/consul/handler/ConsulCacheHandler.java
@@ -0,0 +1,124 @@
+/*
+ * 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.shenyu.sync.data.consul.handler;
+
+import com.google.gson.JsonParseException;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shenyu.common.dto.AppAuthData;
+import org.apache.shenyu.common.dto.MetaData;
+import org.apache.shenyu.common.dto.PluginData;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.sync.data.api.AuthDataSubscriber;
+import org.apache.shenyu.sync.data.api.MetaDataSubscriber;
+import org.apache.shenyu.sync.data.api.PluginDataSubscriber;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Consul cache handler.
+ */
+@Slf4j
+public class ConsulCacheHandler {
+
+ private final PluginDataSubscriber pluginDataSubscriber;
+
+ private final List<MetaDataSubscriber> metaDataSubscribers;
+
+ private final List<AuthDataSubscriber> authDataSubscribers;
+
+ public ConsulCacheHandler(final PluginDataSubscriber pluginDataSubscriber,
+ final List<MetaDataSubscriber> metaDataSubscribers,
+ final List<AuthDataSubscriber> authDataSubscribers) {
+ this.pluginDataSubscriber = pluginDataSubscriber;
+ this.metaDataSubscribers = metaDataSubscribers;
+ this.authDataSubscribers = authDataSubscribers;
+ }
+
+ protected void updatePluginData(final String configInfo) {
+ try {
+ List<PluginData> pluginDataList = new ArrayList<>(GsonUtils.getInstance().toObjectMap(configInfo, PluginData.class).values());
+ pluginDataList.forEach(pluginData -> Optional.ofNullable(pluginDataSubscriber).ifPresent(subscriber -> {
+ subscriber.unSubscribe(pluginData);
+ subscriber.onSubscribe(pluginData);
+ }));
+ } catch (JsonParseException e) {
+ log.error("sync plugin data have error:", e);
+ }
+ }
+
+ protected void updateSelectorMap(final String configInfo) {
+ try {
+ List<SelectorData> selectorDataList = GsonUtils.getInstance().toObjectMapList(configInfo, SelectorData.class).values().stream().flatMap(Collection::stream).collect(Collectors.toList());
+ selectorDataList.forEach(selectorData -> Optional.ofNullable(pluginDataSubscriber).ifPresent(subscriber -> {
+ subscriber.unSelectorSubscribe(selectorData);
+ subscriber.onSelectorSubscribe(selectorData);
+ }));
+ } catch (JsonParseException e) {
+ log.error("sync selector data have error:", e);
+ }
+ }
+
+ protected void updateRuleMap(final String configInfo) {
+ try {
+ List<RuleData> ruleDataList = GsonUtils.getInstance().toObjectMapList(configInfo, RuleData.class).values()
+ .stream().flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ ruleDataList.forEach(ruleData -> Optional.ofNullable(pluginDataSubscriber).ifPresent(subscriber -> {
+ subscriber.unRuleSubscribe(ruleData);
+ subscriber.onRuleSubscribe(ruleData);
+ }));
+ } catch (JsonParseException e) {
+ log.error("sync rule data have error:", e);
+ }
+ }
+
+ protected void updateMetaDataMap(final String configInfo) {
+ try {
+ List<MetaData> metaDataList = new ArrayList<>(GsonUtils.getInstance().toObjectMap(configInfo, MetaData.class).values());
+ metaDataList.forEach(metaData -> metaDataSubscribers.forEach(subscriber -> {
+ subscriber.unSubscribe(metaData);
+ subscriber.onSubscribe(metaData);
+ }));
+ } catch (JsonParseException e) {
+ log.error("sync meta data have error:", e);
+ }
+ }
+
+ protected void updateAuthMap(final String configInfo) {
+ try {
+ List<AppAuthData> appAuthDataList = new ArrayList<>(GsonUtils.getInstance().toObjectMap(configInfo, AppAuthData.class).values());
+ appAuthDataList.forEach(appAuthData -> authDataSubscribers.forEach(subscriber -> {
+ subscriber.unSubscribe(appAuthData);
+ subscriber.onSubscribe(appAuthData);
+ }));
+ } catch (JsonParseException e) {
+ log.error("sync auth data have error:", e);
+ }
+ }
+
+ protected interface OnChange {
+
+ void change(String changeData);
+ }
+}
diff --git a/shenyu-sync-data-center/shenyu-sync-data-consul/src/test/java/org/apache/shenyu/sync/data/consul/ConsulSyncDataServiceTest.java b/shenyu-sync-data-center/shenyu-sync-data-consul/src/test/java/org/apache/shenyu/sync/data/consul/ConsulSyncDataServiceTest.java
new file mode 100644
index 0000000..cb4dab2
--- /dev/null
+++ b/shenyu-sync-data-center/shenyu-sync-data-consul/src/test/java/org/apache/shenyu/sync/data/consul/ConsulSyncDataServiceTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.shenyu.sync.data.consul;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.kv.model.GetValue;
+import lombok.SneakyThrows;
+import org.apache.shenyu.common.constant.ConsulConstants;
+import org.apache.shenyu.common.utils.ReflectUtils;
+import org.apache.shenyu.sync.data.consul.config.ConsulConfig;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * test case for {@link ConsulSyncDataService}.
+ */
+public class ConsulSyncDataServiceTest {
+
+ private ConsulClient consulClient;
+
+ private ConsulSyncDataService consulSyncDataService;
+
+ @Before
+ public void setup() {
+ consulClient = mock(ConsulClient.class);
+ ConsulConfig consulConfig = new ConsulConfig();
+ consulConfig.setWaitTime(1000);
+ consulConfig.setWatchDelay(1000);
+ consulSyncDataService = new ConsulSyncDataService(consulClient, consulConfig, null,
+ Collections.emptyList(), Collections.emptyList());
+ }
+
+ @Test
+ public void testStart() {
+ consulSyncDataService.start();
+ }
+
+ @Test
+ public void testClose() {
+ consulSyncDataService.close();
+ }
+
+ @Test
+ @SneakyThrows
+ public void testWatch() {
+ Long index = 1L;
+ final List<GetValue> list = new ArrayList<>();
+ GetValue getValue = mock(GetValue.class);
+ when(getValue.getModifyIndex()).thenReturn(index);
+ when(getValue.getKey()).thenReturn(ConsulConstants.PLUGIN_DATA);
+ when(getValue.getDecodedValue()).thenReturn("{}");
+ list.add(getValue);
+ Response response = mock(Response.class);
+ when(consulClient.getKVValues(any(), any(), any()))
+ .thenReturn(response);
+ when(response.getValue()).thenReturn(list);
+ when(response.getConsulIndex()).thenReturn(1L);
+
+ Method watchConfigKeyValues = consulSyncDataService.getClass().getDeclaredMethod("watchConfigKeyValues");
+ watchConfigKeyValues.setAccessible(true);
+ watchConfigKeyValues.invoke(consulSyncDataService);
+
+ Map<String, Long> consulIndexes = (Map<String, Long>) ReflectUtils.getFieldValue(consulSyncDataService, "consulIndexes");
+ Assert.assertEquals(index, consulIndexes.get(ConsulConstants.SYNC_PRE_FIX));
+ }
+}
diff --git a/shenyu-sync-data-center/shenyu-sync-data-consul/src/test/java/org/apache/shenyu/sync/data/consul/handler/ConsulCacheHandlerTest.java b/shenyu-sync-data-center/shenyu-sync-data-consul/src/test/java/org/apache/shenyu/sync/data/consul/handler/ConsulCacheHandlerTest.java
new file mode 100644
index 0000000..4e75bf1
--- /dev/null
+++ b/shenyu-sync-data-center/shenyu-sync-data-consul/src/test/java/org/apache/shenyu/sync/data/consul/handler/ConsulCacheHandlerTest.java
@@ -0,0 +1,215 @@
+/*
+ * 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.shenyu.sync.data.consul.handler;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shenyu.common.dto.PluginData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.dto.AppAuthData;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.MetaData;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.sync.data.api.AuthDataSubscriber;
+import org.apache.shenyu.sync.data.api.MetaDataSubscriber;
+import org.apache.shenyu.sync.data.api.PluginDataSubscriber;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Collections;
+
+/**
+ * test case for {@link ConsulCacheHandler}.
+ */
+@Slf4j
+@SuppressWarnings("all")
+public final class ConsulCacheHandlerTest {
+
+ @SneakyThrows
+ @Test
+ public void testUpdatePluginMap() {
+ String pluginName1 = "PLUGIN_NAME_1";
+ String pluginName2 = "PLUGIN_NAME_2";
+ PluginData pluginData1 =
+ PluginData.builder().name(pluginName1).id("plugin_1").config("config_1").build();
+ PluginData pluginData2 =
+ PluginData.builder().name(pluginName2).id("plugin_2").config("config_2").build();
+ String pluginData = GsonUtils.getInstance()
+ .toJson(ImmutableMap.of(pluginName2, pluginData2, pluginName1, pluginData1));
+
+ final List<PluginData> onSubscribeList = new ArrayList<>();
+ final List<PluginData> unsubscribeList = new ArrayList<>();
+ ConsulCacheHandler consulCacheHandler = new ConsulCacheHandler(new PluginDataSubscriber() {
+ @Override
+ public void onSubscribe(final PluginData pluginData) {
+ onSubscribeList.add(pluginData);
+ }
+
+ @Override
+ public void unSubscribe(final PluginData pluginData) {
+ unsubscribeList.add(pluginData);
+ }
+ }, Collections.emptyList(), Collections.emptyList());
+ consulCacheHandler.updatePluginData(pluginData);
+ Assert.assertEquals(2, onSubscribeList.size());
+ Assert.assertEquals(2, unsubscribeList.size());
+ }
+
+ @SneakyThrows
+ @Test
+ public void testUpdateSelectorMap() {
+ String selectorDataPluginName1 = "SELECTOR_DATA_1";
+ String selectorDataPluginName2 = "SELECTOR_DATA_2";
+ SelectorData selectorData1 =
+ SelectorData.builder()
+ .pluginName(selectorDataPluginName1)
+ .id("select_1")
+ .name("SELECT_DATA_NAME_1")
+ .build();
+ SelectorData selectorData2 =
+ SelectorData.builder()
+ .pluginName(selectorDataPluginName2)
+ .id("select_2")
+ .name("SELECT_DATA_NAME_2")
+ .build();
+
+ String selectorDataParam = GsonUtils.getInstance()
+ .toJson(ImmutableMap.of(selectorDataPluginName2, ImmutableList.of(selectorData2),
+ selectorDataPluginName1, ImmutableList.of(selectorData1)));
+ final List<SelectorData> subscribeList = new ArrayList<>();
+ final List<SelectorData> unsubscribeList = new ArrayList<>();
+ ConsulCacheHandler consulCacheHandler = new ConsulCacheHandler(new PluginDataSubscriber() {
+ @Override
+ public void onSelectorSubscribe(final SelectorData selectorData) {
+ subscribeList.add(selectorData);
+ }
+
+ @Override
+ public void unSelectorSubscribe(final SelectorData selectorData) {
+ unsubscribeList.add(selectorData);
+ }
+ }, Collections.emptyList(), Collections.emptyList());
+ consulCacheHandler.updateSelectorMap(selectorDataParam);
+ Assert.assertEquals(2, subscribeList.size());
+ Assert.assertEquals(2, unsubscribeList.size());
+ }
+
+ @SneakyThrows
+ @Test
+ public void testUpdateRuleMap() {
+ String ruleDataId1 = "RULE_DATA_1";
+ String ruleDataId2 = "RULE_DATA_2";
+ String selectorId1 = "ID_1";
+ String selectorId2 = "ID_2";
+ RuleData ruleData1 = RuleData.builder().selectorId(selectorId1).id(ruleDataId1).build();
+ RuleData ruleData2 = RuleData.builder().selectorId(selectorId2).id(ruleDataId2).build();
+ String ruleDataParam = GsonUtils.getInstance()
+ .toJson(
+ ImmutableMap.of(
+ selectorId2,
+ ImmutableList.of(ruleData2),
+ selectorId1,
+ ImmutableList.of(ruleData1)));
+ final List<RuleData> subscribeList = new ArrayList<>();
+ final List<RuleData> unsubscribeList = new ArrayList<>();
+ ConsulCacheHandler consulCacheHandler = new ConsulCacheHandler(new PluginDataSubscriber() {
+ @Override
+ public void onRuleSubscribe(final RuleData ruleData) {
+ subscribeList.add(ruleData);
+ }
+
+ @Override
+ public void unRuleSubscribe(final RuleData ruleData) {
+ unsubscribeList.add(ruleData);
+ }
+ }, Collections.emptyList(), Collections.emptyList());
+ consulCacheHandler.updateRuleMap(ruleDataParam);
+ Assert.assertEquals(2, subscribeList.size());
+ Assert.assertEquals(2, unsubscribeList.size());
+ }
+
+ @SneakyThrows
+ @Test
+ public void testUpdateMetaDataMap() {
+ String metadataPath1 = "METADATA_PATH_1";
+ String metadataPath2 = "METADATA_PATH_2";
+ MetaData metaData1 = MetaData.builder().path(metadataPath1).id("meta_1").build();
+ MetaData metaData2 = MetaData.builder().path(metadataPath2).id("meta_2").build();
+
+ String metaDataParam = GsonUtils.getInstance()
+ .toJson(ImmutableMap.of(metadataPath1, metaData1, metadataPath2, metaData2));
+ final List<MetaData> subscribeList = new ArrayList<>();
+ final List<MetaData> unsubscribeList = new ArrayList<>();
+ MetaDataSubscriber metaDataSubscriber = new MetaDataSubscriber() {
+ @Override
+ public void onSubscribe(final MetaData metaData) {
+ subscribeList.add(metaData);
+ }
+
+ @Override
+ public void unSubscribe(final MetaData metaData) {
+ unsubscribeList.add(metaData);
+ }
+ };
+ ConsulCacheHandler consulCacheHandler = new ConsulCacheHandler(null, Lists.newArrayList(metaDataSubscriber),
+ Collections.emptyList());
+ consulCacheHandler.updateMetaDataMap(metaDataParam);
+ Assert.assertEquals(2, subscribeList.size());
+ Assert.assertEquals(2, unsubscribeList.size());
+ }
+
+ @SneakyThrows
+ @Test
+ public void testUpdateAuthMap() {
+ String mockAppKey = "MOCK_APP_KEY";
+ String mockAppKey2 = "MOCK_APP_KEY2";
+ String mockAppSecret = "MOCK_APP_SECRET";
+ AppAuthData appAuthData =
+ AppAuthData.builder().appKey(mockAppKey).appSecret(mockAppSecret).enabled(true).build();
+ AppAuthData appAuthData2 =
+ AppAuthData.builder().appKey(mockAppKey2).appSecret(mockAppSecret).enabled(true).build();
+
+ String appAuthDataParam = GsonUtils.getInstance()
+ .toJson(ImmutableMap.of(mockAppKey2, appAuthData2, mockAppKey, appAuthData));
+ final List<AppAuthData> subscribeList = new ArrayList<>();
+ final List<AppAuthData> unsubscribeList = new ArrayList<>();
+
+ AuthDataSubscriber authDataSubscriber = new AuthDataSubscriber() {
+ @Override
+ public void onSubscribe(final AppAuthData appAuthData) {
+ subscribeList.add(appAuthData);
+ }
+
+ @Override
+ public void unSubscribe(final AppAuthData appAuthData) {
+ unsubscribeList.add(appAuthData);
+ }
+ };
+ ConsulCacheHandler consulCacheHandler = new ConsulCacheHandler(null,
+ Collections.emptyList(), Lists.newArrayList(authDataSubscriber));
+
+ consulCacheHandler.updateAuthMap(appAuthDataParam);
+ Assert.assertEquals(2, subscribeList.size());
+ Assert.assertEquals(2, unsubscribeList.size());
+ }
+}