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 581bee9f15 [type:fix] Immediate websocket master rediscovery and
configurable check interval (#7808) (#7812)
581bee9f15 is described below
commit 581bee9f15ce0a6778df6cc28eb210eb29ba7c70
Author: Harsh Raj <[email protected]>
AuthorDate: Mon Oct 19 12:53:42 2026 +0530
[type:fix] Immediate websocket master rediscovery and configurable check
interval (#7808) (#7812)
Co-authored-by: Harsh <[email protected]>
---
.../data/websocket/WebsocketSyncDataService.java | 55 ++++++++++++--
.../websocket/client/ShenyuWebsocketClient.java | 49 +++++++++++++
.../data/websocket/config/WebsocketConfig.java | 30 +++++++-
.../websocket/WebsocketSyncDataServiceTest.java | 85 ++++++++++++++++++++++
.../client/ShenyuWebsocketClientTest.java | 71 ++++++++++++++++++
.../data/websocket/config/WebsocketConfigTest.java | 12 ++-
6 files changed, 292 insertions(+), 10 deletions(-)
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataService.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataService.java
index c2f32418a2..76f3dc8946 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataService.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataService.java
@@ -25,6 +25,7 @@ import org.apache.shenyu.common.config.ShenyuConfig;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.enums.RunningModeEnum;
import org.apache.shenyu.common.timer.AbstractRoundTask;
+import org.apache.shenyu.common.timer.TaskEntity;
import org.apache.shenyu.common.timer.Timer;
import org.apache.shenyu.common.timer.TimerTask;
import org.apache.shenyu.common.timer.WheelTimerFactory;
@@ -67,6 +68,8 @@ public class WebsocketSyncDataService implements
SyncDataService {
private final AtomicBoolean initialSyncReady = new AtomicBoolean();
+ private final AtomicBoolean rediscoveryScheduled = new
AtomicBoolean(false);
+
private final PluginDataSubscriber pluginDataSubscriber;
@@ -131,7 +134,8 @@ public class WebsocketSyncDataService implements
SyncDataService {
clients.add(createClient(url));
}
LOG.info("start check task...");
- this.timer.add(timerTask = new AbstractRoundTask(null,
TimeUnit.SECONDS.toMillis(60)) {
+ long intervalSeconds = websocketConfig.getMasterCheckInterval() <= 0 ?
60 : websocketConfig.getMasterCheckInterval();
+ this.timer.add(timerTask = new AbstractRoundTask(null,
TimeUnit.SECONDS.toMillis(intervalSeconds)) {
@Override
public void doRun(final String key, final TimerTask timerTask) {
masterCheck();
@@ -148,8 +152,10 @@ public class WebsocketSyncDataService implements
SyncDataService {
}
if (CollectionUtils.isEmpty(clients)) {
List<String> urls = websocketConfig.getUrls();
- for (String url : urls) {
- clients.add(createClient(url));
+ if (CollectionUtils.isNotEmpty(urls)) {
+ for (String url : urls) {
+ clients.add(createClient(url));
+ }
}
}
Set<URI> masterUrls = new LinkedHashSet<>();
@@ -165,7 +171,9 @@ public class WebsocketSyncDataService implements
SyncDataService {
// check running mode
if (Objects.equals(runningMode,
RunningModeEnum.STANDALONE.name())) {
LOG.info("admin running in standalone mode...");
- timerTask.cancel();
+ if (Objects.nonNull(timerTask)) {
+ timerTask.cancel();
+ }
return;
}
@@ -188,6 +196,14 @@ public class WebsocketSyncDataService implements
SyncDataService {
}
}
}
+ if (CollectionUtils.isEmpty(clients)) {
+ List<String> urls = websocketConfig.getUrls();
+ if (CollectionUtils.isNotEmpty(urls)) {
+ for (String url : urls) {
+ clients.add(createClient(url));
+ }
+ }
+ }
}
private URI parseMasterUri(final String masterUrl) {
@@ -206,6 +222,34 @@ public class WebsocketSyncDataService implements
SyncDataService {
LOG.warn("Ignore invalid websocket master address [{}]", masterUrl);
return null;
}
+
+ private void onMasterLost(final ShenyuWebsocketClient client) {
+ if (closed) {
+ return;
+ }
+ LOG.info("master lost detected from websocket server[{}], scheduling
rediscovery", client.getURI());
+ triggerMasterRediscovery();
+ }
+
+ private void triggerMasterRediscovery() {
+ if (closed) {
+ return;
+ }
+ if (rediscoveryScheduled.compareAndSet(false, true)) {
+ try {
+ this.timer.add(new TimerTask(200, TimeUnit.MILLISECONDS) {
+ @Override
+ public void run(final TaskEntity taskEntity) {
+ rediscoveryScheduled.set(false);
+ masterCheck();
+ }
+ });
+ } catch (Exception ex) {
+ rediscoveryScheduled.set(false);
+ LOG.warn("Failed to schedule master rediscovery task", ex);
+ }
+ }
+ }
@Override
public synchronized void close() {
@@ -248,7 +292,8 @@ public class WebsocketSyncDataService implements
SyncDataService {
this.aiProxyApiKeyDataSubscribers,
namespaceId,
serverProperties.getPort(),
- websocketConfig.isInitialSyncReadiness() ? initialSyncReady :
null);
+ websocketConfig.isInitialSyncReadiness() ? initialSyncReady :
null,
+ this::onMasterLost);
}
/**
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java
index 86df7d5af9..663fe8b83e 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java
@@ -60,6 +60,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
/**
* The type shenyu websocket client.
@@ -125,6 +126,8 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
private volatile Thread reconnectThread;
+ private final Consumer<ShenyuWebsocketClient> onMasterLostCallback;
+
/**
* Instantiates a new shenyu websocket client.
*
@@ -146,6 +149,7 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
final Integer port
) {
super(serverUri);
+ this.onMasterLostCallback = null;
this.namespaceId = namespaceId;
this.addHeader(Constants.SHENYU_NAMESPACE_ID, namespaceId);
this.addHeader(Constants.CLIENT_PORT_NAME, String.valueOf(port));
@@ -212,11 +216,44 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
final String namespaceId,
final Integer port,
final AtomicBoolean initialSyncReady) {
+ this(serverUri, headers, pluginDataSubscriber, metaDataSubscribers,
authDataSubscribers,
+ proxySelectorDataSubscribers,
discoveryUpstreamDataSubscribers, aiProxyApiKeyDataSubscribers,
+ namespaceId, port, initialSyncReady, null);
+ }
+
+ /**
+ * Create a client with an optional startup readiness latch and master
lost callback.
+ * @param serverUri server URI
+ * @param headers headers
+ * @param pluginDataSubscriber plugin subscriber
+ * @param metaDataSubscribers metadata subscribers
+ * @param authDataSubscribers authorization subscribers
+ * @param proxySelectorDataSubscribers proxy selector subscribers
+ * @param discoveryUpstreamDataSubscribers discovery subscribers
+ * @param aiProxyApiKeyDataSubscribers API key subscribers
+ * @param namespaceId namespace
+ * @param port gateway port
+ * @param initialSyncReady startup latch, null for the legacy protocol
+ * @param onMasterLostCallback callback invoked when connection to master
is lost
+ */
+ public ShenyuWebsocketClient(final URI serverUri,
+ final Map<String, String> headers,
+ final PluginDataSubscriber
pluginDataSubscriber,
+ final List<MetaDataSubscriber>
metaDataSubscribers,
+ final List<AuthDataSubscriber>
authDataSubscribers,
+ final List<ProxySelectorDataSubscriber>
proxySelectorDataSubscribers,
+ final List<DiscoveryUpstreamDataSubscriber>
discoveryUpstreamDataSubscribers,
+ final List<AiProxyApiKeyDataSubscriber>
aiProxyApiKeyDataSubscribers,
+ final String namespaceId,
+ final Integer port,
+ final AtomicBoolean initialSyncReady,
+ final Consumer<ShenyuWebsocketClient>
onMasterLostCallback) {
super(serverUri, headers);
if (Objects.nonNull(initialSyncReady)) {
this.initialSyncState = new InitialSyncState(initialSyncReady,
this::resetSyncFailures);
}
this.namespaceId = namespaceId;
+ this.onMasterLostCallback = onMasterLostCallback;
LOG.info("shenyu bootstrap websocket namespaceId: {}", namespaceId);
this.addHeader(Constants.SHENYU_NAMESPACE_ID, namespaceId);
this.addHeader(Constants.CLIENT_PORT_NAME, String.valueOf(port));
@@ -297,10 +334,14 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
if (Objects.equals(WebsocketSyncFrame.EVENT_TYPE, eventType) &&
Objects.nonNull(initialSyncState)) {
initialSyncState.accept(GsonUtils.getInstance().fromJson(result,
WebsocketSyncFrame.class), this::handleResult);
} else if (Objects.equals(DataEventTypeEnum.RUNNING_MODE.name(),
eventType)) {
+ boolean wasMaster = this.isConnectedToMaster;
String mode =
String.valueOf(jsonToMap.get(RunningModeConstants.RUNNING_MODE));
if (!Objects.equals(RunningModeEnum.STANDALONE.name(), mode)) {
this.masterUrl =
String.valueOf(jsonToMap.get(RunningModeConstants.MASTER_URL));
this.isConnectedToMaster =
Boolean.TRUE.equals(jsonToMap.get(RunningModeConstants.IS_MASTER));
+ if (wasMaster && !this.isConnectedToMaster &&
!this.manuallyClosed.get() && Objects.nonNull(onMasterLostCallback)) {
+ onMasterLostCallback.accept(this);
+ }
}
this.runningMode = mode;
} else if (Objects.nonNull(initialSyncState)) {
@@ -318,7 +359,12 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
@Override
public void onClose(final int i, final String s, final boolean b) {
+ boolean wasMaster = this.isConnectedToMaster;
+ this.isConnectedToMaster = false;
this.close();
+ if (wasMaster && !this.manuallyClosed.get() &&
Objects.nonNull(onMasterLostCallback)) {
+ onMasterLostCallback.accept(this);
+ }
}
@Override
@@ -378,6 +424,9 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
this.reconnectBackoff.set(0);
this.sendPing();
send(getInstanceInfo());
+ if (!Objects.equals(RunningModeEnum.STANDALONE.name(),
runningMode)) {
+ send(DataEventTypeEnum.RUNNING_MODE.name());
+ }
LOG.debug("websocket send to [{}] ping message successful",
this.getURI());
}
} catch (Exception e) {
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/config/WebsocketConfig.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/config/WebsocketConfig.java
index e7ce4023d0..19bc1535ad 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/config/WebsocketConfig.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/config/WebsocketConfig.java
@@ -40,6 +40,11 @@ public class WebsocketConfig {
private boolean initialSyncReadiness;
+ /**
+ * Master check interval in seconds.
+ */
+ private int masterCheckInterval = 60;
+
/**
* Whether the opt-in initial synchronization protocol is required.
* @return enabled
@@ -105,6 +110,24 @@ public class WebsocketConfig {
public void setToken(final String token) {
this.token = token;
}
+
+ /**
+ * get master check interval in seconds.
+ *
+ * @return master check interval
+ */
+ public int getMasterCheckInterval() {
+ return masterCheckInterval;
+ }
+
+ /**
+ * set master check interval in seconds.
+ *
+ * @param masterCheckInterval master check interval
+ */
+ public void setMasterCheckInterval(final int masterCheckInterval) {
+ this.masterCheckInterval = masterCheckInterval;
+ }
@Override
public boolean equals(final Object o) {
@@ -118,12 +141,13 @@ public class WebsocketConfig {
return Objects.equals(urls, that.urls)
&& Objects.equals(allowOrigin, that.allowOrigin)
&& Objects.equals(token, that.token)
- && initialSyncReadiness == that.initialSyncReadiness;
+ && initialSyncReadiness == that.initialSyncReadiness
+ && masterCheckInterval == that.masterCheckInterval;
}
@Override
public int hashCode() {
- return Objects.hash(urls, allowOrigin, token, initialSyncReadiness);
+ return Objects.hash(urls, allowOrigin, token, initialSyncReadiness,
masterCheckInterval);
}
@Override
@@ -136,6 +160,8 @@ public class WebsocketConfig {
+ allowOrigin
+ '\''
+ ", token='******'"
+ + ", masterCheckInterval="
+ + masterCheckInterval
+ '}';
}
}
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataServiceTest.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataServiceTest.java
index 65da2272a7..92428f9e31 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataServiceTest.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataServiceTest.java
@@ -36,9 +36,12 @@ import
org.springframework.boot.autoconfigure.web.ServerProperties;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import java.net.URI;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.doThrow;
@@ -134,6 +137,88 @@ public final class WebsocketSyncDataServiceTest {
}
}
+ @Test
+ public void testConfigurableMasterCheckInterval() throws Exception {
+ WebsocketConfig config = new WebsocketConfig();
+ config.setUrls(Collections.emptyList());
+ config.setMasterCheckInterval(15);
+ WebsocketSyncDataService service = new WebsocketSyncDataService(
+ config,
+ new ShenyuConfig(),
+ mock(PluginDataSubscriber.class),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ mock(ServerProperties.class));
+ try {
+ Field timerTaskField =
WebsocketSyncDataService.class.getDeclaredField("timerTask");
+ timerTaskField.setAccessible(true);
+ TimerTask timerTask = (TimerTask) timerTaskField.get(service);
+ Method getDelayMs =
TimerTask.class.getDeclaredMethod("getDelayMs");
+ getDelayMs.setAccessible(true);
+ long delayMs = (long) getDelayMs.invoke(timerTask);
+ assertEquals(15000L, delayMs);
+ } finally {
+ service.close();
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testMasterCheckReseedsClientsWhenAllRemoved() throws Exception
{
+ WebsocketConfig config = new WebsocketConfig();
+
config.setUrls(Collections.singletonList("ws://127.0.0.1:9095/websocket"));
+ WebsocketSyncDataService service = new WebsocketSyncDataService(
+ config,
+ new ShenyuConfig(),
+ mock(PluginDataSubscriber.class),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ mock(ServerProperties.class));
+ Field clientsField =
WebsocketSyncDataService.class.getDeclaredField("clients");
+ clientsField.setAccessible(true);
+ List<ShenyuWebsocketClient> clients = (List<ShenyuWebsocketClient>)
clientsField.get(service);
+ clients.clear();
+ ShenyuWebsocketClient closedClient = mock(ShenyuWebsocketClient.class);
+ when(closedClient.isOpen()).thenReturn(false);
+ clients.add(closedClient);
+
+ try {
+ Method masterCheck =
WebsocketSyncDataService.class.getDeclaredMethod("masterCheck");
+ masterCheck.setAccessible(true);
+ masterCheck.invoke(service);
+ verify(closedClient).nowClose();
+ assertEquals(1, clients.size());
+ } finally {
+ service.close();
+ }
+ }
+
+ @Test
+ public void testOnMasterLostTriggersRediscovery() throws Exception {
+ WebsocketSyncDataService service = createWebsocketSyncDataService();
+ ShenyuWebsocketClient client = mock(ShenyuWebsocketClient.class);
+ when(client.getURI()).thenReturn(URI.create("ws://localhost:9095"));
+
+ try {
+ Method onMasterLost =
WebsocketSyncDataService.class.getDeclaredMethod("onMasterLost",
ShenyuWebsocketClient.class);
+ onMasterLost.setAccessible(true);
+ onMasterLost.invoke(service, client);
+
+ Field rediscoveryScheduledField =
WebsocketSyncDataService.class.getDeclaredField("rediscoveryScheduled");
+ rediscoveryScheduledField.setAccessible(true);
+ AtomicBoolean rediscoveryScheduled = (AtomicBoolean)
rediscoveryScheduledField.get(service);
+ assertTrue(rediscoveryScheduled.get());
+ } finally {
+ service.close();
+ }
+ }
+
private WebsocketSyncDataService createWebsocketSyncDataService() {
WebsocketConfig websocketConfig = new WebsocketConfig();
websocketConfig.setUrls(Collections.emptyList());
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java
index 7c51d9f92b..5409433f3f 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java
@@ -17,11 +17,13 @@
package org.apache.shenyu.plugin.sync.data.websocket.client;
+import org.apache.shenyu.common.constant.RunningModeConstants;
import org.apache.shenyu.common.dto.PluginData;
import org.apache.shenyu.common.dto.WebsocketData;
import org.apache.shenyu.common.dto.WebsocketSyncFrame;
import org.apache.shenyu.common.enums.ConfigGroupEnum;
import org.apache.shenyu.common.enums.DataEventTypeEnum;
+import org.apache.shenyu.common.enums.RunningModeEnum;
import org.apache.shenyu.common.exception.ShenyuException;
import org.apache.shenyu.common.timer.TimerTask;
import org.apache.shenyu.common.utils.GsonUtils;
@@ -44,12 +46,15 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.URI;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -206,6 +211,72 @@ public class ShenyuWebsocketClientTest {
verify(shenyuWebsocketClient).close();
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testOnCloseWithMasterLostNotification() {
+ ShenyuWebsocketClient client = createMockClient();
+ Consumer<ShenyuWebsocketClient> callback = mock(Consumer.class);
+ setField(client, "onMasterLostCallback", callback);
+ setField(client, "isConnectedToMaster", true);
+ doNothing().when(client).close();
+
+ client.onClose(1, "disconnected", true);
+
+ verify(callback).accept(client);
+ assertFalse(client.isConnectedToMaster());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testOnCloseWhenManuallyClosedDoesNotNotifyMasterLost() {
+ ShenyuWebsocketClient client = createMockClient();
+ Consumer<ShenyuWebsocketClient> callback = mock(Consumer.class);
+ setField(client, "onMasterLostCallback", callback);
+ setField(client, "isConnectedToMaster", true);
+ setField(client, "manuallyClosed", new AtomicBoolean(true));
+ doNothing().when(client).close();
+
+ client.onClose(1, "normal closure", true);
+
+ verify(callback, never()).accept(any());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testOnMessageRunningModeDemotionNotifiesMasterLost() {
+ ShenyuWebsocketClient client = createMockClient();
+ Consumer<ShenyuWebsocketClient> callback = mock(Consumer.class);
+ setField(client, "onMasterLostCallback", callback);
+ setField(client, "isConnectedToMaster", true);
+
+ Map<String, Object> map = new HashMap<>();
+ map.put(RunningModeConstants.EVENT_TYPE,
DataEventTypeEnum.RUNNING_MODE.name());
+ map.put(RunningModeConstants.RUNNING_MODE,
RunningModeEnum.CLUSTER.name());
+ map.put(RunningModeConstants.MASTER_URL, "ws://localhost:9096");
+ map.put(RunningModeConstants.IS_MASTER, false);
+ String json = GsonUtils.getInstance().toJson(map);
+
+ client.onMessage(json);
+
+ verify(callback).accept(client);
+ assertFalse(client.isConnectedToMaster());
+ }
+
+ @Test
+ public void testHealthCheckSendsRunningModeWhenClusterAndOpen() throws
Exception {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ doReturn(true).when(client).isOpen();
+ doNothing().when(client).sendPing();
+ doNothing().when(client).send(anyString());
+ setField(client, "runningMode", RunningModeEnum.CLUSTER.name());
+
+ Method healthCheck =
ShenyuWebsocketClient.class.getDeclaredMethod("healthCheck");
+ healthCheck.setAccessible(true);
+ healthCheck.invoke(client);
+
+ verify(client).send(DataEventTypeEnum.RUNNING_MODE.name());
+ }
+
@Test
public void testOnError() {
shenyuWebsocketClient = spy(shenyuWebsocketClient);
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/config/WebsocketConfigTest.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/config/WebsocketConfigTest.java
index ee14c4d552..a487fbfb18 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/config/WebsocketConfigTest.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/config/WebsocketConfigTest.java
@@ -54,6 +54,9 @@ public class WebsocketConfigTest {
assertEquals(URLS, websocketConfig.getUrls());
assertEquals(ALLOW_ORIGIN, websocketConfig.getAllowOrigin());
assertEquals(TOKEN, websocketConfig.getToken());
+ assertEquals(60, websocketConfig.getMasterCheckInterval());
+ websocketConfig.setMasterCheckInterval(30);
+ assertEquals(30, websocketConfig.getMasterCheckInterval());
}
@Test
@@ -66,6 +69,9 @@ public class WebsocketConfigTest {
assertEquals(websocketConfig, that);
that.setInitialSyncReadiness(true);
assertNotEquals(websocketConfig, that);
+ that.setInitialSyncReadiness(false);
+ that.setMasterCheckInterval(10);
+ assertNotEquals(websocketConfig, that);
assertNotEquals(websocketConfig, null);
assertNotEquals(websocketConfig, new Object());
}
@@ -73,13 +79,13 @@ public class WebsocketConfigTest {
@Test
public void testHashCode() {
assertEquals(Objects.hash(websocketConfig.getUrls(),
websocketConfig.getAllowOrigin(),
- websocketConfig.getToken(),
websocketConfig.isInitialSyncReadiness()), websocketConfig.hashCode());
+ websocketConfig.getToken(),
websocketConfig.isInitialSyncReadiness(),
websocketConfig.getMasterCheckInterval()), websocketConfig.hashCode());
}
@Test
public void testToString() {
- String toString = "WebsocketConfig{urls='%s', allowOrigin='%s',
token='******'}";
- String expected = String.format(toString, URLS, ALLOW_ORIGIN);
+ String toString = "WebsocketConfig{urls='%s', allowOrigin='%s',
token='******', masterCheckInterval=%d}";
+ String expected = String.format(toString, URLS, ALLOW_ORIGIN, 60);
assertEquals(expected, websocketConfig.toString());
assertFalse(websocketConfig.toString().contains(TOKEN));
}
