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 e09eb35b68 fix(register): report the bound server port for gateway
instance identity (#7732) (#7861)
e09eb35b68 is described below
commit e09eb35b6837a7c54d6982f204ec3df7cecf2e96
Author: markxu <[email protected]>
AuthorDate: Tue Oct 20 12:36:46 2026 +0800
fix(register): report the bound server port for gateway instance identity
(#7732) (#7861)
Heartbeat, websocket sync, and HTTP long polling wait for the application
server's bound port and ignore the management listener. server.port=0 is
reported as the ephemeral port that was actually bound, and a missing or
non-positive client port is not published.
Fixes #7732
Co-authored-by: Cursor Agent <[email protected]>
Co-authored-by: markxu <[email protected]>
---
.../http/HttpLongPollingDataChangedListener.java | 4 +-
.../HttpLongPollingDataChangedListenerTest.java | 21 +++
.../shenyu/common/utils/BoundServerPort.java | 49 +++++++
.../shenyu/common/utils/BoundServerPortTest.java | 44 ++++++
.../register/client/beat/HeartbeatListener.java | 68 ++++++++--
.../client/beat/HeartbeatBoundPortTest.java | 130 ++++++++++++++++++
.../client/beat/HeartbeatListenerTest.java | 124 ++++++++++++++---
.../beat/HeartbeatListenerConfiguration.java | 7 +-
.../websocket/WebsocketSyncDataConfiguration.java | 8 +-
.../shenyu-sync-data-http/pom.xml | 5 +
.../shenyu/sync/data/http/HttpSyncDataService.java | 46 ++++++-
.../sync/data/http/HttpSyncBoundPortTest.java | 148 +++++++++++++++++++++
.../data/websocket/WebsocketSyncDataService.java | 81 +++++++----
.../data/websocket/WebsocketBoundPortTest.java | 117 ++++++++++++++++
.../websocket/WebsocketMasterRedirectTest.java | 13 +-
.../websocket/WebsocketSyncDataServiceTest.java | 13 +-
16 files changed, 786 insertions(+), 92 deletions(-)
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java
index 9379a9b39c..f9fcd5976a 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java
@@ -89,8 +89,6 @@ public class HttpLongPollingDataChangedListener extends
AbstractDataChangedListe
private static final String X_REAL_PORT = "X-Real-PORT";
- private static final String CLIENT_PORT_ZERO = "0";
-
/**
* Blocked client.
*/
@@ -146,7 +144,7 @@ public class HttpLongPollingDataChangedListener extends
AbstractDataChangedListe
// The header already contains JSON text; encoding it again adds a
string wrapper.
final String bootstrapInfo =
StringUtils.defaultString(request.getHeader(InstanceTypeConstants.BOOTSTRAP_INSTANCE_INFO),
"");
final String clientPort =
StringUtils.defaultString(request.getHeader(X_REAL_PORT), StringUtils.EMPTY);
- if (!CLIENT_PORT_ZERO.equals(clientPort)) {
+ if (NumberUtils.toInt(clientPort, -1) > 0) {
InstanceInfoReportEvent instanceInfoReportEvent =
InstanceInfoReportEvent.builder()
.instanceIp(clientIp)
.instancePort(clientPort)
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListenerTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListenerTest.java
index b926e8beb1..6aed2a0d53 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListenerTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListenerTest.java
@@ -438,6 +438,27 @@ public final class HttpLongPollingDataChangedListenerTest {
clearAllGroupsInCache(DEFAULT_NAMESPACE);
}
+ @Test
+ public void testDoLongPollingIgnoresPlaceholderClientPort() throws
Exception {
+ populateAllGroupsInCache(DEFAULT_NAMESPACE, "serverMd5");
+ try {
+ setAllGroupParamsDefault(this.httpServletRequest);
+
this.httpServletRequest.setParameter(ConfigGroupEnum.PLUGIN.name(),
"clientMd5,500");
+ this.httpServletRequest.addHeader("X-Real-PORT", "0");
+ listener.doLongPolling(this.httpServletRequest,
this.httpServletResponse);
+ Mockito.verify(instanceInfoReportEventPublisher,
Mockito.never()).publish(Mockito.any());
+
+ this.httpServletRequest = new MockHttpServletRequest();
+ this.httpServletResponse = new MockHttpServletResponse();
+ setAllGroupParamsDefault(this.httpServletRequest);
+
this.httpServletRequest.setParameter(ConfigGroupEnum.PLUGIN.name(),
"clientMd5,500");
+ listener.doLongPolling(this.httpServletRequest,
this.httpServletResponse);
+ Mockito.verify(instanceInfoReportEventPublisher,
Mockito.never()).publish(Mockito.any());
+ } finally {
+ clearAllGroupsInCache(DEFAULT_NAMESPACE);
+ }
+ }
+
/**
* test doLongPolling without optional instance information.
*/
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/utils/BoundServerPort.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/utils/BoundServerPort.java
new file mode 100644
index 0000000000..11c2279fcd
--- /dev/null
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/utils/BoundServerPort.java
@@ -0,0 +1,49 @@
+/*
+ * 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.utils;
+
+import java.util.OptionalInt;
+
+/**
+ * Selects the port of the application server that should identify a gateway
instance.
+ */
+public final class BoundServerPort {
+
+ /**
+ * Namespace of Spring Boot's separate management server.
+ */
+ public static final String MANAGEMENT_NAMESPACE = "management";
+
+ private BoundServerPort() {
+ }
+
+ /**
+ * Accept a listener port only after the application server has bound it.
+ * A management server and any non-positive observation are not the
gateway listener.
+ *
+ * @param serverNamespace server namespace, empty for the primary
application server
+ * @param port observed port
+ * @return the bound application port, or empty when the observation must
not be reported
+ */
+ public static OptionalInt resolve(final String serverNamespace, final int
port) {
+ if (MANAGEMENT_NAMESPACE.equals(serverNamespace) || port <= 0) {
+ return OptionalInt.empty();
+ }
+ return OptionalInt.of(port);
+ }
+}
diff --git
a/shenyu-common/src/test/java/org/apache/shenyu/common/utils/BoundServerPortTest.java
b/shenyu-common/src/test/java/org/apache/shenyu/common/utils/BoundServerPortTest.java
new file mode 100644
index 0000000000..fb5b32f9eb
--- /dev/null
+++
b/shenyu-common/src/test/java/org/apache/shenyu/common/utils/BoundServerPortTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.OptionalInt;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test for {@link BoundServerPort}.
+ */
+class BoundServerPortTest {
+
+ @Test
+ void acceptsThePrimaryServerOnceItHasBound() {
+ assertEquals(OptionalInt.of(9195), BoundServerPort.resolve(null,
9195));
+ assertEquals(OptionalInt.of(18080), BoundServerPort.resolve("server",
18080));
+ }
+
+ @Test
+ void rejectsTheManagementServerAndUnboundPorts() {
+
assertTrue(BoundServerPort.resolve(BoundServerPort.MANAGEMENT_NAMESPACE,
19090).isEmpty());
+ assertTrue(BoundServerPort.resolve(null, 0).isEmpty());
+ assertTrue(BoundServerPort.resolve(null, -1).isEmpty());
+ }
+}
diff --git
a/shenyu-register-center/shenyu-register-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListener.java
b/shenyu-register-center/shenyu-register-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListener.java
index c393cda0b0..2fd6fd48bf 100644
---
a/shenyu-register-center/shenyu-register-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListener.java
+++
b/shenyu-register-center/shenyu-register-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListener.java
@@ -28,6 +28,7 @@ import org.apache.shenyu.common.config.ShenyuConfig;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.constant.InstanceTypeConstants;
import org.apache.shenyu.common.utils.AesUtils;
+import org.apache.shenyu.common.utils.BoundServerPort;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.common.utils.IpUtils;
import org.apache.shenyu.common.utils.SystemInfoUtils;
@@ -37,12 +38,14 @@ import org.checkerframework.checker.nullness.qual.NonNull;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.web.ServerProperties;
+import org.springframework.boot.web.context.WebServerInitializedEvent;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.EventListener;
import java.util.List;
+import java.util.Objects;
import java.util.Optional;
+import java.util.OptionalInt;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -71,10 +74,16 @@ public class HeartbeatListener {
private ShenyuBootstrapHeartBeatConfig config;
- public HeartbeatListener(final ShenyuBootstrapHeartBeatConfig config,
final ShenyuConfig shenyuConfig, final ServerProperties serverProperties) {
+ /**
+ * Port of the application listener. Zero until {@link
WebServerInitializedEvent}.
+ */
+ private volatile int boundPort;
+
+ private volatile boolean closed;
+
+ public HeartbeatListener(final ShenyuBootstrapHeartBeatConfig config,
final ShenyuConfig shenyuConfig) {
executor = new ScheduledThreadPoolExecutor(1,
ShenyuThreadFactory.create("scheduled-instance-task", false));
this.shenyuConfig = shenyuConfig;
- LOG.info("Web server initialized on port {}, starting heartbeat
reporter", serverProperties.getPort());
this.username = config.getProps().getProperty(Constants.USER_NAME);
this.password = config.getProps().getProperty(Constants.PASS_WORD);
this.config = config;
@@ -99,16 +108,48 @@ public class HeartbeatListener {
}
}
});
- executor.scheduleAtFixedRate(() -> {
- InstanceBeatInfoDTO instanceBeatInfoDTO = new
InstanceBeatInfoDTO();
-
instanceBeatInfoDTO.setInstancePort(String.valueOf(serverProperties.getPort()));
- instanceBeatInfoDTO.setInstanceIp(IpUtils.getHost());
- instanceBeatInfoDTO.setNamespaceId(shenyuConfig.getNamespace());
-
instanceBeatInfoDTO.setInstanceInfo(SystemInfoUtils.getSystemInfo());
-
instanceBeatInfoDTO.setInstanceType(InstanceTypeConstants.BOOTSTRAP_INSTANCE_TYPE);
- sendHeartbeat(instanceBeatInfoDTO);
- }, INITIAL_DELAY, CHECK_PERIOD, TimeUnit.SECONDS
- );
+ executor.scheduleAtFixedRate(this::reportHeartbeat, INITIAL_DELAY,
CHECK_PERIOD, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Record the application listener port. The management server is ignored,
and nothing is reported
+ * until a positive bound port is observed. Shutdown wins over a late
initialization event.
+ *
+ * @param event web server initialization event
+ */
+ @EventListener
+ public void onWebServerInitialized(final WebServerInitializedEvent event) {
+ if (closed || Objects.isNull(event.getWebServer()) ||
Objects.isNull(event.getApplicationContext())) {
+ return;
+ }
+ OptionalInt resolved =
BoundServerPort.resolve(event.getApplicationContext().getServerNamespace(),
event.getWebServer().getPort());
+ if (resolved.isEmpty()) {
+ LOG.debug("Ignore heartbeat port from namespace {} port {}",
event.getApplicationContext().getServerNamespace(),
event.getWebServer().getPort());
+ return;
+ }
+ int observed = resolved.getAsInt();
+ if (boundPort == observed) {
+ return;
+ }
+ boundPort = observed;
+ LOG.info("Application server bound on port {}, heartbeat reporter will
use it", observed);
+ }
+
+ private void reportHeartbeat() {
+ if (closed) {
+ return;
+ }
+ int port = boundPort;
+ if (port <= 0) {
+ return;
+ }
+ InstanceBeatInfoDTO instanceBeatInfoDTO = new InstanceBeatInfoDTO();
+ instanceBeatInfoDTO.setInstancePort(String.valueOf(port));
+ instanceBeatInfoDTO.setInstanceIp(IpUtils.getHost());
+ instanceBeatInfoDTO.setNamespaceId(shenyuConfig.getNamespace());
+ instanceBeatInfoDTO.setInstanceInfo(SystemInfoUtils.getSystemInfo());
+
instanceBeatInfoDTO.setInstanceType(InstanceTypeConstants.BOOTSTRAP_INSTANCE_TYPE);
+ sendHeartbeat(instanceBeatInfoDTO);
}
private void sendHeartbeat(final InstanceBeatInfoDTO instanceBeatInfoDTO) {
@@ -129,6 +170,7 @@ public class HeartbeatListener {
@EventListener(ContextClosedEvent.class)
public void onShutdown() {
+ closed = true;
executor.shutdown();
try {
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
diff --git
a/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatBoundPortTest.java
b/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatBoundPortTest.java
new file mode 100644
index 0000000000..642ee875e0
--- /dev/null
+++
b/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatBoundPortTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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.register.client.beat;
+
+import org.apache.shenyu.common.config.ShenyuConfig;
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.utils.BoundServerPort;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.common.utils.SystemInfoUtils;
+import org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO;
+import org.apache.shenyu.register.common.utils.RegisterUtils;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import
org.springframework.boot.web.embedded.tomcat.TomcatServletWebServerFactory;
+import
org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
+import
org.springframework.boot.web.servlet.context.ServletWebServerInitializedEvent;
+import org.springframework.boot.web.server.WebServer;
+
+import java.lang.reflect.Method;
+import java.net.ServerSocket;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+/**
+ * Binds real listeners the way {@code server.port=0} and a fixed port do,
then checks the reported identity.
+ */
+class HeartbeatBoundPortTest {
+
+ @Test
+ void reportsDistinctBoundPortsForEphemeralAndFixedListeners() throws
Exception {
+ final WebServer first = start(0);
+ final WebServer second = start(0);
+ int fixedPort;
+ try (ServerSocket reserved = new ServerSocket(0)) {
+ fixedPort = reserved.getLocalPort();
+ }
+ final WebServer fixed = start(fixedPort);
+ final WebServer management = start(0);
+ try (MockedConstruction<ScheduledThreadPoolExecutor> ignoredExecutors
= mockConstruction(ScheduledThreadPoolExecutor.class);
+ MockedStatic<SystemInfoUtils> ignored =
mockStatic(SystemInfoUtils.class);
+ MockedStatic<RegisterUtils> register =
mockStatic(RegisterUtils.class)) {
+ register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
+ final HeartbeatListener ephemeral = listener();
+ final HeartbeatListener other = listener();
+ final HeartbeatListener explicit = listener();
+ publish(ephemeral, first, null);
+ publish(ephemeral, management,
BoundServerPort.MANAGEMENT_NAMESPACE);
+ publish(other, second, null);
+ publish(explicit, fixed, null);
+ final InstanceBeatInfoDTO firstBeat = report(ephemeral, register);
+ final InstanceBeatInfoDTO secondBeat = report(other, register);
+ final InstanceBeatInfoDTO fixedBeat = report(explicit, register);
+ assertTrue(first.getPort() > 0);
+ assertEquals(String.valueOf(first.getPort()),
firstBeat.getInstancePort());
+ assertEquals(String.valueOf(second.getPort()),
secondBeat.getInstancePort());
+ assertEquals(String.valueOf(fixedPort),
fixedBeat.getInstancePort());
+ assertNotEquals(firstBeat.getInstancePort(),
secondBeat.getInstancePort());
+ assertNotEquals(management.getPort(),
Integer.parseInt(firstBeat.getInstancePort()));
+ ephemeral.onShutdown();
+ other.onShutdown();
+ explicit.onShutdown();
+ } finally {
+ first.stop();
+ second.stop();
+ fixed.stop();
+ management.stop();
+ }
+ }
+
+ private InstanceBeatInfoDTO report(final HeartbeatListener listener, final
MockedStatic<RegisterUtils> register) throws Exception {
+ Method reportHeartbeat =
HeartbeatListener.class.getDeclaredMethod("reportHeartbeat");
+ reportHeartbeat.setAccessible(true);
+ reportHeartbeat.invoke(listener);
+ ArgumentCaptor<String> payload = ArgumentCaptor.forClass(String.class);
+ register.verify(() -> RegisterUtils.doHeartBeat(payload.capture(),
anyString(), anyString(), anyString()));
+ register.clearInvocations();
+ return GsonUtils.getInstance().fromJson(payload.getValue(),
InstanceBeatInfoDTO.class);
+ }
+
+ private void publish(final HeartbeatListener listener, final WebServer
server, final String namespace) {
+ ServletWebServerApplicationContext applicationContext =
mock(ServletWebServerApplicationContext.class);
+ when(applicationContext.getServerNamespace()).thenReturn(namespace);
+ listener.onWebServerInitialized(new
ServletWebServerInitializedEvent(server, applicationContext));
+ }
+
+ private HeartbeatListener listener() {
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ ShenyuBootstrapHeartBeatConfig config = new
ShenyuBootstrapHeartBeatConfig();
+ config.setServerLists("http://127.0.0.1:1";);
+ config.setProps(props);
+ ShenyuConfig shenyuConfig = new ShenyuConfig();
+ shenyuConfig.setNamespace("shenyu");
+ return new HeartbeatListener(config, shenyuConfig);
+ }
+
+ private WebServer start(final int port) {
+ WebServer server = new
TomcatServletWebServerFactory(port).getWebServer();
+ server.start();
+ return server;
+ }
+}
diff --git
a/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatListenerTest.java
b/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatListenerTest.java
index 311127594f..2c9c33a374 100644
---
a/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatListenerTest.java
+++
b/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatListenerTest.java
@@ -32,7 +32,9 @@ import org.mockito.MockedConstruction;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
-import org.springframework.boot.autoconfigure.web.ServerProperties;
+import org.springframework.boot.web.context.WebServerApplicationContext;
+import org.springframework.boot.web.context.WebServerInitializedEvent;
+import org.springframework.boot.web.server.WebServer;
import java.lang.reflect.Field;
import java.util.Optional;
@@ -43,9 +45,11 @@ import java.io.IOException;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.never;
@ExtendWith(MockitoExtension.class)
class HeartbeatListenerTest {
@@ -56,13 +60,10 @@ class HeartbeatListenerTest {
private ShenyuConfig shenyuConfig;
- private ServerProperties serverProperties;
-
@BeforeEach
void setUp() {
config = createMockConfig();
shenyuConfig = createMockShenyuConfig();
- serverProperties = createMockServerProperties();
}
private ShenyuBootstrapHeartBeatConfig createMockConfig() {
@@ -86,14 +87,6 @@ class HeartbeatListenerTest {
return config;
}
- private ServerProperties createMockServerProperties() {
-
- ServerProperties properties = new ServerProperties();
- properties.setPort(8080);
-
- return properties;
- }
-
@Test
void testPeriodicHeartbeatSurvivesFailuresOnEveryServer() throws
IOException {
try (MockedConstruction<ScheduledThreadPoolExecutor> executors =
Mockito.mockConstruction(ScheduledThreadPoolExecutor.class);
@@ -102,7 +95,8 @@ class HeartbeatListenerTest {
register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
register.when(() -> RegisterUtils.doHeartBeat(anyString(),
anyString(), anyString(), anyString()))
.thenThrow(new IOException("admin unavailable"));
- heartbeatListener = new HeartbeatListener(config, shenyuConfig,
serverProperties);
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
+ bind(heartbeatListener, null, 8080);
ArgumentCaptor<Runnable> periodic =
ArgumentCaptor.forClass(Runnable.class);
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
assertDoesNotThrow(periodic.getValue()::run);
@@ -120,7 +114,8 @@ class HeartbeatListenerTest {
MockedStatic<SystemInfoUtils> ignored =
Mockito.mockStatic(SystemInfoUtils.class);
MockedStatic<RegisterUtils> register =
Mockito.mockStatic(RegisterUtils.class)) {
register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
- heartbeatListener = new HeartbeatListener(config, shenyuConfig,
serverProperties);
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
+ bind(heartbeatListener, null, 8080);
ArgumentCaptor<Runnable> periodic =
ArgumentCaptor.forClass(Runnable.class);
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
periodic.getValue().run();
@@ -128,7 +123,92 @@ class HeartbeatListenerTest {
register.verify(() -> RegisterUtils.doHeartBeat(payload.capture(),
anyString(), anyString(), anyString()), Mockito.atLeastOnce());
InstanceBeatInfoDTO reported =
GsonUtils.getInstance().fromJson(payload.getAllValues().get(0),
InstanceBeatInfoDTO.class);
assertEquals(InstanceTypeConstants.BOOTSTRAP_INSTANCE_TYPE,
reported.getInstanceType());
+ assertEquals("8080", reported.getInstancePort());
+ }
+ }
+
+ @Test
+ void testDoesNotReportBeforeTheServerBinds() {
+ try (MockedConstruction<ScheduledThreadPoolExecutor> executors =
Mockito.mockConstruction(ScheduledThreadPoolExecutor.class);
+ MockedStatic<SystemInfoUtils> ignored =
Mockito.mockStatic(SystemInfoUtils.class);
+ MockedStatic<RegisterUtils> register =
Mockito.mockStatic(RegisterUtils.class)) {
+ register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
+ ArgumentCaptor<Runnable> periodic =
ArgumentCaptor.forClass(Runnable.class);
+
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
+ periodic.getValue().run();
+ register.verify(() -> RegisterUtils.doHeartBeat(anyString(),
anyString(), anyString(), anyString()), never());
+ assertEquals(1, executors.constructed().size());
+ }
+ }
+
+ @Test
+ void testReportsBoundPortInsteadOfConfiguredZeroAndIgnoresManagement() {
+ try (MockedConstruction<ScheduledThreadPoolExecutor> executors =
Mockito.mockConstruction(ScheduledThreadPoolExecutor.class);
+ MockedStatic<SystemInfoUtils> ignored =
Mockito.mockStatic(SystemInfoUtils.class);
+ MockedStatic<RegisterUtils> register =
Mockito.mockStatic(RegisterUtils.class)) {
+ register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
+ final HeartbeatListener first = new HeartbeatListener(config,
shenyuConfig);
+ final HeartbeatListener second = new HeartbeatListener(config,
shenyuConfig);
+ bind(first, "management", 19090);
+ bind(first, null, 0);
+ bind(first, null, 41234);
+ bind(second, null, 41235);
+ ArgumentCaptor<Runnable> periodic =
ArgumentCaptor.forClass(Runnable.class);
+
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
+
Mockito.verify(executors.constructed().get(1)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
+ periodic.getAllValues().get(0).run();
+ periodic.getAllValues().get(1).run();
+ ArgumentCaptor<String> payload =
ArgumentCaptor.forClass(String.class);
+ register.verify(() -> RegisterUtils.doHeartBeat(payload.capture(),
anyString(), anyString(), anyString()), Mockito.atLeast(2));
+ InstanceBeatInfoDTO reportedFirst =
GsonUtils.getInstance().fromJson(payload.getAllValues().get(0),
InstanceBeatInfoDTO.class);
+ InstanceBeatInfoDTO reportedSecond =
GsonUtils.getInstance().fromJson(payload.getAllValues().get(2),
InstanceBeatInfoDTO.class);
+ assertEquals("41234", reportedFirst.getInstancePort());
+ assertEquals("41235", reportedSecond.getInstancePort());
+ assertNotEquals(reportedFirst.getInstanceIp() + ":" +
reportedFirst.getInstancePort(),
+ reportedSecond.getInstanceIp() + ":" +
reportedSecond.getInstancePort());
+ assertEquals(2, executors.constructed().size());
+ }
+ }
+
+ @Test
+ void testShutdownBeforeInitializationDoesNotReport() {
+ try (MockedConstruction<ScheduledThreadPoolExecutor> executors =
Mockito.mockConstruction(ScheduledThreadPoolExecutor.class);
+ MockedStatic<RegisterUtils> register =
Mockito.mockStatic(RegisterUtils.class)) {
+ register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
+ ArgumentCaptor<Runnable> periodic =
ArgumentCaptor.forClass(Runnable.class);
+
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
+ heartbeatListener.onShutdown();
+ bind(heartbeatListener, null, 9195, true);
+ periodic.getValue().run();
+ register.verify(() -> RegisterUtils.doHeartBeat(anyString(),
anyString(), anyString(), anyString()), never());
+ Mockito.verify(executors.constructed().get(0), Mockito.times(1))
+ .scheduleAtFixedRate(Mockito.any(), Mockito.eq(0L),
Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
+ assertEquals(1, executors.constructed().size());
+ }
+ }
+
+ private void bind(final HeartbeatListener listener, final String
namespace, final int port) {
+ bind(listener, namespace, port, false);
+ }
+
+ private void bind(final HeartbeatListener listener, final String
namespace, final int port, final boolean lenient) {
+ WebServer webServer = Mockito.mock(WebServer.class);
+ WebServerApplicationContext applicationContext =
Mockito.mock(WebServerApplicationContext.class);
+ WebServerInitializedEvent event =
Mockito.mock(WebServerInitializedEvent.class);
+ if (lenient) {
+ Mockito.lenient().when(webServer.getPort()).thenReturn(port);
+
Mockito.lenient().when(applicationContext.getServerNamespace()).thenReturn(namespace);
+ Mockito.lenient().when(event.getWebServer()).thenReturn(webServer);
+
Mockito.lenient().when(event.getApplicationContext()).thenReturn(applicationContext);
+ } else {
+ Mockito.when(webServer.getPort()).thenReturn(port);
+
Mockito.when(applicationContext.getServerNamespace()).thenReturn(namespace);
+ Mockito.when(event.getWebServer()).thenReturn(webServer);
+
Mockito.when(event.getApplicationContext()).thenReturn(applicationContext);
}
+ listener.onWebServerInitialized(event);
}
@Test
@@ -139,7 +219,7 @@ class HeartbeatListenerTest {
.thenReturn(Optional.of("mock-token"));
assertDoesNotThrow(() -> {
- heartbeatListener = new HeartbeatListener(config,
shenyuConfig, serverProperties);
+ heartbeatListener = new HeartbeatListener(config,
shenyuConfig);
});
assertNotNull(heartbeatListener);
@@ -164,7 +244,7 @@ class HeartbeatListenerTest {
.thenReturn(Optional.of("mock-token"));
assertDoesNotThrow(() -> {
- heartbeatListener = new
HeartbeatListener(configWithEncryption, shenyuConfig, serverProperties);
+ heartbeatListener = new
HeartbeatListener(configWithEncryption, shenyuConfig);
});
assertNotNull(heartbeatListener);
@@ -181,7 +261,7 @@ class HeartbeatListenerTest {
registerUtilsMockedStatic.when(() ->
RegisterUtils.doHeartBeat(anyString(), anyString(), anyString(), anyString()))
.then(invocation -> null);
- heartbeatListener = new HeartbeatListener(config, shenyuConfig,
serverProperties);
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
// Use reflection to access private method sendHeartbeat
java.lang.reflect.Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
@@ -215,7 +295,7 @@ class HeartbeatListenerTest {
registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
.thenReturn(Optional.empty());
- heartbeatListener = new HeartbeatListener(config, shenyuConfig,
serverProperties);
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
// Use reflection to access private method sendHeartbeat
java.lang.reflect.Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
@@ -239,7 +319,7 @@ class HeartbeatListenerTest {
registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
.thenReturn(Optional.of("mock-token"));
- heartbeatListener = new HeartbeatListener(config, shenyuConfig,
serverProperties);
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
// Use reflection to access the executor
Field executorField =
HeartbeatListener.class.getDeclaredField("executor");
@@ -262,7 +342,7 @@ class HeartbeatListenerTest {
registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
.thenReturn(Optional.of("mock-token"));
- heartbeatListener = new HeartbeatListener(config, shenyuConfig,
serverProperties);
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
// Use reflection to check private fields
Field usernameField =
HeartbeatListener.class.getDeclaredField("username");
@@ -301,7 +381,7 @@ class HeartbeatListenerTest {
Mockito.contains("localhost:9096"), anyString(),
anyString()))
.then(invocation -> null);
- heartbeatListener = new HeartbeatListener(config, shenyuConfig,
serverProperties);
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
// Use reflection to access private method sendHeartbeat
java.lang.reflect.Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
@@ -332,7 +412,7 @@ class HeartbeatListenerTest {
registerUtilsMockedStatic.when(() ->
RegisterUtils.doHeartBeat(anyString(), anyString(), anyString(), anyString()))
.thenThrow(new RuntimeException("Connection failed"));
- heartbeatListener = new HeartbeatListener(config, shenyuConfig,
serverProperties);
+ heartbeatListener = new HeartbeatListener(config, shenyuConfig);
// Use reflection to access private method sendHeartbeat
java.lang.reflect.Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-client/shenyu-spring-boot-starter-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListenerConfiguration.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-client/shenyu-spring-boot-starter-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListenerConfiguration.java
index 2fd1a464f3..dfbbe4eaaf 100644
---
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-client/shenyu-spring-boot-starter-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListenerConfiguration.java
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-client/shenyu-spring-boot-starter-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListenerConfiguration.java
@@ -19,7 +19,6 @@ package org.apache.shenyu.register.client.beat;
import org.apache.shenyu.common.config.ShenyuConfig;
import
org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
-import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -37,14 +36,12 @@ public class HeartbeatListenerConfiguration {
*
* @param shenyuBootstrapHeartBeatConfig the shenyuBootstrapHeartBeatConfig
* @param shenyuConfig the shenyu config
- * @param serverProperties the server properties
* @return the heartbeat bean listener.
*/
@Bean
public HeartbeatListener heartbeatListener(final
ShenyuBootstrapHeartBeatConfig shenyuBootstrapHeartBeatConfig,
- final ShenyuConfig shenyuConfig,
- final ServerProperties
serverProperties) {
- return new HeartbeatListener(shenyuBootstrapHeartBeatConfig,
shenyuConfig, serverProperties);
+ final ShenyuConfig
shenyuConfig) {
+ return new HeartbeatListener(shenyuBootstrapHeartBeatConfig,
shenyuConfig);
}
/**
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-websocket/src/main/java/org/apache/shenyu/springboot/starter/sync/data/websocket/WebsocketSyncDataConfiguration.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-websocket/src/main/java/org/apache/shenyu/springboot/starter/sync/data/websocket/WebsocketSyncDataConfiguration.java
index f6047ff711..d2bfe143e3 100644
---
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-websocket/src/main/java/org/apache/shenyu/springboot/starter/sync/data/websocket/WebsocketSyncDataConfiguration.java
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-sync-data-center/shenyu-spring-boot-starter-sync-data-websocket/src/main/java/org/apache/shenyu/springboot/starter/sync/data/websocket/WebsocketSyncDataConfiguration.java
@@ -31,7 +31,6 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -62,7 +61,6 @@ public class WebsocketSyncDataConfiguration {
* @param proxySelectorSubscribers the proxySelector subscribers
* @param discoveryUpstreamSubscribers the discoveryUpstream subscribers
* @param aiProxyApiKeySubscribers the ai proxy api key subscribers
- * @param serverProperties the serverProperties
* @return the sync data service
*/
@Bean
@@ -75,8 +73,7 @@ public class WebsocketSyncDataConfiguration {
final ObjectProvider<List<ProxySelectorDataSubscriber>>
proxySelectorSubscribers,
final ObjectProvider<List<DiscoveryUpstreamDataSubscriber>>
discoveryUpstreamSubscribers,
- final ObjectProvider<List<AiProxyApiKeyDataSubscriber>>
aiProxyApiKeySubscribers,
- final ServerProperties serverProperties) {
+ final ObjectProvider<List<AiProxyApiKeyDataSubscriber>>
aiProxyApiKeySubscribers) {
LOGGER.info("you use websocket sync shenyu data.......");
return new WebsocketSyncDataService(
websocketConfig.getIfAvailable(WebsocketConfig::new),
@@ -86,8 +83,7 @@ public class WebsocketSyncDataConfiguration {
authSubscribers.getIfAvailable(Collections::emptyList),
proxySelectorSubscribers.getIfAvailable(Collections::emptyList),
discoveryUpstreamSubscribers.getIfAvailable(Collections::emptyList),
-
aiProxyApiKeySubscribers.getIfAvailable(Collections::emptyList),
- serverProperties);
+
aiProxyApiKeySubscribers.getIfAvailable(Collections::emptyList));
}
/**
diff --git a/shenyu-sync-data-center/shenyu-sync-data-http/pom.xml
b/shenyu-sync-data-center/shenyu-sync-data-http/pom.xml
index 3639fda2cf..7525caf9a5 100644
--- a/shenyu-sync-data-center/shenyu-sync-data-http/pom.xml
+++ b/shenyu-sync-data-center/shenyu-sync-data-http/pom.xml
@@ -60,5 +60,10 @@
<artifactId>spring-web</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot</artifactId>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-http/src/main/java/org/apache/shenyu/sync/data/http/HttpSyncDataService.java
b/shenyu-sync-data-center/shenyu-sync-data-http/src/main/java/org/apache/shenyu/sync/data/http/HttpSyncDataService.java
index 75ecd3520f..ab26186f65 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-http/src/main/java/org/apache/shenyu/sync/data/http/HttpSyncDataService.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-http/src/main/java/org/apache/shenyu/sync/data/http/HttpSyncDataService.java
@@ -36,6 +36,7 @@ import
org.apache.shenyu.common.constant.InstanceTypeConstants;
import org.apache.shenyu.common.dto.ConfigData;
import org.apache.shenyu.common.enums.ConfigGroupEnum;
import org.apache.shenyu.common.exception.ShenyuException;
+import org.apache.shenyu.common.utils.BoundServerPort;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.common.utils.SystemInfoUtils;
import org.apache.shenyu.common.utils.ThreadUtils;
@@ -50,7 +51,8 @@ import org.apache.shenyu.sync.data.http.config.HttpConfig;
import org.apache.shenyu.sync.data.http.refresh.DataRefreshFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.web.context.WebServerInitializedEvent;
+import org.springframework.context.event.EventListener;
import org.springframework.util.Assert;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
@@ -60,6 +62,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
+import java.util.OptionalInt;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
@@ -90,8 +93,10 @@ public class HttpSyncDataService implements SyncDataService {
private final ShenyuConfig shenyuConfig;
- @Value("${server.port}")
- private int port;
+ /**
+ * Port of the application listener. Zero until the server initialization
event.
+ */
+ private volatile int boundPort;
public HttpSyncDataService(final HttpConfig httpConfig,
final PluginDataSubscriber pluginDataSubscriber,
@@ -118,6 +123,30 @@ public class HttpSyncDataService implements
SyncDataService {
this.start();
}
+ /**
+ * Record the application listener port. The management server and any
port that is not yet bound
+ * are ignored, so long polling does not publish a placeholder.
+ *
+ * @param event web server initialization event
+ */
+ @EventListener
+ public void onWebServerInitialized(final WebServerInitializedEvent event) {
+ if (Objects.isNull(event.getWebServer()) ||
Objects.isNull(event.getApplicationContext())) {
+ return;
+ }
+ OptionalInt resolved =
BoundServerPort.resolve(event.getApplicationContext().getServerNamespace(),
event.getWebServer().getPort());
+ if (resolved.isEmpty()) {
+ LOG.debug("Ignore http sync port from namespace {} port {}",
event.getApplicationContext().getServerNamespace(),
event.getWebServer().getPort());
+ return;
+ }
+ int observed = resolved.getAsInt();
+ if (boundPort == observed) {
+ return;
+ }
+ boundPort = observed;
+ LOG.info("Application server bound on port {}, http sync will report
it", observed);
+ }
+
private void start() {
// It could be initialized multiple times, so you need to control that.
if (RUNNING.compareAndSet(false, true)) {
@@ -212,12 +241,15 @@ public class HttpSyncDataService implements
SyncDataService {
}
params.put("namespaceId",
Lists.newArrayList(shenyuConfig.getNamespace()));
LOG.debug("listener params: [{}]", params);
- Headers headers = new Headers.Builder()
+ Headers.Builder headerBuilder = new Headers.Builder()
.add(Constants.X_ACCESS_TOKEN,
this.accessTokenManager.getAccessToken())
.add("Content-Type", "application/x-www-form-urlencoded")
- .add("X-Real-PORT", port + "")
- .add(InstanceTypeConstants.BOOTSTRAP_INSTANCE_INFO,
SystemInfoUtils.getSystemInfo())
- .build();
+ .add(InstanceTypeConstants.BOOTSTRAP_INSTANCE_INFO,
SystemInfoUtils.getSystemInfo());
+ int port = boundPort;
+ if (port > 0) {
+ headerBuilder.add("X-Real-PORT", String.valueOf(port));
+ }
+ Headers headers = headerBuilder.build();
String listenerUrl = server +
Constants.SHENYU_ADMIN_PATH_CONFIGS_LISTENER;
String uri =
UriComponentsBuilder.fromHttpUrl(listenerUrl).queryParams(params).build(true).toUriString();
Request request = new Request.Builder()
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-http/src/test/java/org/apache/shenyu/sync/data/http/HttpSyncBoundPortTest.java
b/shenyu-sync-data-center/shenyu-sync-data-http/src/test/java/org/apache/shenyu/sync/data/http/HttpSyncBoundPortTest.java
new file mode 100644
index 0000000000..791560bbc1
--- /dev/null
+++
b/shenyu-sync-data-center/shenyu-sync-data-http/src/test/java/org/apache/shenyu/sync/data/http/HttpSyncBoundPortTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.http;
+
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.shenyu.common.config.ShenyuConfig;
+import org.apache.shenyu.common.utils.BoundServerPort;
+import org.apache.shenyu.sync.data.api.PluginDataSubscriber;
+import org.apache.shenyu.sync.data.http.config.HttpConfig;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.web.context.WebServerApplicationContext;
+import org.springframework.boot.web.context.WebServerInitializedEvent;
+import org.springframework.boot.web.server.WebServer;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.net.ServerSocket;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * HTTP long polling reports the bound application port and omits a
placeholder.
+ */
+class HttpSyncBoundPortTest {
+
+ @Test
+ void reportsDistinctBoundPortsAndSkipsPlaceholders() throws Exception {
+ try (ServerSocket firstSocket = new ServerSocket(0);
+ ServerSocket secondSocket = new ServerSocket(0);
+ ServerSocket fixedSocket = new ServerSocket(0)) {
+ final int firstPort = firstSocket.getLocalPort();
+ final int secondPort = secondSocket.getLocalPort();
+ final int fixedPort = fixedSocket.getLocalPort();
+ assertTrue(firstPort > 0);
+ assertNotEquals(firstPort, secondPort);
+ HttpSyncDataService first = service();
+ HttpSyncDataService second = service();
+ try {
+ assertNull(poll(first).header("X-Real-PORT"));
+ publish(first, 0, null);
+ publish(first, secondPort,
BoundServerPort.MANAGEMENT_NAMESPACE);
+ assertNull(poll(first).header("X-Real-PORT"));
+ publish(first, firstPort, null);
+ publish(second, secondPort, null);
+ publish(second, fixedPort, "server");
+ assertEquals(String.valueOf(firstPort),
poll(first).header("X-Real-PORT"));
+ assertEquals(String.valueOf(fixedPort),
poll(second).header("X-Real-PORT"));
+ assertNotEquals(poll(first).header("X-Real-PORT"),
poll(second).header("X-Real-PORT"));
+ } finally {
+ first.close();
+ second.close();
+ }
+ }
+ }
+
+ private Request poll(final HttpSyncDataService service) {
+ AtomicReference<Request> captured = new AtomicReference<>();
+ OkHttpClient client = (OkHttpClient)
ReflectionTestUtils.getField(service, "okHttpClient");
+
when(Objects.requireNonNull(client).newCall(any(Request.class))).thenAnswer(invocation
-> {
+ captured.set(invocation.getArgument(0));
+ return listenerCall();
+ });
+ ReflectionTestUtils.invokeMethod(service, "doLongPolling",
"http://127.0.0.1:1";);
+ return Objects.requireNonNull(captured.get());
+ }
+
+ private void publish(final HttpSyncDataService service, final int port,
final String namespace) {
+ WebServer webServer = mock(WebServer.class);
+ when(webServer.getPort()).thenReturn(port);
+ WebServerApplicationContext applicationContext =
mock(WebServerApplicationContext.class);
+ when(applicationContext.getServerNamespace()).thenReturn(namespace);
+ WebServerInitializedEvent event =
mock(WebServerInitializedEvent.class);
+ when(event.getWebServer()).thenReturn(webServer);
+ when(event.getApplicationContext()).thenReturn(applicationContext);
+ service.onWebServerInitialized(event);
+ }
+
+ private HttpSyncDataService service() throws Exception {
+ OkHttpClient client = mock(OkHttpClient.class);
+ when(client.newCall(any(Request.class))).thenAnswer(invocation ->
fetchCall());
+ HttpConfig httpConfig = new HttpConfig();
+ httpConfig.setUrl("http://127.0.0.1:1";);
+ ShenyuConfig shenyuConfig = new ShenyuConfig();
+ shenyuConfig.setNamespace("shenyu");
+ AccessTokenManager accessTokenManager = mock(AccessTokenManager.class);
+ when(accessTokenManager.getAccessToken()).thenReturn("token");
+ HttpSyncDataService service = new HttpSyncDataService(httpConfig,
mock(PluginDataSubscriber.class), client,
+ Collections.emptyList(), Collections.emptyList(),
Collections.emptyList(), Collections.emptyList(),
+ Collections.emptyList(), accessTokenManager, shenyuConfig);
+ ExecutorService executor = (ExecutorService)
ReflectionTestUtils.getField(service, "executor");
+ service.close();
+ if (Objects.nonNull(executor)) {
+ executor.awaitTermination(2, TimeUnit.SECONDS);
+ }
+ return service;
+ }
+
+ private Call fetchCall() throws Exception {
+ Call call = mock(Call.class);
+ Response response = mock(Response.class);
+ ResponseBody body = mock(ResponseBody.class);
+ when(response.isSuccessful()).thenReturn(true);
+ when(response.body()).thenReturn(body);
+ when(body.string()).thenReturn("{\"code\":200,\"data\":{}}");
+ when(call.execute()).thenReturn(response);
+ return call;
+ }
+
+ private Call listenerCall() throws Exception {
+ Call call = mock(Call.class);
+ Response response = mock(Response.class);
+ ResponseBody body = mock(ResponseBody.class);
+ when(response.isSuccessful()).thenReturn(true);
+ when(response.body()).thenReturn(body);
+ when(body.string()).thenReturn("{\"data\":null}");
+ when(call.execute()).thenReturn(response);
+ return call;
+ }
+}
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 76f3dc8946..e006e7f804 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
@@ -23,6 +23,7 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.shenyu.common.config.ShenyuConfig;
import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.utils.BoundServerPort;
import org.apache.shenyu.common.enums.RunningModeEnum;
import org.apache.shenyu.common.timer.AbstractRoundTask;
import org.apache.shenyu.common.timer.TaskEntity;
@@ -39,9 +40,11 @@ import
org.apache.shenyu.sync.data.api.ProxySelectorDataSubscriber;
import org.apache.shenyu.sync.data.api.SyncDataService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.web.ServerProperties;
+import org.springframework.boot.web.context.WebServerInitializedEvent;
+import org.springframework.context.event.EventListener;
import java.net.URI;
+import java.util.OptionalInt;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashSet;
@@ -93,7 +96,10 @@ public class WebsocketSyncDataService implements
SyncDataService {
private boolean closed;
- private final ServerProperties serverProperties;
+ /**
+ * Port of the application listener. Zero until the server initialization
event.
+ */
+ private volatile int boundPort;
/**
* Instantiates a new Websocket sync cache.
@@ -105,7 +111,7 @@ public class WebsocketSyncDataService implements
SyncDataService {
* @param authDataSubscribers the auth data subscribers
* @param proxySelectorDataSubscribers the proxy selector data subscribers
* @param discoveryUpstreamDataSubscribers the discovery upstream data
subscribers
- * @param serverProperties serverProperties
+ * @param aiProxyApiKeyDataSubscribers the ai proxy api key subscribers
*/
public WebsocketSyncDataService(
final WebsocketConfig websocketConfig,
@@ -116,8 +122,7 @@ public class WebsocketSyncDataService implements
SyncDataService {
final List<ProxySelectorDataSubscriber>
proxySelectorDataSubscribers,
final List<DiscoveryUpstreamDataSubscriber>
discoveryUpstreamDataSubscribers,
final
List<org.apache.shenyu.sync.data.api.AiProxyApiKeyDataSubscriber>
- aiProxyApiKeyDataSubscribers,
- final ServerProperties serverProperties) {
+ aiProxyApiKeyDataSubscribers) {
this.timer = WheelTimerFactory.newWheelTimer();
this.websocketConfig = websocketConfig;
this.pluginDataSubscriber = pluginDataSubscriber;
@@ -127,12 +132,6 @@ public class WebsocketSyncDataService implements
SyncDataService {
this.discoveryUpstreamDataSubscribers =
discoveryUpstreamDataSubscribers;
this.aiProxyApiKeyDataSubscribers = aiProxyApiKeyDataSubscribers;
this.namespaceId = shenyuConfig.getNamespace();
- this.serverProperties = serverProperties;
- LOG.info("start init connecting...");
- List<String> urls = websocketConfig.getUrls();
- for (String url : urls) {
- clients.add(createClient(url));
- }
LOG.info("start check task...");
long intervalSeconds = websocketConfig.getMasterCheckInterval() <= 0 ?
60 : websocketConfig.getMasterCheckInterval();
this.timer.add(timerTask = new AbstractRoundTask(null,
TimeUnit.SECONDS.toMillis(intervalSeconds)) {
@@ -151,12 +150,7 @@ public class WebsocketSyncDataService implements
SyncDataService {
LOG.debug("master checking task start...");
}
if (CollectionUtils.isEmpty(clients)) {
- List<String> urls = websocketConfig.getUrls();
- if (CollectionUtils.isNotEmpty(urls)) {
- for (String url : urls) {
- clients.add(createClient(url));
- }
- }
+ addSeedClients();
}
Set<URI> masterUrls = new LinkedHashSet<>();
Iterator<ShenyuWebsocketClient> iterator = clients.iterator();
@@ -188,7 +182,7 @@ public class WebsocketSyncDataService implements
SyncDataService {
}
}
for (URI masterUri : masterUrls) {
- if (clients.stream().noneMatch(client ->
masterUri.equals(client.getURI()))) {
+ if (boundPort > 0 && clients.stream().noneMatch(client ->
masterUri.equals(client.getURI()))) {
try {
clients.add(createClient(masterUri.toString()));
} catch (RuntimeException ex) {
@@ -197,12 +191,49 @@ 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));
- }
- }
+ addSeedClients();
+ }
+ }
+
+ /**
+ * Connect using the application listener port. A management server is not
the gateway listener,
+ * and clients stay disconnected until a positive bound port is known.
Closing the service first
+ * discards a later initialization event.
+ *
+ * @param event web server initialization event
+ */
+ @EventListener
+ public synchronized void onWebServerInitialized(final
WebServerInitializedEvent event) {
+ if (closed || Objects.isNull(event.getWebServer()) ||
Objects.isNull(event.getApplicationContext())) {
+ return;
+ }
+ OptionalInt resolved =
BoundServerPort.resolve(event.getApplicationContext().getServerNamespace(),
event.getWebServer().getPort());
+ if (resolved.isEmpty()) {
+ LOG.debug("Ignore websocket client port from namespace {} port
{}", event.getApplicationContext().getServerNamespace(),
event.getWebServer().getPort());
+ return;
+ }
+ int observed = resolved.getAsInt();
+ if (boundPort == observed) {
+ return;
+ }
+ boundPort = observed;
+ LOG.info("Application server bound on port {}, websocket sync will
report it", observed);
+ if (CollectionUtils.isEmpty(clients)) {
+ addSeedClients();
+ }
+ }
+
+ private void addSeedClients() {
+ if (boundPort <= 0) {
+ return;
+ }
+ List<String> urls = websocketConfig.getUrls();
+ if (CollectionUtils.isEmpty(urls)) {
+ return;
+ }
+ LOG.info("start init connecting on port {}...", boundPort);
+ for (String url : urls) {
+ clients.add(createClient(url));
}
}
@@ -291,7 +322,7 @@ public class WebsocketSyncDataService implements
SyncDataService {
discoveryUpstreamDataSubscribers,
this.aiProxyApiKeyDataSubscribers,
namespaceId,
- serverProperties.getPort(),
+ boundPort,
websocketConfig.isInitialSyncReadiness() ? initialSyncReady :
null,
this::onMasterLost);
}
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketBoundPortTest.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketBoundPortTest.java
new file mode 100644
index 0000000000..cafb5e4315
--- /dev/null
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketBoundPortTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.sync.data.websocket;
+
+import org.apache.shenyu.common.config.ShenyuConfig;
+import org.apache.shenyu.common.utils.BoundServerPort;
+import
org.apache.shenyu.plugin.sync.data.websocket.client.ShenyuWebsocketClient;
+import org.apache.shenyu.plugin.sync.data.websocket.config.WebsocketConfig;
+import org.apache.shenyu.sync.data.api.PluginDataSubscriber;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import
org.springframework.boot.web.embedded.tomcat.TomcatServletWebServerFactory;
+import org.springframework.boot.web.server.WebServer;
+import
org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
+import
org.springframework.boot.web.servlet.context.ServletWebServerInitializedEvent;
+
+import java.lang.reflect.Field;
+import java.net.ServerSocket;
+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.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.when;
+
+/**
+ * Websocket clients report the port Tomcat actually bound, including {@code
server.port=0}.
+ */
+class WebsocketBoundPortTest {
+
+ @Test
+ void waitsForTheBoundPortAndKeepsDistinctIdentities() throws Exception {
+ final WebServer ephemeral = start(0);
+ final WebServer other = start(0);
+ int fixedPort;
+ try (ServerSocket reserved = new ServerSocket(0)) {
+ fixedPort = reserved.getLocalPort();
+ }
+ final WebServer fixed = start(fixedPort);
+ final WebServer management = start(0);
+ List<Integer> reportedPorts = new ArrayList<>();
+ try (MockedConstruction<ShenyuWebsocketClient> clients =
mockConstruction(ShenyuWebsocketClient.class,
+ (mock, context) -> reportedPorts.add((Integer)
context.arguments().get(9)))) {
+ final WebsocketSyncDataService first = service();
+ assertTrue(clientList(first).isEmpty());
+ publish(first, management, BoundServerPort.MANAGEMENT_NAMESPACE);
+ publish(first, ephemeral, null);
+ final WebsocketSyncDataService second = service();
+ publish(second, other, null);
+ WebsocketSyncDataService explicit = service();
+ publish(explicit, fixed, null);
+ assertEquals(1, clientList(first).size());
+ assertEquals(ephemeral.getPort(), reportedPorts.get(0));
+ assertEquals(other.getPort(), reportedPorts.get(1));
+ assertEquals(fixedPort, reportedPorts.get(2));
+ assertNotEquals(ephemeral.getPort(), other.getPort());
+ assertNotEquals(management.getPort(), reportedPorts.get(0));
+ first.close();
+ publish(first, ephemeral, null);
+ assertEquals(3, clients.constructed().size());
+ second.close();
+ explicit.close();
+ } finally {
+ ephemeral.stop();
+ other.stop();
+ fixed.stop();
+ management.stop();
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private List<ShenyuWebsocketClient> clientList(final
WebsocketSyncDataService service) throws Exception {
+ Field field =
WebsocketSyncDataService.class.getDeclaredField("clients");
+ field.setAccessible(true);
+ return (List<ShenyuWebsocketClient>) field.get(service);
+ }
+
+ private void publish(final WebsocketSyncDataService service, final
WebServer server, final String namespace) {
+ ServletWebServerApplicationContext applicationContext =
mock(ServletWebServerApplicationContext.class);
+ when(applicationContext.getServerNamespace()).thenReturn(namespace);
+ service.onWebServerInitialized(new
ServletWebServerInitializedEvent(server, applicationContext));
+ }
+
+ private WebsocketSyncDataService service() {
+ WebsocketConfig config = new WebsocketConfig();
+
config.setUrls(Collections.singletonList("ws://127.0.0.1:1/websocket"));
+ ShenyuConfig shenyuConfig = new ShenyuConfig();
+ shenyuConfig.setNamespace("shenyu");
+ return new WebsocketSyncDataService(config, shenyuConfig,
mock(PluginDataSubscriber.class),
+ Collections.emptyList(), Collections.emptyList(),
Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
+ }
+
+ private WebServer start(final int port) {
+ WebServer server = new
TomcatServletWebServerFactory(port).getWebServer();
+ server.start();
+ return server;
+ }
+}
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketMasterRedirectTest.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketMasterRedirectTest.java
index 922085d44e..80f71ebc84 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketMasterRedirectTest.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketMasterRedirectTest.java
@@ -36,7 +36,9 @@ import org.junit.jupiter.params.provider.NullAndEmptySource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
-import org.springframework.boot.autoconfigure.web.ServerProperties;
+import
org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
+import
org.springframework.boot.web.servlet.context.ServletWebServerInitializedEvent;
+import org.springframework.boot.web.server.WebServer;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
@@ -208,15 +210,18 @@ class WebsocketMasterRedirectTest {
config.setAllowOrigin("https://gateway.example";);
ShenyuConfig shenyu = new ShenyuConfig();
shenyu.setNamespace("cluster-test");
- ServerProperties properties = new ServerProperties();
- properties.setPort(9195);
PluginDataSubscriber subscriber = mock(PluginDataSubscriber.class);
doAnswer(invocation -> {
applied.countDown();
return null;
}).when(subscriber).onSubscribe(any());
service = new WebsocketSyncDataService(config, shenyu, subscriber,
Collections.emptyList(), Collections.emptyList(),
- Collections.emptyList(), Collections.emptyList(),
Collections.emptyList(), properties);
+ Collections.emptyList(), Collections.emptyList(),
Collections.emptyList());
+ WebServer webServer = mock(WebServer.class);
+ when(webServer.getPort()).thenReturn(9195);
+ ServletWebServerApplicationContext applicationContext =
mock(ServletWebServerApplicationContext.class);
+ when(applicationContext.getServerNamespace()).thenReturn(null);
+ service.onWebServerInitialized(new
ServletWebServerInitializedEvent(webServer, applicationContext));
}
private TestServer startServer(final boolean master) throws
InterruptedException {
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 92428f9e31..e1b2cd99d4 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
@@ -32,7 +32,6 @@ import
org.apache.shenyu.sync.data.api.ProxySelectorDataSubscriber;
import org.junit.jupiter.api.Test;
import org.mockito.InOrder;
import org.mockito.MockedStatic;
-import org.springframework.boot.autoconfigure.web.ServerProperties;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
@@ -150,8 +149,7 @@ public final class WebsocketSyncDataServiceTest {
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
- Collections.emptyList(),
- mock(ServerProperties.class));
+ Collections.emptyList());
try {
Field timerTaskField =
WebsocketSyncDataService.class.getDeclaredField("timerTask");
timerTaskField.setAccessible(true);
@@ -178,8 +176,7 @@ public final class WebsocketSyncDataServiceTest {
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
- Collections.emptyList(),
- mock(ServerProperties.class));
+ Collections.emptyList());
Field clientsField =
WebsocketSyncDataService.class.getDeclaredField("clients");
clientsField.setAccessible(true);
List<ShenyuWebsocketClient> clients = (List<ShenyuWebsocketClient>)
clientsField.get(service);
@@ -187,6 +184,9 @@ public final class WebsocketSyncDataServiceTest {
ShenyuWebsocketClient closedClient = mock(ShenyuWebsocketClient.class);
when(closedClient.isOpen()).thenReturn(false);
clients.add(closedClient);
+ Field boundPort =
WebsocketSyncDataService.class.getDeclaredField("boundPort");
+ boundPort.setAccessible(true);
+ boundPort.setInt(service, 9195);
try {
Method masterCheck =
WebsocketSyncDataService.class.getDeclaredMethod("masterCheck");
@@ -230,7 +230,6 @@ public final class WebsocketSyncDataServiceTest {
Collections.<AuthDataSubscriber>emptyList(),
Collections.<ProxySelectorDataSubscriber>emptyList(),
Collections.<DiscoveryUpstreamDataSubscriber>emptyList(),
- Collections.<AiProxyApiKeyDataSubscriber>emptyList(),
- mock(ServerProperties.class));
+ Collections.<AiProxyApiKeyDataSubscriber>emptyList());
}
}
