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 4913433cd54a..ddf5878055b2 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 @@ -43,11 +43,20 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.shenyu.common.concurrent.MemorySafeTaskQueue; +import org.apache.shenyu.common.concurrent.ShenyuThreadFactory; +import org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor; + import java.net.URI; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; /** * The type shenyu websocket client. @@ -58,23 +67,48 @@ public final class ShenyuWebsocketClient extends WebSocketClient { * logger. */ private static final Logger LOG = LoggerFactory.getLogger(ShenyuWebsocketClient.class); - + + private static final int RECONNECT_EXECUTOR_CORE_POOL_SIZE = 1; + + private static final int RECONNECT_EXECUTOR_MAX_POOL_SIZE = 8; + + private static final long RECONNECT_EXECUTOR_KEEP_ALIVE_MS = TimeUnit.SECONDS.toMillis(60); + + private static final ExecutorService RECONNECT_EXECUTOR = new ShenyuThreadPoolExecutor( + RECONNECT_EXECUTOR_CORE_POOL_SIZE, + RECONNECT_EXECUTOR_MAX_POOL_SIZE, + RECONNECT_EXECUTOR_KEEP_ALIVE_MS, + TimeUnit.MILLISECONDS, + new MemorySafeTaskQueue<>(Constants.THE_256_MB), + ShenyuThreadFactory.create("websocket-reconnect", true), + new ThreadPoolExecutor.AbortPolicy()); + + private static final long MIN_RECONNECT_BACKOFF_MS = TimeUnit.SECONDS.toMillis(1); + + private static final long MAX_RECONNECT_BACKOFF_MS = TimeUnit.SECONDS.toMillis(60); + private volatile boolean alreadySync = Boolean.FALSE; - + private final WebsocketDataHandler websocketDataHandler; - + private final Timer timer; - + private TimerTask timerTask; - + private String runningMode; - + private String masterUrl; - + private volatile boolean isConnectedToMaster; - + private final String namespaceId; + private final AtomicBoolean reconnecting = new AtomicBoolean(false); + + private volatile long lastReconnectAttemptTime; + + private final AtomicInteger reconnectBackoff = new AtomicInteger(0); + /** * Instantiates a new shenyu websocket client. * @@ -238,11 +272,13 @@ public void nowClose() { private void healthCheck() { try { if (!this.isOpen()) { - this.reconnectBlocking(); + if (this.reconnecting.compareAndSet(false, true)) { + RECONNECT_EXECUTOR.submit(this::doReconnect); + } } else { + this.reconnectBackoff.set(0); this.sendPing(); send(getInstanceInfo()); -// send(DataEventTypeEnum.RUNNING_MODE.name()); LOG.debug("websocket send to [{}] ping message successful", this.getURI()); } } catch (Exception e) { @@ -250,6 +286,40 @@ private void healthCheck() { } } + private void doReconnect() { + try { + long backoff = calculateBackoff(); + long since = System.currentTimeMillis() - lastReconnectAttemptTime; + long waitMs = backoff - since; + if (waitMs > 0) { + Thread.sleep(waitMs); + } + try { + this.reconnectBlocking(); + } finally { + lastReconnectAttemptTime = System.currentTimeMillis(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (Exception e) { + reconnectBackoff.set(Math.min(reconnectBackoff.get() + 1, 10)); + LOG.error("websocket reconnect server[{}] error", this.getURI(), e); + } finally { + this.reconnecting.set(false); + } + } + + private long calculateBackoff() { + int failures = reconnectBackoff.get(); + if (failures <= 0) { + return 0; + } + long base = Math.min( + MIN_RECONNECT_BACKOFF_MS * (1L << Math.min(failures - 1, 10)), + MAX_RECONNECT_BACKOFF_MS); + return base + (long) (base * 0.5 * ThreadLocalRandom.current().nextDouble()); + } + private String getInstanceInfo() { // Combine instance and host information Map combinedInfo = Map.of( 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 0936a435dd09..dcb42b7db6c0 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 @@ -35,43 +35,56 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.net.URI; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import org.mockito.Answers; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.withSettings; /** * add test case for {@link ShenyuWebsocketClient}. */ @ExtendWith(MockitoExtension.class) public class ShenyuWebsocketClientTest { - + @InjectMocks private ShenyuWebsocketClient shenyuWebsocketClient; - + @Mock private URI serverUri; - + @Mock private PluginDataSubscriber pluginDataSubscriber; - + @Mock private List metaDataSubscribers; - + @Mock private List authDataSubscribers; - + @Mock private ScheduledThreadPoolExecutor executor; - + private WebsocketData websocketData; - + @BeforeEach public void setUp() { websocketData = new WebsocketData<>(); @@ -83,7 +96,7 @@ public void setUp() { list.add(pluginData); websocketData.setData(list); } - + @Test public void testOnOpen() { shenyuWebsocketClient = spy(shenyuWebsocketClient); @@ -94,7 +107,7 @@ public void testOnOpen() { verify(shenyuWebsocketClient).send(DataEventTypeEnum.RUNNING_MODE.name()); verify(shenyuWebsocketClient).send(DataEventTypeEnum.MYSELF.name()); } - + @Test public void testOnMessage() { doNothing().when(pluginDataSubscriber).onSubscribe(any()); @@ -102,7 +115,7 @@ public void testOnMessage() { shenyuWebsocketClient.onMessage(json); verify(pluginDataSubscriber).onSubscribe(any()); } - + @Test public void testOnClose() { shenyuWebsocketClient = spy(shenyuWebsocketClient); @@ -110,10 +123,212 @@ public void testOnClose() { shenyuWebsocketClient.onClose(1, "shenyu-plugin-grpc", true); verify(shenyuWebsocketClient).close(); } - + @Test public void testOnError() { shenyuWebsocketClient = spy(shenyuWebsocketClient); Assertions.assertDoesNotThrow(() -> shenyuWebsocketClient.onError(new ShenyuException("test"))); } + + // ========== reconnect/backoff tests ========== + + private ShenyuWebsocketClient createMockClient() { + ShenyuWebsocketClient client = mock(ShenyuWebsocketClient.class, + withSettings().defaultAnswer(Answers.CALLS_REAL_METHODS)); + setField(client, "reconnecting", new AtomicBoolean(false)); + setField(client, "reconnectBackoff", new AtomicInteger(0)); + setField(client, "lastReconnectAttemptTime", 0L); + return client; + } + + private void setField(final Object target, final String name, final Object value) { + try { + Field field = ShenyuWebsocketClient.class.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + private Object getField(final Object target, final String name) { + try { + Field field = ShenyuWebsocketClient.class.getDeclaredField(name); + field.setAccessible(true); + return field.get(target); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + private Object invokePrivate(final Object target, final String methodName) { + try { + Method method = ShenyuWebsocketClient.class.getDeclaredMethod(methodName); + method.setAccessible(true); + return method.invoke(target); + } catch (InvocationTargetException e) { + Throwable cause = e.getCause(); + if (cause instanceof RuntimeException) { + throw (RuntimeException) cause; + } + throw new RuntimeException(cause); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + // ---------- calculateBackoff tests ---------- + + @Test + void testCalculateBackoffReturnsZeroForNoFailures() { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnectBackoff", new AtomicInteger(0)); + + long backoff = (long) invokePrivate(client, "calculateBackoff"); + + assertEquals(0, backoff); + } + + @Test + void testCalculateBackoffExponentialGrowth() { + ShenyuWebsocketClient client = createMockClient(); + + setField(client, "reconnectBackoff", new AtomicInteger(1)); + long backoff1 = (long) invokePrivate(client, "calculateBackoff"); + assertTrue(backoff1 >= 1000 && backoff1 <= 1500, + () -> "Expected [1000, 1500] but got " + backoff1); + + setField(client, "reconnectBackoff", new AtomicInteger(2)); + long backoff2 = (long) invokePrivate(client, "calculateBackoff"); + assertTrue(backoff2 >= 2000 && backoff2 <= 3000, + () -> "Expected [2000, 3000] but got " + backoff2); + + setField(client, "reconnectBackoff", new AtomicInteger(4)); + long backoff4 = (long) invokePrivate(client, "calculateBackoff"); + assertTrue(backoff4 >= 8000 && backoff4 <= 12000, + () -> "Expected [8000, 12000] but got " + backoff4); + } + + @Test + void testCalculateBackoffMaxCap() { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnectBackoff", new AtomicInteger(10)); + + long backoff = (long) invokePrivate(client, "calculateBackoff"); + + assertTrue(backoff >= 60000 && backoff <= 90000, + () -> "Expected [60000, 90000] but got " + backoff); + } + + @Test + void testCalculateBackoffIncludesJitter() { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnectBackoff", new AtomicInteger(1)); + boolean varied = false; + long first = (long) invokePrivate(client, "calculateBackoff"); + for (int i = 0; i < 20; i++) { + if ((long) invokePrivate(client, "calculateBackoff") != first) { + varied = true; + break; + } + } + assertTrue(varied, "Backoff should vary due to jitter"); + } + + // ---------- healthCheck tests ---------- + + @Test + void testHealthCheckDoesNotDoubleSubmitWhenAlreadyReconnecting() { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnecting", new AtomicBoolean(true)); + doReturn(false).when(client).isOpen(); + + invokePrivate(client, "healthCheck"); + + assertTrue(((AtomicBoolean) getField(client, "reconnecting")).get()); + verify(client).isOpen(); + } + + @Test + void testHealthCheckResetsBackoffAndSendsPingWhenOpen() { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnectBackoff", new AtomicInteger(5)); + doReturn(true).when(client).isOpen(); + doNothing().when(client).sendPing(); + doNothing().when(client).send(anyString()); + doReturn(URI.create("ws://localhost:9090")).when(client).getURI(); + + invokePrivate(client, "healthCheck"); + + assertEquals(0, ((AtomicInteger) getField(client, "reconnectBackoff")).get()); + verify(client).sendPing(); + } + + // ---------- doReconnect tests ---------- + // reconnectBlocking() is stubbed to throw, so no real socket connection is attempted. + + @Test + void testDoReconnectIncrementsBackoffOnFailure() throws InterruptedException { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnectBackoff", new AtomicInteger(0)); + doThrow(new RuntimeException("test")).when(client).reconnectBlocking(); + doReturn(URI.create("ws://localhost:9090")).when(client).getURI(); + + invokePrivate(client, "doReconnect"); + + assertEquals(1, ((AtomicInteger) getField(client, "reconnectBackoff")).get()); + assertFalse(((AtomicBoolean) getField(client, "reconnecting")).get()); + } + + @Test + void testDoReconnectBackoffCappedAtTen() throws InterruptedException { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnectBackoff", new AtomicInteger(10)); + doThrow(new RuntimeException("test")).when(client).reconnectBlocking(); + doReturn(URI.create("ws://localhost:9090")).when(client).getURI(); + + invokePrivate(client, "doReconnect"); + + assertEquals(10, ((AtomicInteger) getField(client, "reconnectBackoff")).get()); + } + + @Test + void testDoReconnectResetsReconnectingOnFailure() throws InterruptedException { + ShenyuWebsocketClient client = createMockClient(); + doThrow(new RuntimeException("test")).when(client).reconnectBlocking(); + doReturn(URI.create("ws://localhost:9090")).when(client).getURI(); + + invokePrivate(client, "doReconnect"); + + assertFalse(((AtomicBoolean) getField(client, "reconnecting")).get()); + } + + @Test + void testDoReconnectAppliesBackoffSleep() throws InterruptedException { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnectBackoff", new AtomicInteger(1)); + setField(client, "lastReconnectAttemptTime", System.currentTimeMillis()); + doThrow(new RuntimeException("test")).when(client).reconnectBlocking(); + doReturn(URI.create("ws://localhost:9090")).when(client).getURI(); + + long start = System.currentTimeMillis(); + invokePrivate(client, "doReconnect"); + long elapsed = System.currentTimeMillis() - start; + + assertTrue(elapsed >= 800, + () -> "Expected >= 800ms backoff sleep, got " + elapsed + "ms"); + } + + @Test + void testDoReconnectPreservesInterruptStatus() { + ShenyuWebsocketClient client = createMockClient(); + setField(client, "reconnectBackoff", new AtomicInteger(1)); + setField(client, "lastReconnectAttemptTime", System.currentTimeMillis()); + + Thread.currentThread().interrupt(); + invokePrivate(client, "doReconnect"); + + assertTrue(Thread.interrupted(), "Interrupt status should be preserved after reconnect"); + assertFalse(((AtomicBoolean) getField(client, "reconnecting")).get()); + } }