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 1e10196da9 feat(admin): opt in to compact WebSocket discovery payloads
(#7883)
1e10196da9 is described below
commit 1e10196da99c1e6d3e2f93df83c515d79b25e1ad
Author: cloudbase-ai <[email protected]>
AuthorDate: Fri Oct 23 11:29:20 2026 +0800
feat(admin): opt in to compact WebSocket discovery payloads (#7883)
Signed-off-by: cloudbase-ai <[email protected]>
Co-authored-by: cloudbase-ai
<[email protected]>
---
shenyu-admin/README.md | 19 +++
.../admin/config/WebSocketSyncConfiguration.java | 5 +-
.../config/properties/WebsocketSyncProperties.java | 23 ++++
.../websocket/WebsocketDataChangedListener.java | 29 +++++
.../admin/model/dto/DiscoverySyncPayload.java | 70 ++++++++++
shenyu-admin/src/main/resources/application.yml | 2 +
.../config/WebSocketSyncConfigurationTest.java | 3 +-
.../properties/WebsocketSyncPropertiesTest.java | 5 +-
.../WebsocketDataChangedListenerTest.java | 143 +++++++++++++++++++++
9 files changed, 295 insertions(+), 4 deletions(-)
diff --git a/shenyu-admin/README.md b/shenyu-admin/README.md
index 3f9ba6958e..f878ac9d2e 100644
--- a/shenyu-admin/README.md
+++ b/shenyu-admin/README.md
@@ -1,3 +1,22 @@
# ShenYu Admin API Document
You can check http://localhost:9095/swagger-ui.html to check all restful api
and models.
+
+## Compact Discovery messages (WebSocket only)
+
+`shenyu.sync.websocket.compact-discovery-payload` defaults to `false`.
Enabling it projects
+Discovery messages at the Admin-to-gateway serialization boundary. It omits
the outer
+`discoveryHandlerId` and each upstream's `id`, `dateUpdated`,
`discoveryHandlerId`, and
+repeated `namespaceId`. The enclosing namespace and selector identities,
creation timestamp
+(for warmup), protocol, URL, manual status, weight, and opaque `props` are
preserved.
+Empty upstream lists retain their clearing semantics. This is a full runtime
item, not a patch.
+
+Audit every gateway consuming that Admin before enabling it, including old
versions,
+external plugins and WASM guests: they may use fields that built-in handlers
do not read.
+There is no per-session capability negotiation in this first, opt-in WebSocket
implementation.
+Leave the option disabled for unaudited or mixed fleets. Disable it and
resynchronize/reconnect
+consumers to return to the legacy payload. Configure all Admin nodes
consistently.
+
+This option does not change internal Admin events, cluster forwarding, HTTP
polling,
+registry-backed synchronization, selectors/rules, or the WebSocket envelope
and initial-sync
+completion protocol. No extra JSON parse/delete/serialize pass is used to
build compact messages.
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/WebSocketSyncConfiguration.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/WebSocketSyncConfiguration.java
index b58aea5ba9..af85c7f2f4 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/WebSocketSyncConfiguration.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/WebSocketSyncConfiguration.java
@@ -51,12 +51,13 @@ public class WebSocketSyncConfiguration {
/**
* Config event listener data changed listener.
*
+ * @param properties WebSocket sync properties
* @return the data changed listener
*/
@Bean
@ConditionalOnMissingBean(WebsocketDataChangedListener.class)
- public DataChangedListener websocketDataChangedListener() {
- return new WebsocketDataChangedListener();
+ public DataChangedListener websocketDataChangedListener(final
WebsocketSyncProperties properties) {
+ return new
WebsocketDataChangedListener(properties.isCompactDiscoveryPayload());
}
/**
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/WebsocketSyncProperties.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/WebsocketSyncProperties.java
index 897c8429f6..7261299ca1 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/WebsocketSyncProperties.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/WebsocketSyncProperties.java
@@ -47,8 +47,31 @@ public class WebsocketSyncProperties {
*/
private String token;
+ /**
+ * Omit Discovery persistence fields from gateway messages. Opt in only
after auditing all consumers.
+ */
+ private boolean compactDiscoveryPayload;
+
private final Reconciliation reconciliation = new Reconciliation();
+ /**
+ * Whether Discovery messages use the gateway-only projection.
+ *
+ * @return whether compact messages are enabled
+ */
+ public boolean isCompactDiscoveryPayload() {
+ return compactDiscoveryPayload;
+ }
+
+ /**
+ * Enable the Discovery projection for WebSocket consumers.
+ *
+ * @param compactDiscoveryPayload whether to omit persistence fields
+ */
+ public void setCompactDiscoveryPayload(final boolean
compactDiscoveryPayload) {
+ this.compactDiscoveryPayload = compactDiscoveryPayload;
+ }
+
/**
* Gets the value of enabled.
*
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketDataChangedListener.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketDataChangedListener.java
index a034c62677..9e2b600265 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketDataChangedListener.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketDataChangedListener.java
@@ -22,6 +22,7 @@ import static
org.apache.shenyu.common.constant.Constants.SYS_DEFAULT_NAMESPACE_
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.shenyu.admin.listener.DataChangedListener;
+import org.apache.shenyu.admin.model.dto.DiscoverySyncPayload;
import org.apache.shenyu.common.dto.AppAuthData;
import org.apache.shenyu.common.dto.DiscoverySyncData;
import org.apache.shenyu.common.dto.MetaData;
@@ -54,6 +55,24 @@ public class WebsocketDataChangedListener implements
DataChangedListener {
private static final Logger LOG =
LoggerFactory.getLogger(WebsocketDataChangedListener.class);
+ private final boolean compactDiscoveryPayload;
+
+ /**
+ * Create a listener using the legacy payload.
+ */
+ public WebsocketDataChangedListener() {
+ this(false);
+ }
+
+ /**
+ * Create a listener with optional Discovery payload projection.
+ *
+ * @param compactDiscoveryPayload whether to omit Discovery management
fields
+ */
+ public WebsocketDataChangedListener(final boolean compactDiscoveryPayload)
{
+ this.compactDiscoveryPayload = compactDiscoveryPayload;
+ }
+
@Override
public void onPluginChanged(final List<PluginData> pluginDataList, final
DataEventTypeEnum eventType) {
sendByNamespace(pluginDataList, eventType, ConfigGroupEnum.PLUGIN,
PluginData::getNamespaceId);
@@ -142,6 +161,16 @@ public class WebsocketDataChangedListener implements
DataChangedListener {
@Override
public void onDiscoveryUpstreamChanged(final List<DiscoverySyncData>
discoveryUpstreamList, final DataEventTypeEnum eventType) {
+ if (compactDiscoveryPayload &&
CollectionUtils.isNotEmpty(discoveryUpstreamList)) {
+ List<DiscoverySyncPayload> payloads = new
ArrayList<>(discoveryUpstreamList.size());
+ for (DiscoverySyncData data : discoveryUpstreamList) {
+ if (Objects.nonNull(data)) {
+ payloads.add(DiscoverySyncPayload.from(data));
+ }
+ }
+ sendByNamespace(payloads, eventType,
ConfigGroupEnum.DISCOVER_UPSTREAM, DiscoverySyncPayload::namespaceId);
+ return;
+ }
sendByNamespace(discoveryUpstreamList, eventType,
ConfigGroupEnum.DISCOVER_UPSTREAM, DiscoverySyncData::getNamespaceId);
}
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/model/dto/DiscoverySyncPayload.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/model/dto/DiscoverySyncPayload.java
new file mode 100644
index 0000000000..277c07923f
--- /dev/null
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/model/dto/DiscoverySyncPayload.java
@@ -0,0 +1,70 @@
+/*
+ * 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.model.dto;
+
+import org.apache.shenyu.common.dto.DiscoverySyncData;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+
+import java.sql.Timestamp;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * Gateway-only Discovery projection. Persistence identifiers remain in the
Admin event.
+ *
+ * @param selectorId selector identity
+ * @param pluginName plugin handler identity
+ * @param selectorName selector name used by TCP routing
+ * @param namespaceId namespace identity
+ * @param upstreamDataList runtime upstream configuration
+ */
+public record DiscoverySyncPayload(String selectorId, String pluginName,
String selectorName,
+ String namespaceId, List<UpstreamPayload>
upstreamDataList) {
+
+ /**
+ * Project without mutating the shared event or interpreting
plugin-specific props.
+ *
+ * @param data source Discovery data
+ * @return gateway payload
+ */
+ public static DiscoverySyncPayload from(final DiscoverySyncData data) {
+ List<UpstreamPayload> upstreams =
Objects.isNull(data.getUpstreamDataList()) ? null
+ :
data.getUpstreamDataList().stream().map(UpstreamPayload::from).collect(Collectors.toList());
+ return new DiscoverySyncPayload(data.getSelectorId(),
data.getPluginName(), data.getSelectorName(),
+ data.getNamespaceId(), upstreams);
+ }
+
+ /**
+ * Runtime fields understood by the built-in gateway handlers.
+ *
+ * @param dateCreated creation timestamp used for load-balancer warmup
+ * @param protocol protocol prefix
+ * @param url upstream address
+ * @param status manual health status
+ * @param weight load-balancing weight
+ * @param props opaque plugin metadata
+ */
+ public record UpstreamPayload(Timestamp dateCreated, String protocol,
String url, int status, int weight, String props) {
+
+ private static UpstreamPayload from(final DiscoveryUpstreamData data) {
+ return Objects.isNull(data) ? null : new
UpstreamPayload(data.getDateCreated(), data.getProtocol(),
+ data.getUrl(), data.getStatus(), data.getWeight(),
data.getProps());
+ }
+ }
+}
diff --git a/shenyu-admin/src/main/resources/application.yml
b/shenyu-admin/src/main/resources/application.yml
index c7cf7fe0ef..fc139629a6 100755
--- a/shenyu-admin/src/main/resources/application.yml
+++ b/shenyu-admin/src/main/resources/application.yml
@@ -70,6 +70,8 @@ shenyu:
websocket:
enabled: true
messageMaxSize: 10240
+ # Opt in only after checking built-in and external gateway consumers.
See shenyu-admin/README.md.
+ compact-discovery-payload: false
token: ${SHENYU_SYNC_WEBSOCKET_TOKEN:}
allowOrigins: ws://localhost:9095;ws://localhost:9195;
# Reconciles gateway sessions connected to this admin node with the
shared
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/WebSocketSyncConfigurationTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/WebSocketSyncConfigurationTest.java
index 0ef0be9f58..e1d81217a9 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/WebSocketSyncConfigurationTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/WebSocketSyncConfigurationTest.java
@@ -17,6 +17,7 @@
package org.apache.shenyu.admin.config;
+import org.apache.shenyu.admin.config.properties.WebsocketSyncProperties;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -26,7 +27,7 @@ public class WebSocketSyncConfigurationTest {
@Test
public void testWebsocketDataChangedListener() {
WebSocketSyncConfiguration websocketListener = new
WebSocketSyncConfiguration();
- assertNotNull(websocketListener.websocketDataChangedListener());
+ assertNotNull(websocketListener.websocketDataChangedListener(new
WebsocketSyncProperties()));
}
@Test
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/WebsocketSyncPropertiesTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/WebsocketSyncPropertiesTest.java
index e6b7f844b6..ec86b35626 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/WebsocketSyncPropertiesTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/WebsocketSyncPropertiesTest.java
@@ -35,11 +35,13 @@ public final class WebsocketSyncPropertiesTest extends
AbstractConfigurationTest
@Test
public void testWebsocketSyncPropertiesDefaultValue() {
assertThat(new WebsocketSyncProperties().isEnabled(), is(true));
+ Assertions.assertFalse(new
WebsocketSyncProperties().isCompactDiscoveryPayload());
}
@Test
public void testWebsocketSyncPropertiesSetValue() {
- load(WebsocketSyncPropertiesConfiguration.class,
"shenyu.sync.websocket.enabled=false");
+ load(WebsocketSyncPropertiesConfiguration.class,
"shenyu.sync.websocket.enabled=false",
+ "shenyu.sync.websocket.compact-discovery-payload=true");
WebsocketSyncProperties websocketSyncProperties =
getContext().getBean(WebsocketSyncProperties.class);
websocketSyncProperties.setMessageMaxSize(0);
websocketSyncProperties.setAllowOrigins("allowOrigins");
@@ -48,6 +50,7 @@ public final class WebsocketSyncPropertiesTest extends
AbstractConfigurationTest
Assertions.assertEquals(websocketSyncProperties.getMessageMaxSize(),
0);
Assertions.assertEquals(websocketSyncProperties.getAllowOrigins(),
"allowOrigins");
Assertions.assertEquals(websocketSyncProperties.getToken(), "token");
+
Assertions.assertTrue(websocketSyncProperties.isCompactDiscoveryPayload());
}
@Configuration
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketDataChangedListenerTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketDataChangedListenerTest.java
index 4b0ccfdf60..165a0ff545 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketDataChangedListenerTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketDataChangedListenerTest.java
@@ -18,13 +18,18 @@
package org.apache.shenyu.admin.listener.websocket;
import com.fasterxml.jackson.databind.JsonNode;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.shenyu.admin.config.WebSocketSyncConfiguration;
+import org.apache.shenyu.admin.config.properties.WebsocketSyncProperties;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.dto.AppAuthData;
import org.apache.shenyu.common.dto.AuthParamData;
import org.apache.shenyu.common.dto.AuthPathData;
import org.apache.shenyu.common.dto.ConditionData;
import org.apache.shenyu.common.dto.DiscoverySyncData;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
import org.apache.shenyu.common.dto.MetaData;
import org.apache.shenyu.common.dto.PluginData;
import org.apache.shenyu.common.dto.ProxyApiKeyData;
@@ -32,17 +37,24 @@ import org.apache.shenyu.common.dto.ProxySelectorData;
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 org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.MockedStatic;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.stubbing.Answer;
+import java.sql.Timestamp;
import java.util.ArrayList;
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.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.argThat;
@@ -344,6 +356,137 @@ public final class WebsocketDataChangedListenerTest {
}
}
+ @ParameterizedTest
+ @EnumSource(value = DataEventTypeEnum.class, names = {"UPDATE", "CREATE",
"DELETE", "MYSELF", "REFRESH"})
+ public void testCompactDiscoveryPreservesRuntimeData(final
DataEventTypeEnum eventType) {
+ DiscoverySyncData data = discoveryData();
+ final String original = GsonUtils.getInstance().toJson(data);
+ final String legacy = discoveryMessage(new
WebsocketDataChangedListener(), data, eventType);
+ WebsocketSyncProperties properties = new WebsocketSyncProperties();
+ properties.setCompactDiscoveryPayload(true);
+ WebsocketDataChangedListener configured =
(WebsocketDataChangedListener) new
WebSocketSyncConfiguration().websocketDataChangedListener(properties);
+ String compact = discoveryMessage(configured, data, eventType);
+ JsonObject compactFrame =
JsonParser.parseString(compact).getAsJsonObject();
+ JsonObject compactData =
compactFrame.getAsJsonArray("data").get(0).getAsJsonObject();
+ JsonObject compactUpstream =
compactData.getAsJsonArray("upstreamDataList").get(0).getAsJsonObject();
+ assertEquals(eventType.name(),
compactFrame.get("eventType").getAsString());
+ assertEquals("DISCOVER_UPSTREAM",
compactFrame.get("groupType").getAsString());
+ assertFalse(compactData.has("discoveryHandlerId"));
+ for (String field : List.of("id", "dateUpdated", "discoveryHandlerId",
"namespaceId")) {
+ assertFalse(compactUpstream.has(field), field);
+ }
+ DiscoverySyncData decoded =
GsonUtils.getInstance().fromJson(compactData.toString(),
DiscoverySyncData.class);
+ assertEquals(data.getSelectorId(), decoded.getSelectorId());
+ assertEquals(data.getSelectorName(), decoded.getSelectorName());
+ assertEquals(data.getPluginName(), decoded.getPluginName());
+ assertEquals(data.getNamespaceId(), decoded.getNamespaceId());
+ DiscoveryUpstreamData expected = data.getUpstreamDataList().get(0);
+ DiscoveryUpstreamData actual = decoded.getUpstreamDataList().get(0);
+ assertEquals(expected.getDateCreated(), actual.getDateCreated());
+ assertEquals(expected.getProtocol(), actual.getProtocol());
+ assertEquals(expected.getUrl(), actual.getUrl());
+ assertEquals(expected.getStatus(), actual.getStatus());
+ assertEquals(expected.getWeight(), actual.getWeight());
+ assertEquals(expected.getProps(), actual.getProps());
+ assertEquals(original, GsonUtils.getInstance().toJson(data));
+ assertTrue(legacy.contains("dateUpdated"));
+ assertTrue(compact.length() < legacy.length());
+ }
+
+ @Test
+ public void testCompactDiscoveryRetainsEmptyUpstreamList() {
+ DiscoverySyncData data = discoveryData();
+ data.setUpstreamDataList(Collections.emptyList());
+ String message = discoveryMessage(new
WebsocketDataChangedListener(true), data, DataEventTypeEnum.UPDATE);
+ JsonObject decoded =
JsonParser.parseString(message).getAsJsonObject().getAsJsonArray("data").get(0).getAsJsonObject();
+ assertTrue(decoded.getAsJsonArray("upstreamDataList").isEmpty());
+ }
+
+ @Test
+ public void testCompactDiscoverySeparatesNamespacesAndSkipsNullRows() {
+ DiscoverySyncData first = discoveryData();
+ DiscoverySyncData second = discoveryData();
+ second.setNamespaceId("other-namespace");
+ second.setSelectorId("other-selector");
+ List<DiscoverySyncData> data = new ArrayList<>();
+ data.add(first);
+ data.add(null);
+ data.add(second);
+ try (MockedStatic<WebsocketCollector> collector =
mockStatic(WebsocketCollector.class)) {
+ new
WebsocketDataChangedListener(true).onDiscoveryUpstreamChanged(data,
DataEventTypeEnum.UPDATE);
+ for (DiscoverySyncData item : List.of(first, second)) {
+ collector.verify(() ->
WebsocketCollector.send(eq(item.getNamespaceId()), argThat(message -> {
+ JsonObject frame =
JsonParser.parseString(message).getAsJsonObject();
+ return frame.getAsJsonArray("data").size() == 1
+ &&
item.getSelectorId().equals(frame.getAsJsonArray("data").get(0).getAsJsonObject().get("selectorId").getAsString());
+ }), eq(DataEventTypeEnum.UPDATE)));
+ }
+ }
+ }
+
+ @Test
+ public void testCompactDiscoveryPreservesMissingListAndDefaultNamespace() {
+ DiscoverySyncData data = discoveryData();
+ data.setNamespaceId(null);
+ data.setUpstreamDataList(null);
+ try (MockedStatic<WebsocketCollector> collector =
mockStatic(WebsocketCollector.class)) {
+ new
WebsocketDataChangedListener(true).onDiscoveryUpstreamChanged(Collections.singletonList(data),
DataEventTypeEnum.DELETE);
+ collector.verify(() ->
WebsocketCollector.send(eq(Constants.SYS_DEFAULT_NAMESPACE_ID), argThat(message
-> {
+ JsonObject item =
JsonParser.parseString(message).getAsJsonObject().getAsJsonArray("data").get(0).getAsJsonObject();
+ return !item.has("upstreamDataList") &&
"selector-id".equals(item.get("selectorId").getAsString());
+ }), eq(DataEventTypeEnum.DELETE)));
+ }
+ }
+
+ @Test
+ public void testCompactDiscoveryPreservesNullUpstreamEntries() {
+ DiscoverySyncData data = discoveryData();
+ List<DiscoveryUpstreamData> upstreams = new
ArrayList<>(data.getUpstreamDataList());
+ upstreams.add(null);
+ data.setUpstreamDataList(upstreams);
+ String message = discoveryMessage(new
WebsocketDataChangedListener(true), data, DataEventTypeEnum.UPDATE);
+ JsonObject item =
JsonParser.parseString(message).getAsJsonObject().getAsJsonArray("data").get(0).getAsJsonObject();
+ assertEquals(2, item.getAsJsonArray("upstreamDataList").size());
+
assertTrue(item.getAsJsonArray("upstreamDataList").get(1).isJsonNull());
+ }
+
+ private String discoveryMessage(final WebsocketDataChangedListener
listener, final DiscoverySyncData data,
+ final DataEventTypeEnum eventType) {
+ List<String> messages = new ArrayList<>();
+ try (MockedStatic<WebsocketCollector> collector =
mockStatic(WebsocketCollector.class)) {
+ collector.when(() ->
WebsocketCollector.send(eq(data.getNamespaceId()), anyString(), eq(eventType)))
+ .thenAnswer(invocation -> {
+ messages.add(invocation.getArgument(1));
+ return null;
+ });
+
listener.onDiscoveryUpstreamChanged(Collections.singletonList(data), eventType);
+ }
+ assertEquals(1, messages.size());
+ return messages.get(0);
+ }
+
+ private DiscoverySyncData discoveryData() {
+ DiscoveryUpstreamData upstream = new DiscoveryUpstreamData();
+ upstream.setId("row-id");
+ upstream.setDateCreated(new Timestamp(1700000000000L));
+ upstream.setDateUpdated(new Timestamp(1700000001000L));
+ upstream.setDiscoveryHandlerId("handler-id");
+ upstream.setProtocol("http://";);
+ upstream.setUrl("127.0.0.1:8080");
+ upstream.setStatus(0);
+ upstream.setWeight(100);
+ upstream.setProps("{\"warmup\":600000,\"healthCheckEnabled\":false}");
+ upstream.setNamespaceId(Constants.SYS_DEFAULT_NAMESPACE_ID);
+ DiscoverySyncData data = new DiscoverySyncData();
+ data.setSelectorId("selector-id");
+ data.setSelectorName("tcp-route");
+ data.setPluginName("divide");
+ data.setNamespaceId(Constants.SYS_DEFAULT_NAMESPACE_ID);
+ data.setDiscoveryHandlerId("handler-id");
+ data.setUpstreamDataList(Collections.singletonList(upstream));
+ return data;
+ }
+
/**
* test DiscoverySyncData with empty list — no send.
*/
