This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 799dbfe93a refactor: extract shared discovery upstream data handler
base (#6605) (#7487)
799dbfe93a is described below
commit 799dbfe93a4ef3c29616cc2f39c17da1261e6253
Author: wy471x <[email protected]>
AuthorDate: Fri Oct 16 09:43:03 2026 +0800
refactor: extract shared discovery upstream data handler base (#6605)
(#7487)
- Add AbstractDiscoveryUpstreamDataHandler in shenyu-plugin-base
implementing
the common null-check -> convert -> gray-split -> submit flow, mirroring
the
proven AbstractDubboPluginDataHandler pattern.
- Add UpstreamProps, a typed deserializer that centralizes the warmup, gray
and
healthCheckEnabled property names and defaults so they cannot drift
between
plugins.
- Make the divide, websocket and grpc discovery upstream handlers extend the
base and reuse UpstreamProps instead of duplicating the conversion body
with
inline string literals.
- Cover the shared flow, the props parser and the websocket handler with
unit
tests, and extend the grpc handler test with the upstream data path.
---
.../AbstractDiscoveryUpstreamDataHandler.java | 72 ++++++++++
.../shenyu/plugin/base/utils/UpstreamProps.java | 103 +++++++++++++++
.../AbstractDiscoveryUpstreamDataHandlerTest.java | 135 +++++++++++++++++++
.../plugin/base/utils/UpstreamPropsTest.java | 63 +++++++++
.../divide/handler/DivideUpstreamDataHandler.java | 72 +++++-----
.../handler/GrpcDiscoveryUpstreamDataHandler.java | 58 ++++----
.../GrpcDiscoveryUpstreamDataHandlerTest.java | 33 ++++-
.../handler/WebSocketUpstreamDataHandler.java | 61 ++++-----
.../handler/WebSocketUpstreamDataHandlerTest.java | 147 +++++++++++++++++++++
9 files changed, 640 insertions(+), 104 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/handler/AbstractDiscoveryUpstreamDataHandler.java
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/handler/AbstractDiscoveryUpstreamDataHandler.java
new file mode 100644
index 0000000000..074bf8fc2a
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/handler/AbstractDiscoveryUpstreamDataHandler.java
@@ -0,0 +1,72 @@
+/*
+ * 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.plugin.base.handler;
+
+import org.apache.shenyu.common.dto.DiscoverySyncData;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+import org.apache.shenyu.sync.data.api.DiscoveryUpstreamKey;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * The common discovery upstream data handler.
+ *
+ * <p>Implements the shared null-check, gray-split and submit flow for
discovery upstream data,
+ * so that the concrete handlers only provide their upstream conversion and
cache access.
+ *
+ * @param <T> the upstream type
+ */
+public abstract class AbstractDiscoveryUpstreamDataHandler<T> implements
DiscoveryUpstreamDataHandler {
+
+ @Override
+ public void handlerDiscoveryUpstreamData(final DiscoverySyncData
discoverySyncData) {
+ if (Objects.isNull(discoverySyncData) ||
Objects.isNull(discoverySyncData.getSelectorId())) {
+ return;
+ }
+ final String selectorId = discoverySyncData.getSelectorId();
+ final List<T> upstreamList =
convertUpstreamList(discoverySyncData.getUpstreamDataList());
+ final List<T> grayUpstreamList =
upstreamList.stream().filter(this::isGray).collect(Collectors.toList());
+ if (grayUpstreamList.isEmpty()) {
+ submitUpstreamData(selectorId, upstreamList);
+ } else {
+ submitUpstreamData(selectorId, grayUpstreamList);
+ }
+ afterSubmitUpstreamData(selectorId);
+ }
+
+ @Override
+ public void removeDiscoveryUpstreamData(final DiscoveryUpstreamKey key) {
+ if (Objects.isNull(key) || Objects.isNull(key.selectorId())) {
+ return;
+ }
+ removeUpstreamData(key.selectorId());
+ }
+
+ protected abstract List<T> convertUpstreamList(List<DiscoveryUpstreamData>
upstreamList);
+
+ protected abstract boolean isGray(T upstream);
+
+ protected abstract void submitUpstreamData(String selectorId, List<T>
upstreamList);
+
+ protected abstract void removeUpstreamData(String selectorId);
+
+ protected void afterSubmitUpstreamData(final String selectorId) {
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/utils/UpstreamProps.java
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/utils/UpstreamProps.java
new file mode 100644
index 0000000000..4161bcc8ed
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/utils/UpstreamProps.java
@@ -0,0 +1,103 @@
+/*
+ * 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.plugin.base.utils;
+
+import org.apache.shenyu.common.utils.GsonUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Typed view over the raw props payload of a discovery upstream.
+ *
+ * <p>Centralizes the property names and defaults shared by the discovery
upstream handlers,
+ * so that warmup, gray and health check flags cannot drift between plugins.
+ */
+public final class UpstreamProps {
+
+ private static final String WARMUP = "warmup";
+
+ private static final String GRAY = "gray";
+
+ private static final String HEALTH_CHECK_ENABLED = "healthCheckEnabled";
+
+ private static final String DEFAULT_WARMUP = "10";
+
+ private static final String DEFAULT_GRAY = "false";
+
+ private static final String DEFAULT_HEALTH_CHECK_ENABLED = "true";
+
+ private final Map<String, String> props;
+
+ private UpstreamProps(final Map<String, String> props) {
+ this.props = props;
+ }
+
+ /**
+ * Parse the raw props payload of a discovery upstream.
+ *
+ * @param props the raw props json, may be null
+ * @return the parsed props
+ */
+ public static UpstreamProps parse(final String props) {
+ return new UpstreamProps(Optional.ofNullable(props)
+ .map(ps -> GsonUtils.getInstance().toObjectMap(ps,
String.class))
+ .orElseGet(HashMap::new));
+ }
+
+ /**
+ * Get the warmup, {@code 10} by default.
+ *
+ * @return the warmup
+ */
+ public int getWarmup() {
+ return Integer.parseInt(getProperty(WARMUP, DEFAULT_WARMUP));
+ }
+
+ /**
+ * Whether the upstream is a gray upstream, {@code false} by default.
+ *
+ * @return true if the upstream is gray
+ */
+ public boolean isGray() {
+ return Boolean.parseBoolean(getProperty(GRAY, DEFAULT_GRAY));
+ }
+
+ /**
+ * Whether the health check is enabled, {@code true} by default.
+ *
+ * @return true if the health check is enabled
+ */
+ public boolean isHealthCheckEnabled() {
+ return Boolean.parseBoolean(getProperty(HEALTH_CHECK_ENABLED,
DEFAULT_HEALTH_CHECK_ENABLED));
+ }
+
+ /**
+ * Get all parsed props as a mutable map.
+ *
+ * @return a copy of the parsed props
+ */
+ public Map<String, String> toMap() {
+ return new HashMap<>(props);
+ }
+
+ private String getProperty(final String key, final String defaultValue) {
+ return Optional.ofNullable(props.get(key)).orElse(defaultValue);
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/handler/AbstractDiscoveryUpstreamDataHandlerTest.java
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/handler/AbstractDiscoveryUpstreamDataHandlerTest.java
new file mode 100644
index 0000000000..bdb02774fb
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/handler/AbstractDiscoveryUpstreamDataHandlerTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.plugin.base.handler;
+
+import org.apache.shenyu.common.dto.DiscoverySyncData;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+import org.apache.shenyu.sync.data.api.DiscoveryUpstreamKey;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test case for {@link AbstractDiscoveryUpstreamDataHandler}.
+ */
+class AbstractDiscoveryUpstreamDataHandlerTest {
+
+ @Test
+ void testIgnoreNullDiscoverySyncData() {
+ final TestHandler handler = new TestHandler();
+
+ handler.handlerDiscoveryUpstreamData(null);
+ final DiscoverySyncData discoverySyncData = new DiscoverySyncData();
+ handler.handlerDiscoveryUpstreamData(discoverySyncData);
+
+ assertTrue(handler.submitted.isEmpty());
+ }
+
+ @Test
+ void testSubmitAllUpstreamsWhenNoGrayUpstream() {
+ final TestHandler handler = new TestHandler();
+
+ handler.handlerDiscoveryUpstreamData(discoverySyncData("a", "b"));
+
+ assertEquals(List.of("a", "b"), handler.submitted);
+ assertEquals(1, handler.afterSubmitCount);
+ }
+
+ @Test
+ void testSubmitOnlyGrayUpstreamsWhenPresent() {
+ final TestHandler handler = new TestHandler();
+ handler.grayUrl = "gray";
+
+ handler.handlerDiscoveryUpstreamData(discoverySyncData("gray",
"normal"));
+
+ assertEquals(List.of("gray"), handler.submitted);
+ }
+
+ @Test
+ void testRemoveDiscoveryUpstreamData() {
+ final TestHandler handler = new TestHandler();
+
+ handler.removeDiscoveryUpstreamData(null);
+ handler.removeDiscoveryUpstreamData(new DiscoveryUpstreamKey("test",
null, null));
+ assertTrue(handler.removed.isEmpty());
+
+ handler.removeDiscoveryUpstreamData(new DiscoveryUpstreamKey("test",
"selector", null));
+ assertEquals(List.of("selector"), handler.removed);
+ }
+
+ private DiscoverySyncData discoverySyncData(final String... urls) {
+ final DiscoverySyncData discoverySyncData = new DiscoverySyncData();
+ discoverySyncData.setSelectorId("selector");
+
discoverySyncData.setUpstreamDataList(Arrays.stream(urls).map(this::upstream).collect(Collectors.toList()));
+ return discoverySyncData;
+ }
+
+ private DiscoveryUpstreamData upstream(final String url) {
+ final DiscoveryUpstreamData upstream = new DiscoveryUpstreamData();
+ upstream.setUrl(url);
+ return upstream;
+ }
+
+ private static final class TestHandler extends
AbstractDiscoveryUpstreamDataHandler<String> {
+
+ private final List<String> submitted = new ArrayList<>();
+
+ private final List<String> removed = new ArrayList<>();
+
+ private String grayUrl;
+
+ private int afterSubmitCount;
+
+ @Override
+ protected List<String> convertUpstreamList(final
List<DiscoveryUpstreamData> upstreamList) {
+ return
upstreamList.stream().map(DiscoveryUpstreamData::getUrl).collect(Collectors.toList());
+ }
+
+ @Override
+ protected boolean isGray(final String upstream) {
+ return Objects.equals(grayUrl, upstream);
+ }
+
+ @Override
+ protected void submitUpstreamData(final String selectorId, final
List<String> upstreamList) {
+ submitted.addAll(upstreamList);
+ }
+
+ @Override
+ protected void afterSubmitUpstreamData(final String selectorId) {
+ afterSubmitCount++;
+ }
+
+ @Override
+ protected void removeUpstreamData(final String selectorId) {
+ removed.add(selectorId);
+ }
+
+ @Override
+ public String pluginName() {
+ return "test";
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/utils/UpstreamPropsTest.java
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/utils/UpstreamPropsTest.java
new file mode 100644
index 0000000000..4331e29947
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/utils/UpstreamPropsTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.plugin.base.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test case for {@link UpstreamProps}.
+ */
+class UpstreamPropsTest {
+
+ @Test
+ void testParseNullPropsAppliesDefaults() {
+ final UpstreamProps props = UpstreamProps.parse(null);
+
+ assertEquals(10, props.getWarmup());
+ assertFalse(props.isGray());
+ assertTrue(props.isHealthCheckEnabled());
+ assertTrue(props.toMap().isEmpty());
+ }
+
+ @Test
+ void testParsePropsOverridesDefaults() {
+ final UpstreamProps props =
UpstreamProps.parse("{\"warmup\":\"20\",\"gray\":\"true\",\"healthCheckEnabled\":\"false\"}");
+
+ assertEquals(20, props.getWarmup());
+ assertTrue(props.isGray());
+ assertFalse(props.isHealthCheckEnabled());
+ }
+
+ @Test
+ void testToMapReturnsCopyOfAllProps() {
+ final UpstreamProps props =
UpstreamProps.parse("{\"warmup\":\"20\",\"az\":\"az1\"}");
+ final Map<String, String> map = props.toMap();
+
+ assertEquals("20", map.get("warmup"));
+ assertEquals("az1", map.get("az"));
+
+ map.put("az", "az2");
+ assertEquals("az1", props.toMap().get("az"));
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/handler/DivideUpstreamDataHandler.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/handler/DivideUpstreamDataHandler.java
index 7617a83250..1413785cfc 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/handler/DivideUpstreamDataHandler.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/handler/DivideUpstreamDataHandler.java
@@ -17,78 +17,47 @@
package org.apache.shenyu.plugin.divide.handler;
-import org.apache.shenyu.common.dto.DiscoverySyncData;
import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
import org.apache.shenyu.common.enums.PluginEnum;
-import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.loadbalancer.cache.UpstreamCacheManager;
import org.apache.shenyu.loadbalancer.entity.Upstream;
import org.apache.shenyu.plugin.base.cache.MetaDataCache;
-import org.apache.shenyu.plugin.base.handler.DiscoveryUpstreamDataHandler;
-import org.apache.shenyu.sync.data.api.DiscoveryUpstreamKey;
+import
org.apache.shenyu.plugin.base.handler.AbstractDiscoveryUpstreamDataHandler;
+import org.apache.shenyu.plugin.base.utils.UpstreamProps;
import org.springframework.util.ObjectUtils;
import java.sql.Timestamp;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;
/**
* upstreamList data change.
*/
-public class DivideUpstreamDataHandler implements DiscoveryUpstreamDataHandler
{
-
- @Override
- public void handlerDiscoveryUpstreamData(final DiscoverySyncData
discoverySyncData) {
- if (Objects.isNull(discoverySyncData) ||
Objects.isNull(discoverySyncData.getSelectorId())) {
- return;
- }
- List<DiscoveryUpstreamData> upstreamList =
discoverySyncData.getUpstreamDataList();
- final List<Upstream> upstreams = convertUpstreamList(upstreamList);
- final List<Upstream> grayUpstreamList =
upstreams.stream().filter(Upstream::isGray).toList();
- if (!grayUpstreamList.isEmpty()) {
-
UpstreamCacheManager.getInstance().submit(discoverySyncData.getSelectorId(),
grayUpstreamList);
- } else {
-
UpstreamCacheManager.getInstance().submit(discoverySyncData.getSelectorId(),
upstreams);
- }
- // the update is also need to clean, but there is no way to
- // distinguish between crate and update, so it is always clean
- MetaDataCache.getInstance().clean();
- }
-
- @Override
- public void removeDiscoveryUpstreamData(final DiscoveryUpstreamKey key) {
- if (Objects.isNull(key) || Objects.isNull(key.selectorId())) {
- return;
- }
- UpstreamCacheManager.getInstance().removeByKey(key.selectorId());
- }
+public class DivideUpstreamDataHandler extends
AbstractDiscoveryUpstreamDataHandler<Upstream> {
@Override
public String pluginName() {
return PluginEnum.DIVIDE.getName();
}
- private List<Upstream> convertUpstreamList(final
List<DiscoveryUpstreamData> upstreamList) {
+ @Override
+ protected List<Upstream> convertUpstreamList(final
List<DiscoveryUpstreamData> upstreamList) {
if (ObjectUtils.isEmpty(upstreamList)) {
return Collections.emptyList();
}
return upstreamList.stream().map(u -> {
- Map<String, String> metadata = Optional.ofNullable(u.getProps())
- .map(ps -> GsonUtils.getInstance().toObjectMap(ps,
String.class))
- .map(HashMap::new)
- .orElseGet(HashMap::new);
+ UpstreamProps props = UpstreamProps.parse(u.getProps());
+ Map<String, String> metadata = props.toMap();
Upstream upstream = Upstream.builder()
.protocol(u.getProtocol())
.url(u.getUrl())
.weight(u.getWeight())
- .warmup(Integer.parseInt(metadata.getOrDefault("warmup",
"10")))
- .gray(Boolean.parseBoolean(metadata.getOrDefault("gray",
"false")))
-
.healthCheckEnabled(Boolean.parseBoolean(metadata.getOrDefault("healthCheckEnabled",
"true")))
+ .warmup(props.getWarmup())
+ .gray(props.isGray())
+ .healthCheckEnabled(props.isHealthCheckEnabled())
.status(0 == u.getStatus())
.timestamp(Optional.ofNullable(u.getDateCreated()).map(Timestamp::getTime).orElse(System.currentTimeMillis()))
.build();
@@ -97,4 +66,25 @@ public class DivideUpstreamDataHandler implements
DiscoveryUpstreamDataHandler {
}).collect(Collectors.toList());
}
+ @Override
+ protected boolean isGray(final Upstream upstream) {
+ return upstream.isGray();
+ }
+
+ @Override
+ protected void submitUpstreamData(final String selectorId, final
List<Upstream> upstreamList) {
+ UpstreamCacheManager.getInstance().submit(selectorId, upstreamList);
+ }
+
+ @Override
+ protected void afterSubmitUpstreamData(final String selectorId) {
+ // the update is also need to clean, but there is no way to
+ // distinguish between crate and update, so it is always clean
+ MetaDataCache.getInstance().clean();
+ }
+
+ @Override
+ protected void removeUpstreamData(final String selectorId) {
+ UpstreamCacheManager.getInstance().removeByKey(selectorId);
+ }
}
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-grpc/src/main/java/org/apache/shenyu/plugin/grpc/handler/GrpcDiscoveryUpstreamDataHandler.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-grpc/src/main/java/org/apache/shenyu/plugin/grpc/handler/GrpcDiscoveryUpstreamDataHandler.java
index 9af925d983..bbfb6c017c 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-grpc/src/main/java/org/apache/shenyu/plugin/grpc/handler/GrpcDiscoveryUpstreamDataHandler.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-grpc/src/main/java/org/apache/shenyu/plugin/grpc/handler/GrpcDiscoveryUpstreamDataHandler.java
@@ -21,12 +21,11 @@ import org.apache.shenyu.common.dto.DiscoverySyncData;
import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
import org.apache.shenyu.common.dto.convert.selector.GrpcUpstream;
import org.apache.shenyu.common.enums.PluginEnum;
-import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.common.utils.JsonUtils;
-import org.apache.shenyu.plugin.base.handler.DiscoveryUpstreamDataHandler;
+import
org.apache.shenyu.plugin.base.handler.AbstractDiscoveryUpstreamDataHandler;
+import org.apache.shenyu.plugin.base.utils.UpstreamProps;
import org.apache.shenyu.plugin.grpc.cache.ApplicationConfigCache;
import org.apache.shenyu.plugin.grpc.cache.GrpcClientCache;
-import org.apache.shenyu.sync.data.api.DiscoveryUpstreamKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.ObjectUtils;
@@ -36,60 +35,63 @@ import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
-import java.util.Properties;
import java.util.stream.Collectors;
/**
* GrpcDiscoveryUpstreamDataHandler.
*/
-public class GrpcDiscoveryUpstreamDataHandler implements
DiscoveryUpstreamDataHandler {
-
+public class GrpcDiscoveryUpstreamDataHandler extends
AbstractDiscoveryUpstreamDataHandler<GrpcUpstream> {
+
private static final Logger LOG =
LoggerFactory.getLogger(GrpcDiscoveryUpstreamDataHandler.class);
@Override
public void handlerDiscoveryUpstreamData(final DiscoverySyncData
discoverySyncData) {
- if (Objects.isNull(discoverySyncData) ||
Objects.isNull(discoverySyncData.getSelectorId())) {
- return;
- }
- LOG.info("discovery grpc upstream data:{}",
JsonUtils.toJson(discoverySyncData));
- final String selectorId = discoverySyncData.getSelectorId();
- final List<GrpcUpstream> upstreams =
convertUpstreamList(discoverySyncData.getUpstreamDataList());
- final List<GrpcUpstream> grayUpstreamList =
upstreams.stream().filter(GrpcUpstream::isGray).toList();
- if (!grayUpstreamList.isEmpty()) {
-
ApplicationConfigCache.getInstance().handlerUpstream(discoverySyncData.getSelectorId(),
grayUpstreamList);
- } else {
-
ApplicationConfigCache.getInstance().handlerUpstream(discoverySyncData.getSelectorId(),
upstreams);
+ if (Objects.nonNull(discoverySyncData) &&
Objects.nonNull(discoverySyncData.getSelectorId())) {
+ LOG.info("discovery grpc upstream data:{}",
JsonUtils.toJson(discoverySyncData));
}
- GrpcClientCache.initGrpcClient(selectorId);
+ super.handlerDiscoveryUpstreamData(discoverySyncData);
}
@Override
- public void removeDiscoveryUpstreamData(final DiscoveryUpstreamKey key) {
- if (Objects.isNull(key) || Objects.isNull(key.selectorId())) {
- return;
- }
- ApplicationConfigCache.getInstance().invalidate(key.selectorId());
+ public String pluginName() {
+ return PluginEnum.GRPC.getName();
}
- private List<GrpcUpstream> convertUpstreamList(final
List<DiscoveryUpstreamData> upstreamList) {
+ @Override
+ protected List<GrpcUpstream> convertUpstreamList(final
List<DiscoveryUpstreamData> upstreamList) {
if (ObjectUtils.isEmpty(upstreamList)) {
return Collections.emptyList();
}
return upstreamList.stream().map(u -> {
- Properties properties = Optional.ofNullable(u.getProps()).map(ps
-> GsonUtils.getInstance().fromJson(ps, Properties.class)).orElse(new
Properties());
+ UpstreamProps props = UpstreamProps.parse(u.getProps());
return GrpcUpstream.builder()
.protocol(u.getProtocol())
.upstreamUrl(u.getUrl())
.weight(u.getWeight())
.status(0 == u.getStatus())
.timestamp(Optional.ofNullable(u.getDateCreated()).map(Timestamp::getTime).orElse(System.currentTimeMillis()))
-
.healthCheckEnabled(Boolean.parseBoolean(properties.getProperty("healthCheckEnabled",
"true")))
+ .healthCheckEnabled(props.isHealthCheckEnabled())
.build();
}).collect(Collectors.toList());
}
@Override
- public String pluginName() {
- return PluginEnum.GRPC.getName();
+ protected boolean isGray(final GrpcUpstream upstream) {
+ return upstream.isGray();
+ }
+
+ @Override
+ protected void submitUpstreamData(final String selectorId, final
List<GrpcUpstream> upstreamList) {
+ ApplicationConfigCache.getInstance().handlerUpstream(selectorId,
upstreamList);
+ }
+
+ @Override
+ protected void afterSubmitUpstreamData(final String selectorId) {
+ GrpcClientCache.initGrpcClient(selectorId);
+ }
+
+ @Override
+ protected void removeUpstreamData(final String selectorId) {
+ ApplicationConfigCache.getInstance().invalidate(selectorId);
}
}
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-grpc/src/test/java/org/apache/shenyu/plugin/grpc/handler/GrpcDiscoveryUpstreamDataHandlerTest.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-grpc/src/test/java/org/apache/shenyu/plugin/grpc/handler/GrpcDiscoveryUpstreamDataHandlerTest.java
index 90b8421281..db886a9001 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-grpc/src/test/java/org/apache/shenyu/plugin/grpc/handler/GrpcDiscoveryUpstreamDataHandlerTest.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-grpc/src/test/java/org/apache/shenyu/plugin/grpc/handler/GrpcDiscoveryUpstreamDataHandlerTest.java
@@ -17,11 +17,20 @@
package org.apache.shenyu.plugin.grpc.handler;
+import org.apache.shenyu.common.dto.DiscoverySyncData;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+import org.apache.shenyu.common.dto.convert.selector.GrpcUpstream;
import org.apache.shenyu.sync.data.api.DiscoveryUpstreamKey;
+import org.apache.shenyu.plugin.grpc.cache.ApplicationConfigCache;
import org.apache.shenyu.plugin.grpc.cache.GrpcClientCache;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -31,7 +40,29 @@ public final class GrpcDiscoveryUpstreamDataHandlerTest {
@AfterEach
public void tearDown() {
- GrpcClientCache.removeClient(SELECTOR_ID);
+ ApplicationConfigCache.getInstance().invalidate(SELECTOR_ID);
+ }
+
+ @Test
+ public void testHandlerDiscoveryUpstreamData() {
+ DiscoveryUpstreamData discoveryUpstreamData =
DiscoveryUpstreamData.builder()
+ .protocol("grpc://")
+ .url("127.0.0.1:9090")
+ .weight(100)
+ .status(0)
+ .props("{\"healthCheckEnabled\":\"false\"}")
+ .build();
+ DiscoverySyncData discoverySyncData = new DiscoverySyncData();
+ discoverySyncData.setSelectorId(SELECTOR_ID);
+
discoverySyncData.setUpstreamDataList(Collections.singletonList(discoveryUpstreamData));
+
+ new
GrpcDiscoveryUpstreamDataHandler().handlerDiscoveryUpstreamData(discoverySyncData);
+
+ List<GrpcUpstream> upstreamList =
ApplicationConfigCache.getInstance().getGrpcUpstreamListCache(SELECTOR_ID);
+ assertEquals(1, upstreamList.size());
+ assertEquals("grpc://", upstreamList.get(0).getProtocol());
+ assertFalse(upstreamList.get(0).isHealthCheckEnabled());
+ assertNotNull(GrpcClientCache.getGrpcClient(SELECTOR_ID));
}
@Test
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-websocket/src/main/java/org/apache/shenyu/plugin/websocket/handler/WebSocketUpstreamDataHandler.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-websocket/src/main/java/org/apache/shenyu/plugin/websocket/handler/WebSocketUpstreamDataHandler.java
index 5619f18e52..b0285ec7da 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-websocket/src/main/java/org/apache/shenyu/plugin/websocket/handler/WebSocketUpstreamDataHandler.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-websocket/src/main/java/org/apache/shenyu/plugin/websocket/handler/WebSocketUpstreamDataHandler.java
@@ -17,66 +17,44 @@
package org.apache.shenyu.plugin.websocket.handler;
-import org.apache.shenyu.common.dto.DiscoverySyncData;
import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
import org.apache.shenyu.common.enums.PluginEnum;
-import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.loadbalancer.cache.UpstreamCacheManager;
import org.apache.shenyu.loadbalancer.entity.Upstream;
import org.apache.shenyu.plugin.base.cache.MetaDataCache;
-import org.apache.shenyu.plugin.base.handler.DiscoveryUpstreamDataHandler;
-import org.apache.shenyu.sync.data.api.DiscoveryUpstreamKey;
+import
org.apache.shenyu.plugin.base.handler.AbstractDiscoveryUpstreamDataHandler;
+import org.apache.shenyu.plugin.base.utils.UpstreamProps;
import org.springframework.util.ObjectUtils;
import java.sql.Timestamp;
import java.util.Collections;
import java.util.List;
-import java.util.Objects;
import java.util.Optional;
-import java.util.Properties;
import java.util.stream.Collectors;
-/*
+/**
* upstreamList data change.
*/
-public class WebSocketUpstreamDataHandler implements
DiscoveryUpstreamDataHandler {
+public class WebSocketUpstreamDataHandler extends
AbstractDiscoveryUpstreamDataHandler<Upstream> {
@Override
- public void handlerDiscoveryUpstreamData(final DiscoverySyncData
discoverySyncData) {
- if (Objects.isNull(discoverySyncData) ||
Objects.isNull(discoverySyncData.getSelectorId())) {
- return;
- }
- List<DiscoveryUpstreamData> upstreamList =
discoverySyncData.getUpstreamDataList();
- final List<Upstream> upstreams = convertUpstreamList(upstreamList);
- final List<Upstream> grayUpstreamList =
upstreams.stream().filter(Upstream::isGray).toList();
- if (!grayUpstreamList.isEmpty()) {
-
UpstreamCacheManager.getInstance().submit(discoverySyncData.getSelectorId(),
grayUpstreamList);
- } else {
-
UpstreamCacheManager.getInstance().submit(discoverySyncData.getSelectorId(),
upstreams);
- }
- MetaDataCache.getInstance().clean();
+ public String pluginName() {
+ return PluginEnum.WEB_SOCKET.getName();
}
@Override
- public void removeDiscoveryUpstreamData(final DiscoveryUpstreamKey key) {
- if (Objects.isNull(key) || Objects.isNull(key.selectorId())) {
- return;
- }
- UpstreamCacheManager.getInstance().removeByKey(key.selectorId());
- }
-
- private List<Upstream> convertUpstreamList(final
List<DiscoveryUpstreamData> upstreamList) {
+ protected List<Upstream> convertUpstreamList(final
List<DiscoveryUpstreamData> upstreamList) {
if (ObjectUtils.isEmpty(upstreamList)) {
return Collections.emptyList();
}
return upstreamList.stream().map(u -> {
- Properties properties = Optional.ofNullable(u.getProps()).map(ps
-> GsonUtils.getInstance().fromJson(ps, Properties.class)).orElse(new
Properties());
+ UpstreamProps props = UpstreamProps.parse(u.getProps());
return Upstream.builder()
.protocol(u.getProtocol())
.url(u.getUrl())
.weight(u.getWeight())
- .warmup(Integer.parseInt(properties.getProperty("warmup",
"10")))
-
.healthCheckEnabled(Boolean.parseBoolean(properties.getProperty("healthCheckEnabled",
"true")))
+ .warmup(props.getWarmup())
+ .healthCheckEnabled(props.isHealthCheckEnabled())
.status(0 == u.getStatus())
.timestamp(Optional.ofNullable(u.getDateCreated()).map(Timestamp::getTime).orElse(System.currentTimeMillis()))
.build();
@@ -84,7 +62,22 @@ public class WebSocketUpstreamDataHandler implements
DiscoveryUpstreamDataHandle
}
@Override
- public String pluginName() {
- return PluginEnum.WEB_SOCKET.getName();
+ protected boolean isGray(final Upstream upstream) {
+ return upstream.isGray();
+ }
+
+ @Override
+ protected void submitUpstreamData(final String selectorId, final
List<Upstream> upstreamList) {
+ UpstreamCacheManager.getInstance().submit(selectorId, upstreamList);
+ }
+
+ @Override
+ protected void afterSubmitUpstreamData(final String selectorId) {
+ MetaDataCache.getInstance().clean();
+ }
+
+ @Override
+ protected void removeUpstreamData(final String selectorId) {
+ UpstreamCacheManager.getInstance().removeByKey(selectorId);
}
}
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-websocket/src/test/java/org/apache/shenyu/plugin/websocket/handler/WebSocketUpstreamDataHandlerTest.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-websocket/src/test/java/org/apache/shenyu/plugin/websocket/handler/WebSocketUpstreamDataHandlerTest.java
new file mode 100644
index 0000000000..835c6abaac
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-websocket/src/test/java/org/apache/shenyu/plugin/websocket/handler/WebSocketUpstreamDataHandlerTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.plugin.websocket.handler;
+
+import org.apache.shenyu.common.dto.DiscoverySyncData;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+import org.apache.shenyu.common.enums.PluginEnum;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.common.utils.UpstreamCheckUtils;
+import org.apache.shenyu.loadbalancer.cache.UpstreamCacheManager;
+import org.apache.shenyu.loadbalancer.entity.Upstream;
+import org.apache.shenyu.sync.data.api.DiscoveryUpstreamKey;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import java.sql.Timestamp;
+import java.util.List;
+import java.util.Properties;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test case for {@link WebSocketUpstreamDataHandler}.
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class WebSocketUpstreamDataHandlerTest {
+
+ private DiscoverySyncData discoverySyncData;
+
+ private WebSocketUpstreamDataHandler webSocketUpstreamDataHandler;
+
+ private MockedStatic<UpstreamCheckUtils> mockCheckUtils;
+
+ @BeforeEach
+ public void setUp() {
+ this.webSocketUpstreamDataHandler = new WebSocketUpstreamDataHandler();
+ List<DiscoveryUpstreamData> upstreamList = Stream.of(3)
+ .map(weight -> DiscoveryUpstreamData.builder()
+ .url("mock-" + weight)
+ .dateUpdated(new Timestamp(System.currentTimeMillis()))
+ .build())
+ .collect(Collectors.toList());
+ this.discoverySyncData = mock(DiscoverySyncData.class);
+ when(discoverySyncData.getSelectorId()).thenReturn("handler");
+ when(discoverySyncData.getUpstreamDataList()).thenReturn(upstreamList);
+
+ // mock static
+ mockCheckUtils = mockStatic(UpstreamCheckUtils.class);
+ mockCheckUtils.when(() -> UpstreamCheckUtils.checkUrl(anyString(),
anyInt())).thenReturn(true);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ UpstreamCacheManager.getInstance().removeByKey("handler");
+ mockCheckUtils.close();
+ }
+
+ /**
+ * Handler discovery upstream data test.
+ */
+ @Test
+ public void handlerDiscoveryUpstreamDataTest() {
+
webSocketUpstreamDataHandler.handlerDiscoveryUpstreamData(discoverySyncData);
+ List<Upstream> result =
UpstreamCacheManager.getInstance().findUpstreamListBySelectorId("handler");
+ assertEquals(discoverySyncData.getUpstreamDataList().get(0).getUrl(),
result.get(0).getUrl());
+ DiscoverySyncData emptySyncData = new DiscoverySyncData();
+ emptySyncData.setSelectorId(null);
+
webSocketUpstreamDataHandler.handlerDiscoveryUpstreamData(emptySyncData);
+ }
+
+ /**
+ * Handler discovery upstream props test.
+ */
+ @Test
+ public void handlerDiscoveryUpstreamDataWithPropsTest() {
+ Properties properties = new Properties();
+ properties.setProperty("warmup", "20");
+ properties.setProperty("healthCheckEnabled", "false");
+ DiscoveryUpstreamData discoveryUpstreamData =
DiscoveryUpstreamData.builder()
+ .url("mock-props")
+ .status(0)
+ .weight(10)
+ .props(GsonUtils.getInstance().toJson(properties))
+ .dateCreated(new Timestamp(System.currentTimeMillis()))
+ .build();
+
when(discoverySyncData.getUpstreamDataList()).thenReturn(List.of(discoveryUpstreamData));
+
+
webSocketUpstreamDataHandler.handlerDiscoveryUpstreamData(discoverySyncData);
+
+ Upstream upstream =
UpstreamCacheManager.getInstance().findUpstreamListBySelectorId("handler").get(0);
+ assertEquals(20, upstream.getWarmup());
+ assertFalse(upstream.isHealthCheckEnabled());
+ }
+
+ /**
+ * Remove discovery upstream data test.
+ */
+ @Test
+ public void removeDiscoveryUpstreamDataTest() {
+
webSocketUpstreamDataHandler.handlerDiscoveryUpstreamData(discoverySyncData);
+
assertNotNull(UpstreamCacheManager.getInstance().findUpstreamListBySelectorId("handler"));
+
+
webSocketUpstreamDataHandler.removeDiscoveryUpstreamData(DiscoveryUpstreamKey.from(discoverySyncData));
+
+
assertNull(UpstreamCacheManager.getInstance().findUpstreamListBySelectorId("handler"));
+ }
+
+ /**
+ * Plugin name test.
+ */
+ @Test
+ public void pluginNameTest() {
+ assertEquals(PluginEnum.WEB_SOCKET.getName(),
webSocketUpstreamDataHandler.pluginName());
+ }
+}
